291 lines
11 KiB
Python
291 lines
11 KiB
Python
"""The runner: graph plus image in, numbers out. No HTTP, no database."""
|
|
import pytest
|
|
|
|
# The vision tree requires Python >=3.13 (vs-task/vs-measure/vs-pm2d declare it
|
|
# upstream, see pyproject.toml's `vision` extra), while backend and frontend
|
|
# stay on 3.11. This must run before `import numpy` - numpy itself is only
|
|
# installed alongside the vision extra - so that `uv run pytest` on the
|
|
# default 3.11 environment reports this module as skipped, with a reason,
|
|
# instead of erroring out of the whole collection.
|
|
pytest.importorskip(
|
|
"visionsuite.task",
|
|
reason=(
|
|
"the vision tree requires Python >=3.13 (vs-task/vs-measure/vs-pm2d "
|
|
"declare it upstream); run it with "
|
|
"`uv run --python 3.13 --extra vision --extra dev pytest src/vision/tests`"
|
|
),
|
|
)
|
|
|
|
import numpy as np
|
|
|
|
from src.vision import runner
|
|
from src.vision.runner import RunOutcome, engine_version, run_graph
|
|
|
|
|
|
def a_disc(radius: int = 60, size: int = 400) -> np.ndarray:
|
|
"""A light disc on a dark ground: a fake part, but a measurable one."""
|
|
image = np.zeros((size, size), dtype=np.uint8)
|
|
yy, xx = np.ogrid[:size, :size]
|
|
centre = size // 2
|
|
image[(yy - centre) ** 2 + (xx - centre) ** 2 <= radius ** 2] = 255
|
|
return image
|
|
|
|
|
|
def a_circle_graph() -> dict:
|
|
"""A two-tool graph, built with vs-task's own API."""
|
|
from visionsuite.task import Task
|
|
from visionsuite.tools.detect import DetectCircle
|
|
from visionsuite.tools.dimensions import Diameter
|
|
|
|
task = Task(name="prova")
|
|
found = task.add(DetectCircle(name="disco"))
|
|
found.center_row, found.center_col, found.radius = 200.0, 200.0, 60.0
|
|
found.scan_size = 30.0
|
|
|
|
quota = task.add(Diameter(name="diametro"))
|
|
quota.nominal_unit = "px" # #46: v2, senza calibrazione la quota resta in px
|
|
quota.link("circle", found, "circle")
|
|
return task.to_dict()
|
|
|
|
|
|
def test_the_graph_produces_named_outputs():
|
|
outcome = run_graph(a_disc(), a_circle_graph())
|
|
|
|
assert isinstance(outcome, RunOutcome)
|
|
assert outcome.failures == []
|
|
# Keys are "tool id dot output name" - the same shape a subtask stores in
|
|
# vision_output.
|
|
assert any(key.endswith(".diameter") for key in outcome.outputs)
|
|
diameter = next(v for k, v in outcome.outputs.items() if k.endswith(".diameter"))
|
|
assert diameter == pytest.approx(120.0, abs=4.0)
|
|
|
|
|
|
def test_a_broken_tool_does_not_sink_the_others():
|
|
"""vs-task does not stop the run on an error: the runner reports it instead."""
|
|
graph = a_circle_graph()
|
|
graph["tools"].append(
|
|
{"id": "ghost", "kind": "DetectCircle", "name": "assente",
|
|
"values": {"center_row": 5.0, "center_col": 5.0, "radius": 900.0}}
|
|
)
|
|
|
|
outcome = run_graph(a_disc(), graph)
|
|
|
|
assert [f.tool_id for f in outcome.failures] == ["ghost"]
|
|
assert any(key.endswith(".diameter") for key in outcome.outputs)
|
|
|
|
|
|
class _FakeResult:
|
|
"""Stands in for a vs-task tool result: only what run_graph touches."""
|
|
|
|
def __init__(self, tool_id, tool_name, ok, outputs=None, error=None):
|
|
self.tool_id = tool_id
|
|
self.tool_name = tool_name
|
|
self.ok = ok
|
|
self.outputs = outputs or {}
|
|
self.error = error
|
|
self.overlay = [] # #46: v2, ogni risultato porta l'overlay
|
|
|
|
|
|
class _FakeReport(list):
|
|
"""#46: run_graph ora legge `.outcome` dal report di Session.run."""
|
|
|
|
outcome = "pass"
|
|
|
|
|
|
def test_a_nan_output_is_dropped_not_measured(monkeypatch):
|
|
"""vs-measure emits NaN on an empty slice (caliper.py:112 per the
|
|
revisione) - a NaN must not become a quote's value (#16)."""
|
|
def fake_run(self, image, tag=None):
|
|
return _FakeReport([_FakeResult("t1", "diametro", True, outputs={"diameter": float("nan")})])
|
|
monkeypatch.setattr(runner.Session, "run", fake_run)
|
|
|
|
outcome = run_graph(a_disc(), a_circle_graph())
|
|
|
|
assert outcome.outputs == {}
|
|
assert outcome.invalid_outputs == {"t1.diameter": "nan"}
|
|
|
|
|
|
def test_an_infinite_output_is_dropped_not_measured(monkeypatch):
|
|
def fake_run(self, image, tag=None):
|
|
return _FakeReport([_FakeResult("t1", "diametro", True, outputs={"diameter": float("inf")})])
|
|
monkeypatch.setattr(runner.Session, "run", fake_run)
|
|
|
|
outcome = run_graph(a_disc(), a_circle_graph())
|
|
|
|
assert outcome.outputs == {}
|
|
assert outcome.invalid_outputs == {"t1.diameter": "infinite"}
|
|
|
|
|
|
def test_a_bool_output_is_not_treated_as_a_measurable_number(monkeypatch):
|
|
"""bool is a subclass of int - a flag output must not become a quote."""
|
|
def fake_run(self, image, tag=None):
|
|
return _FakeReport([_FakeResult("t1", "flag", True, outputs={"present": True})])
|
|
monkeypatch.setattr(runner.Session, "run", fake_run)
|
|
|
|
outcome = run_graph(a_disc(), a_circle_graph())
|
|
|
|
assert outcome.outputs == {}
|
|
assert outcome.invalid_outputs == {}
|
|
|
|
|
|
def test_the_engine_says_which_version_it_is():
|
|
version = engine_version()
|
|
assert len(version) == 40 # a full git SHA
|
|
|
|
|
|
def test_engine_version_honours_the_environment_variable(monkeypatch):
|
|
"""A build-stamped VISION_ENGINE_VERSION wins over git, if set (Ruling R8:
|
|
the container has no usable .git for the vendored submodule, so the
|
|
worker image stamps the commit at build time instead of discovering it)."""
|
|
stamped = "a" * 40
|
|
monkeypatch.setenv("VISION_ENGINE_VERSION", stamped)
|
|
runner._engine_version_from_git.cache_clear()
|
|
|
|
try:
|
|
assert engine_version() == stamped
|
|
finally:
|
|
runner._engine_version_from_git.cache_clear()
|
|
|
|
|
|
def test_engine_version_raises_rather_than_inventing_one(monkeypatch):
|
|
"""No env var and no resolvable git HEAD: refuse loudly. An engine that
|
|
cannot say what it is has no business producing measurements - it must
|
|
never fall back to a guess, an empty string, or "unknown"."""
|
|
monkeypatch.delenv("VISION_ENGINE_VERSION", raising=False)
|
|
runner._engine_version_from_git.cache_clear()
|
|
|
|
def fake_check_output(*args, **kwargs):
|
|
raise FileNotFoundError("git: command not found")
|
|
|
|
monkeypatch.setattr(runner.subprocess, "check_output", fake_check_output)
|
|
|
|
try:
|
|
with pytest.raises(RuntimeError, match="VISION_ENGINE_VERSION"):
|
|
engine_version()
|
|
finally:
|
|
runner._engine_version_from_git.cache_clear()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# #46: ricette v2 complete - overlay, calibrazione, recipe_dir con DXF.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
import hashlib
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import cv2
|
|
|
|
_VS = Path(__file__).resolve().parents[3] / "vendor" / "visionsuite"
|
|
_SAMPLE_DXF = _VS / "packages" / "vs-dxf" / "tests" / "fixtures" / "sample.dxf"
|
|
# Il disegno di prova: rettangolo 200x100 mm con foro r=10 mm.
|
|
_PX_PER_MM = 4.0
|
|
|
|
|
|
def _dxf_scene() -> np.ndarray:
|
|
"""Il pezzo del disegno rasterizzato a 4 px/mm, margine 60 mm."""
|
|
margin = 60.0
|
|
w = int((200.0 + 2 * margin) * _PX_PER_MM)
|
|
h = int((100.0 + 2 * margin) * _PX_PER_MM)
|
|
img = np.full((h, w), 255, dtype=np.uint8)
|
|
cx, cy = w // 2, h // 2
|
|
hw, hh = 200.0 * _PX_PER_MM / 2, 100.0 * _PX_PER_MM / 2
|
|
cv2.rectangle(img, (int(cx - hw), int(cy - hh)), (int(cx + hw), int(cy + hh)), 0, -1)
|
|
cv2.circle(img, (cx, cy), int(10.0 * _PX_PER_MM), 255, -1)
|
|
return img
|
|
|
|
|
|
def _dxf_recipe() -> dict:
|
|
"""CompareDxf con ancora a due punti, collegati a due DetectPointScan.
|
|
|
|
Le due scansioni trovano i bordi sinistro e destro del pezzo una riga
|
|
sopra il lato in basso: origine del disegno = angolo in basso a sinistra,
|
|
direzione a -> b = asse x del disegno.
|
|
"""
|
|
h, w = _dxf_scene().shape
|
|
cx, cy = w // 2, h // 2
|
|
hw, hh = 200.0 * _PX_PER_MM / 2, 100.0 * _PX_PER_MM / 2
|
|
row = cy + hh - 1.0
|
|
|
|
def scan(tid: str, col: float) -> dict:
|
|
return {"id": tid, "name": tid, "kind": "DetectPointScan",
|
|
"values": {"start_row": row, "start_col": col - 30.0,
|
|
"end_row": row, "end_col": col + 30.0,
|
|
"width": 5.0, "select": "first"},
|
|
"links": {}}
|
|
|
|
return {
|
|
"schema_version": 2,
|
|
"name": "dxf",
|
|
"meta": {"dxf": {"disegno": { # sha256 obbligatorio in meta (v2)
|
|
"path": "disegno.dxf",
|
|
"sha256": hashlib.sha256(_SAMPLE_DXF.read_bytes()).hexdigest()}}},
|
|
"tools": [
|
|
scan("t-a", cx - hw),
|
|
scan("t-b", cx + hw),
|
|
{"id": "t-dxf", "name": "confronto", "kind": "CompareDxf",
|
|
"values": {"dxf": "disegno", "mm_per_px": 1.0 / _PX_PER_MM,
|
|
"anchor": "two_points",
|
|
"band_plus": 8.0, "band_minus": 8.0},
|
|
"links": {"point_a": {"tool": "t-a", "property": "point"},
|
|
"point_b": {"tool": "t-b", "property": "point"}}},
|
|
],
|
|
}
|
|
|
|
|
|
def test_a_circle_recipe_returns_overlay_and_outcome():
|
|
outcome = run_graph(a_disc(), a_circle_graph())
|
|
|
|
assert outcome.outcome in ("pass", "warning", "fail")
|
|
assert outcome.overlay, "DetectCircle must draw something"
|
|
assert all("tool_id" in item for item in outcome.overlay)
|
|
|
|
|
|
def test_calibration_dict_gives_mm_quotes():
|
|
graph = a_circle_graph()
|
|
for tool in graph["tools"]:
|
|
if tool["kind"] == "Diameter":
|
|
tool["values"]["nominal_unit"] = "mm"
|
|
|
|
outcome = run_graph(a_disc(), graph, calibration={"mm_per_px": 0.5})
|
|
|
|
assert outcome.failures == []
|
|
diameter = next(v for k, v in outcome.outputs.items() if k.endswith(".value"))
|
|
assert diameter == pytest.approx(60.0, abs=2.0) # 120 px * 0.5 mm/px
|
|
|
|
|
|
def test_compare_dxf_two_points_with_recipe_dir(tmp_path):
|
|
shutil.copy(_SAMPLE_DXF, tmp_path / "disegno.dxf")
|
|
|
|
outcome = run_graph(_dxf_scene(), _dxf_recipe(), recipe_dir=tmp_path)
|
|
|
|
assert outcome.failures == []
|
|
assert outcome.outputs["t-dxf.n_points"] > 1000
|
|
assert outcome.outputs["t-dxf.points_in_pct"] > 95.0
|
|
roles = {item.get("role") for item in outcome.overlay}
|
|
assert {"reference", "found"} <= roles
|
|
assert "t-dxf" in {item["tool_id"] for item in outcome.overlay}
|
|
|
|
|
|
def test_missing_meta_file_is_a_graph_build_error(tmp_path):
|
|
with pytest.raises(runner.GraphBuildError):
|
|
run_graph(_dxf_scene(), _dxf_recipe(), recipe_dir=tmp_path)
|
|
|
|
|
|
def test_validate_lists_numeric_outputs(tmp_path):
|
|
shutil.copy(_SAMPLE_DXF, tmp_path / "disegno.dxf")
|
|
graph = _dxf_recipe()
|
|
|
|
ok, errors, outputs = runner.validate_graph(graph, tmp_path)
|
|
|
|
assert ok, errors
|
|
keys = {o["key"] for o in outputs}
|
|
assert {"t-dxf.points_in_pct", "t-dxf.n_points"} <= keys
|
|
assert all(set(o) == {"key", "quantity", "unit"} for o in outputs)
|
|
|
|
|
|
def test_validate_reports_a_missing_file(tmp_path):
|
|
ok, errors, outputs = runner.validate_graph(_dxf_recipe(), tmp_path)
|
|
|
|
assert not ok and errors and outputs == []
|