operatore: acquisizione immagine, proxy execute e overlay per camera_measure/dxf_compare (#46 )

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-07 11:55:42 +02:00
16 changed files with 371 additions and 701 deletions
@@ -5,25 +5,19 @@ the numbers: that is what lets a station's configuration change without touching
 the frontend or the recipes.
 """
 import io
 import json
 from fastapi.responses import FileResponse
 from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
 from PIL import Image
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 from src.backend.api.middleware.api_key import require_maker, require_measurement_tec
-from src.backend.api.routers.files import (
+from src.backend.api.routers.files import ALLOWED_IMAGE_TYPES, validate_file_size
    ALLOWED_IMAGE_TYPES, resolve_upload_path, validate_file_size,
 )
 from src.backend.config import settings
 from src.backend.database import get_db
 from src.backend.models.api.vision import (
-    VisionExecuteResponse, VisionMeasurementResult, VisionRecipeResponse,
+    VisionExecuteResponse, VisionMeasurementResult, VisionReferenceImageResponse,
    VisionRecipeUploadResponse, VisionReferenceImageResponse,
 )
 from src.backend.models.orm.vision import VisionResult
 from src.backend.models.orm.task import RecipeTask
 from src.backend.models.orm.user import User
 from src.backend.services import vision_service
@@ -132,7 +126,7 @@ async def execute(
 ) -> VisionExecuteResponse:
    task = await _task_or_404(db, task_id)
-    result, saved, report = await vision_service.execute_task(
+    result, saved = await vision_service.execute_task(
        db,
        task,
        await _read_image(image),
@@ -141,7 +135,6 @@ async def execute(
        lot_number=lot_number,
        serial_number=serial_number,
        production_run_id=production_run_id,
        image_ext=_IMAGE_EXTENSIONS.get(image.content_type, "png"),
    )
    await db.commit()
@@ -154,111 +147,6 @@ async def execute(
            )
            for m in saved
        ],
        outcome=report.get("outcome"),
        overlay=report.get("overlay") or [],
        image_url=f"/api/vision/results/{result.id}/image",
        failures=report.get("failures") or [],
    )
@router.get("/results/{result_id}/image")
 async def get_result_image(
    result_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(require_measurement_tec),
 ) -> FileResponse:
    """The image a vision run measured, as saved by `execute` (#46)."""
    result = (await db.execute(
        select(VisionResult).where(VisionResult.id == result_id)
    )).scalar_one_or_none()
    if result is None or not result.image_path:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Result image not found",
        )
    return FileResponse(resolve_upload_path(result.image_path))
 def _422(errors: list[str]) -> HTTPException:
    # detail e' una lista di stringhe: il frontend legge i 422 solo da `detail`.
    return HTTPException(
        status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=errors,
    )
@router.post("/tasks/{task_id}/recipe", response_model=VisionRecipeUploadResponse)
 async def upload_recipe(
    task_id: int,
    recipe: UploadFile = File(...),
    files: list[UploadFile] = File(default=[]),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(require_maker),
 ) -> VisionRecipeUploadResponse:
    """Load a VisionSuite v2 recipe plus its meta files onto a task (#46)."""
    task = await _task_or_404(db, task_id)
    if task.task_type not in vision_service.VISION_TASK_TYPES:
        raise _422([f"task type {task.task_type!r} does not run a vision recipe"])
    raw = await recipe.read()
    if not validate_file_size(len(raw)):
        raise _422([f"recipe exceeds {settings.max_upload_size_mb}MB"])
    try:
        graph = json.loads(raw)
    except ValueError:
        raise _422(["the recipe is not valid JSON"])
    if not isinstance(graph, dict):
        raise _422(["the recipe must be a JSON object"])
    uploads: list[tuple[str, bytes]] = []
    for f in files:
        content = await f.read()
        if not validate_file_size(len(content)):
            raise _422([f"file {f.filename!r} exceeds {settings.max_upload_size_mb}MB"])
        uploads.append((f.filename or "", content))
    placed, errors = vision_service.match_recipe_files(graph, uploads)
    if errors:
        raise _422(errors)
    # The folder is content-addressed and immutable; writing it before the
    # validation is harmless (an orphan folder is never referenced) and the
    # worker needs it on disk to validate meta files.
    try:
        recipe_dir = vision_service.store_recipe_folder(task.id, graph, placed)
    except ValueError as exc:
        raise _422([str(exc)])
    check = await vision_service.validate_recipe(graph, recipe_dir)
    if not check["ok"]:
        raise _422([str(e) for e in check["errors"]] or ["the recipe is not valid"])
    task.vision_json = graph
    task.vision_recipe_dir = recipe_dir
    await db.commit()
    return VisionRecipeUploadResponse(
        task_id=task.id,
        recipe_dir=recipe_dir,
        files=vision_service.list_recipe_files(recipe_dir),
        outputs=check["outputs"],
    )
@router.get("/tasks/{task_id}/recipe", response_model=VisionRecipeResponse)
 async def get_recipe(
    task_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(require_maker),
 ) -> VisionRecipeResponse:
    task = await _task_or_404(db, task_id)
    outputs: list[dict] = []
    if task.vision_json:
        check = await vision_service.validate_recipe(
            task.vision_json, task.vision_recipe_dir,
        )
        outputs = check.get("outputs", [])
    return VisionRecipeResponse(
        task_id=task.id,
        recipe_dir=task.vision_recipe_dir,
        recipe=task.vision_json,
        files=vision_service.list_recipe_files(task.vision_recipe_dir),
        outputs=outputs,
    )
@@ -1,30 +0,0 @@
 """recipe_tasks.vision_recipe_dir: cartella della ricetta VisionSuite v2 (#46)
 Percorso relativo sotto upload_path della cartella immutabile che contiene
 recipe.json e i file di meta (DXF, modelli .npz, ...).
 Revision ID: 016_vision_recipe_dir
 Revises: 015_api_key_hash
 Create Date: 2026-10-07
 """
 from typing import Sequence, Union
 from alembic import op
 import sqlalchemy as sa
 revision: str = '016_vision_recipe_dir'
 down_revision: Union[str, None] = '015_api_key_hash'
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 def upgrade() -> None:
    # batch: SQLite non sa fare DROP COLUMN nativo su tutte le versioni
    with op.batch_alter_table('recipe_tasks') as batch:
        batch.add_column(sa.Column('vision_recipe_dir', sa.String(500), nullable=True))
 def downgrade() -> None:
    with op.batch_alter_table('recipe_tasks') as batch:
        batch.drop_column('vision_recipe_dir')
@@ -16,34 +16,6 @@ class VisionExecuteResponse(BaseModel):
    vision_result_id: int
    engine_version: str
    measurements: list[VisionMeasurementResult]
    # Esito VisionSuite (informativo: il verdetto lo danno utl/ltl), overlay
    # in px immagine, URL dell'immagine salvata e strumenti falliti (#46).
    outcome: str | None = None
    overlay: list[dict] = []
    image_url: str | None = None
    failures: list[dict] = []
 class VisionRecipeOutput(BaseModel):
    """One numeric output a recipe offers to the quotes."""
    key: str
    quantity: str | None = None
    unit: str | None = None
 class VisionRecipeUploadResponse(BaseModel):
    task_id: int
    recipe_dir: str
    files: list[str]
    outputs: list[VisionRecipeOutput]
 class VisionRecipeResponse(BaseModel):
    task_id: int
    recipe_dir: str | None = None
    recipe: dict | None = None
    files: list[str]
    outputs: list[VisionRecipeOutput]
 class VisionReferenceImageResponse(BaseModel):
@@ -18,8 +18,7 @@ if TYPE_CHECKING:
 TASK_TYPES = ("note", "measure", "drawing", "dxf_compare", "camera_measure")
 # Types the operator screen treats as "there are quotes to take here".
-# dxf_compare produce quote come camera_measure (#46)
+MEASURING_TASK_TYPES = ("measure", "camera_measure")
 MEASURING_TASK_TYPES = ("measure", "camera_measure", "dxf_compare")
 class RecipeTask(Base):
@@ -52,9 +51,6 @@ class RecipeTask(Base):
    # server routes it and never interprets it, which is what keeps VisionSuite
    # out of this container.
    vision_json: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True)
    # Cartella immutabile (relativa a upload_path) con recipe.json e i file di
    # meta della ricetta VisionSuite v2 (#46).
    vision_recipe_dir: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
    # Relationships
    version: Mapped["RecipeVersion"] = relationship(back_populates="tasks")
@@ -115,8 +115,6 @@ async def _copy_tasks_to_version(
            # a camera_measure task unable to execute from the moment a
            # recipe with measurements is next edited.
            vision_json=task.vision_json,
            # Cartella immutabile: basta il riferimento (#46).
            vision_recipe_dir=task.vision_recipe_dir,
        )
        db.add(new_task)
        await db.flush()  # get new_task.id
@@ -5,7 +5,6 @@ to the worker over HTTP and stays ignorant of how a number was produced.
 """
 from __future__ import annotations
 import hashlib
 import json
 import math
 from pathlib import Path
@@ -47,13 +46,7 @@ def _is_measurable(value: object) -> bool:
    )
-async def _call_worker(
+async def _call_worker(image_bytes: bytes, graph: dict) -> dict:
    image_bytes: bytes,
    graph: dict,
    *,
    recipe_dir: str | None = None,
    calibration: dict | None = None,
 ) -> dict:
    """The one place that talks to the worker; the tests replace it.
    The worker's fault must not become the API's fault (spec's edge-case
@@ -67,19 +60,12 @@ async def _call_worker(
    is not the worker's JSON contract must not turn into a bare KeyError
    further down.
    """
    # recipe_dir (relativo, #46) e calibration sono opzionali: se assenti non
    # si mandano, il worker usa i suoi default.
    data = {"graph": json.dumps(graph)}
    if recipe_dir:
        data["recipe_dir"] = recipe_dir
    if calibration:
        data["calibration"] = json.dumps(calibration)
    try:
        async with httpx.AsyncClient(timeout=120.0) as client:
            response = await client.post(
                f"{settings.vision_worker_url}/run",
                files={"image": ("frame.png", image_bytes, "image/png")},
-                data=data,
+                data={"graph": json.dumps(graph)},
            )
    except httpx.TimeoutException as exc:
        raise HTTPException(
@@ -136,188 +122,6 @@ async def _call_worker(
    return report
 async def validate_recipe(graph: dict, recipe_dir: str | None) -> dict:
    """Ask the worker's /validate whether a v2 recipe is loadable (#46).
    Returns {"ok", "errors", "outputs"}. A worker 422 (e.g. recipe_dir
    refused) is folded into ok=False; transport faults become 502/504 as in
    `_call_worker`. Replaced by the tests.
    """
    data = {"graph": json.dumps(graph)}
    if recipe_dir:
        data["recipe_dir"] = recipe_dir
    try:
        async with httpx.AsyncClient(timeout=60.0) as client:
            response = await client.post(
                f"{settings.vision_worker_url}/validate", data=data,
            )
    except httpx.TimeoutException as exc:
        raise HTTPException(
            status_code=status.HTTP_504_GATEWAY_TIMEOUT,
            detail=f"the vision worker did not answer in time: {exc}",
        ) from exc
    except httpx.HTTPError as exc:
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail=f"the vision worker is unreachable: {exc}",
        ) from exc
    if response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT:
        try:
            detail = response.json().get("detail", response.text)
        except ValueError:
            detail = response.text
        errors = detail if isinstance(detail, list) else [str(detail)]
        return {"ok": False, "errors": [str(e) for e in errors], "outputs": []}
    if response.status_code != 200:
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail=f"the vision worker refused the job: {response.text}",
        )
    try:
        body = response.json()
    except ValueError as exc:
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail="the vision worker's answer was malformed: not valid JSON",
        ) from exc
    if not isinstance(body, dict) or "ok" not in body:
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail="the vision worker's answer was malformed: missing ok",
        )
    body.setdefault("errors", [])
    body.setdefault("outputs", [])
    return body
 # Task types that run a vision graph (#46).
 VISION_TASK_TYPES = ("camera_measure", "dxf_compare")
 def _safe_join(root: Path, relative: str) -> Path:
    """root/relative, refusing absolute paths, '..' and anything leaving root."""
    candidate = Path(relative)
    if candidate.is_absolute() or ".." in candidate.parts or not candidate.parts:
        raise ValueError(f"unsafe path: {relative}")
    full = (root / candidate).resolve()
    if not full.is_relative_to(root.resolve()):
        raise ValueError(f"unsafe path: {relative}")
    return full
 def collect_meta_paths(node: object) -> list[str]:
    """Every `path` string found in any dict under the recipe's `meta`."""
    found: list[str] = []
    if isinstance(node, dict):
        for key, value in node.items():
            if key == "path" and isinstance(value, str):
                found.append(value)
            else:
                found.extend(collect_meta_paths(value))
    elif isinstance(node, list):
        for item in node:
            found.extend(collect_meta_paths(item))
    return found
 def match_recipe_files(
    recipe: dict, uploads: list[tuple[str, bytes]],
 ) -> tuple[dict[str, bytes], list[str]]:
    """Pair each uploaded file with a meta path by base name (#46).
    Returns ({relative path: bytes}, errors). Errors list every upload that
    has no name or matches no meta path, and every unsafe meta path. A meta
    path left without a file is not an error here: /validate decides.
    """
    paths = collect_meta_paths(recipe.get("meta", {}))
    by_name: dict[str, str] = {}
    for p in paths:
        by_name.setdefault(Path(p.replace("\\", "/")).name, p)
    placed: dict[str, bytes] = {}
    errors: list[str] = []
    for name, content in uploads:
        if not name:
            errors.append("a file has no name")
            continue
        target = by_name.get(Path(name.replace("\\", "/")).name)
        if target is None:
            errors.append(f"file {name!r} matches no path in the recipe's meta")
            continue
        try:
            _safe_join(Path("."), target)
        except ValueError:
            errors.append(f"meta path {target!r} is not allowed")
            continue
        placed[target] = content
    return placed, errors
 def store_recipe_folder(
    task_id: int, recipe: dict, files: dict[str, bytes],
 ) -> str:
    """Write the immutable recipe folder; return its path relative to uploads.
    Folder name = first 16 hex of the recipe's sha256. If it already exists it
    is left untouched (immutable: other task versions may reference it).
    """
    canonical = json.dumps(recipe, sort_keys=True, separators=(",", ":"))
    # Il digest copre anche i file (path ordinato + sha256 del contenuto): la
    # stessa ricetta con file diversi (es. primo upload senza il DXF) deve dare
    # una cartella diversa, altrimenti quella incompleta resterebbe per sempre
    # perche' immutabile (#46).
    h = hashlib.sha256(canonical.encode("utf-8"))
    for rel in sorted(files):
        h.update(b"\0" + rel.encode("utf-8") + b"\0")
        h.update(hashlib.sha256(files[rel]).digest())
    digest = h.hexdigest()[:16]
    relative = f"vision/recipes/{task_id}/{digest}"
    folder = settings.upload_path / relative
    if folder.exists():
        return relative
    folder.mkdir(parents=True, exist_ok=True)
    (folder / "recipe.json").write_text(
        json.dumps(recipe, indent=2, ensure_ascii=False), encoding="utf-8",
    )
    for rel, content in files.items():
        target = _safe_join(folder, rel)
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(content)
    return relative
 def list_recipe_files(recipe_dir: str | None) -> list[str]:
    """Files of a recipe folder, relative to it (recipe.json excluded)."""
    if not recipe_dir:
        return []
    try:
        folder = _safe_join(settings.upload_path, recipe_dir)
    except ValueError:
        return []
    if not folder.is_dir():
        return []
    return sorted(
        p.relative_to(folder).as_posix()
        for p in folder.rglob("*") if p.is_file() and p.name != "recipe.json"
    )
 def save_result_files(
    task_id: int, image_bytes: bytes, overlay: list, image_ext: str = "png",
 ) -> tuple[str, str]:
    """Save the executed image and its overlay JSON side by side (#46).
    Returns (image_path, overlay_path), both relative to uploads.
    """
    stem = uuid4().hex
    base = Path("vision") / "results" / str(task_id)
    (settings.upload_path / base).mkdir(parents=True, exist_ok=True)
    image_rel = (base / f"{stem}.{image_ext}").as_posix()
    overlay_rel = (base / f"{stem}.overlay.json").as_posix()
    (settings.upload_path / image_rel).write_bytes(image_bytes)
    (settings.upload_path / overlay_rel).write_text(json.dumps(overlay), encoding="utf-8")
    return image_rel, overlay_rel
 async def execute_task(
    db: AsyncSession,
    task: RecipeTask,
@@ -328,9 +132,8 @@ async def execute_task(
    lot_number: str | None = None,
    serial_number: str | None = None,
    production_run_id: int | None = None,
-    image_ext: str = "png",
+) -> tuple[VisionResult, list[Measurement]]:
-) -> tuple[VisionResult, list[Measurement], dict]:
+    if task.task_type != "camera_measure":
    if task.task_type not in VISION_TASK_TYPES:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=f"task type {task.task_type!r} is not a camera measurement",
@@ -407,11 +210,7 @@ async def execute_task(
            ),
        )
-    # Nessuna fonte di calibrazione esiste oggi (ne' task ne' stazione): non si
+    report = await _call_worker(image_bytes, task.vision_json)
    # passa `calibration`, la ricetta porta la sua (#46).
    report = await _call_worker(
        image_bytes, task.vision_json, recipe_dir=task.vision_recipe_dir,
    )
    outputs = report["outputs"]
    # Every quote or none. A half-measured piece read as a whole one is worse
@@ -432,12 +231,8 @@ async def execute_task(
            ),
        )
    overlay = report.get("overlay") or []
    image_path, overlay_path = save_result_files(task.id, image_bytes, overlay, image_ext)
    result = VisionResult(
        task_id=task.id,
        image_path=image_path,
        overlay_path=overlay_path,
        engine_version=report["engine_version"],
        executed_on="server",
        graph_snapshot=task.vision_json,
@@ -462,7 +257,7 @@ async def execute_task(
            enforce_tolerance_gate=False,
        ))
-    return result, saved, report
+    return result, saved
 def _reference_file_path(reference: VisionReferenceImage) -> Path:
@@ -582,6 +377,4 @@ async def preview(db: AsyncSession, task: RecipeTask, image_bytes: bytes) -> dic
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail="this task carries no vision graph",
        )
-    return await _call_worker(
+    return await _call_worker(image_bytes, task.vision_json)
        image_bytes, task.vision_json, recipe_dir=task.vision_recipe_dir,
    )
@@ -131,21 +131,3 @@ class TestApiKeyHashMigration:
        command.upgrade(cfg, "head")  # and up again
        assert "api_keys" in _inspect(db_path).get_table_names()
 class TestVisionRecipeDirMigration:
    """016_vision_recipe_dir: recipe_tasks.vision_recipe_dir, su e giu (#46)."""
    def test_016_upgrade_and_downgrade(self, sqlite_database_url):
        db_path, _ = sqlite_database_url
        cfg = _alembic_config()
        def cols():
            return {c["name"] for c in _inspect(db_path).get_columns("recipe_tasks")}
        command.upgrade(cfg, "head")
        assert "vision_recipe_dir" in cols()
        command.downgrade(cfg, "015_api_key_hash")
        assert "vision_recipe_dir" not in cols()
        command.upgrade(cfg, "head")  # e di nuovo su
        assert "vision_recipe_dir" in cols()
@@ -66,7 +66,7 @@ async def two_vision_quotes(db_session, recipe_id: int):
 def worker_says(monkeypatch):
    """The worker does not run in tests: its response is faked here."""
    def _install(outputs: dict, failures: list | None = None):
-        async def fake_call(image_bytes: bytes, graph: dict, **kwargs) -> dict:
+        async def fake_call(image_bytes: bytes, graph: dict) -> dict:
            return {
                "outputs": outputs,
                "failures": failures or [],
@@ -316,7 +316,7 @@ async def test_a_pending_fail_blocks_before_the_worker_is_called(
    )
    await db_session.commit()
-    async def worker_must_not_be_called(image_bytes: bytes, graph: dict, **kwargs) -> dict:
+    async def worker_must_not_be_called(image_bytes: bytes, graph: dict) -> dict:
        raise AssertionError("the worker must not run while the operator is blocked")
    monkeypatch.setattr(
        "src.backend.services.vision_service._call_worker", worker_must_not_be_called,
@@ -53,7 +53,7 @@ async def a_camera_task(db_session, recipe_id: int):
@pytest.fixture
 def worker_says(monkeypatch):
    def _install(outputs: dict):
-        async def fake_call(image_bytes: bytes, graph: dict, **kwargs) -> dict:
+        async def fake_call(image_bytes: bytes, graph: dict) -> dict:
            return {"outputs": outputs, "failures": [],
                    "engine_version": "e" * 40, "duration_ms": 3}
        monkeypatch.setattr(
@@ -1,282 +0,0 @@
 """Ricetta VisionSuite v2 su un task, execute dxf_compare, copy-on-write (#46)."""
 import io
 import json
 import pytest
 from PIL import Image
 from sqlalchemy import select
 from src.backend.config import settings
 from src.backend.models.orm.recipe import RecipeVersion
 from src.backend.models.orm.task import RecipeSubtask, RecipeTask
 from src.backend.models.orm.vision import VisionResult
 from src.backend.tests.conftest import auth_headers, create_test_recipe
 RECIPE = {
    "schema_version": 2, "name": "dxf", "tools": [],
    "meta": {"dxf": {"d1": {"path": "dxf/part.dxf"}}, "models": {"m": {"path": "models/m.npz"}}},
 }
 OUTPUTS = [{"key": "cmp.points_in_pct", "quantity": "ratio", "unit": "%"}]
 GOOD_FILES = (("part.dxf", b"0 EOF"), ("m.npz", b"npz"))
@pytest.fixture(autouse=True)
 def redirect_uploads(monkeypatch, tmp_path):
    # upload_path = parents[2] / upload_dir: un assoluto sostituisce la base
    monkeypatch.setattr(settings, "upload_dir", str(tmp_path))
@pytest.fixture
 def validate_says(monkeypatch):
    """Il worker /validate non gira nei test: risposta finta."""
    def _install(ok=True, errors=None, outputs=None):
        async def fake(graph, recipe_dir):
            return {
                "ok": ok, "errors": errors or [],
                "outputs": outputs if outputs is not None else OUTPUTS,
            }
        monkeypatch.setattr("src.backend.services.vision_service.validate_recipe", fake)
    return _install
 def an_image() -> bytes:
    buf = io.BytesIO()
    Image.new("L", (32, 32)).save(buf, format="PNG")
    return buf.getvalue()
 async def make_task(db_session, user, task_type="dxf_compare"):
    recipe = await create_test_recipe(db_session, user.id)
    version = (await db_session.execute(
        select(RecipeVersion).where(RecipeVersion.recipe_id == recipe.id)
    )).scalars().first()
    task = (await db_session.execute(
        select(RecipeTask).where(RecipeTask.version_id == version.id)
    )).scalars().first()
    subtask = (await db_session.execute(
        select(RecipeSubtask).where(RecipeSubtask.task_id == task.id)
    )).scalars().first()
    task.task_type = task_type
    await db_session.commit()
    return task, subtask
 def upload(client, user, task, recipe=RECIPE, files=()):
    return client.post(
        f"/api/vision/tasks/{task.id}/recipe",
        headers=auth_headers(user),
        files=[("recipe", ("r.json", json.dumps(recipe).encode(), "application/json"))]
        + [("files", (n, c, "application/octet-stream")) for n, c in files],
    )
@pytest.mark.asyncio
 async def test_upload_ok_stores_folder_and_task(client, db_session, maker_user, validate_says):
    validate_says()
    task, _ = await make_task(db_session, maker_user)
    resp = await upload(client, maker_user, task, files=GOOD_FILES)
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["outputs"] == OUTPUTS
    assert sorted(body["files"]) == ["dxf/part.dxf", "models/m.npz"]
    assert body["recipe_dir"].startswith(f"vision/recipes/{task.id}/")
    folder = settings.upload_path / body["recipe_dir"]
    assert (folder / "recipe.json").is_file() and (folder / "dxf" / "part.dxf").is_file()
    await db_session.refresh(task)
    assert task.vision_json == RECIPE and task.vision_recipe_dir == body["recipe_dir"]
    got = await client.get(f"/api/vision/tasks/{task.id}/recipe", headers=auth_headers(maker_user))
    assert got.status_code == 200
    assert got.json()["recipe"] == RECIPE and got.json()["outputs"] == OUTPUTS
@pytest.mark.asyncio
 async def test_get_without_recipe_is_empty(client, db_session, maker_user, validate_says):
    validate_says()
    task, _ = await make_task(db_session, maker_user)
    got = await client.get(f"/api/vision/tasks/{task.id}/recipe", headers=auth_headers(maker_user))
    assert got.json() == {
        "task_id": task.id, "recipe_dir": None, "recipe": None, "files": [], "outputs": [],
    }
@pytest.mark.asyncio
 async def test_unmatched_file_name_is_422_list(client, db_session, maker_user, validate_says):
    validate_says()
    task, _ = await make_task(db_session, maker_user)
    resp = await upload(client, maker_user, task, files=[("other.dxf", b"x")])
    assert resp.status_code == 422
    assert isinstance(resp.json()["detail"], list) and "other.dxf" in resp.json()["detail"][0]
    await db_session.refresh(task)
    assert task.vision_json is None and task.vision_recipe_dir is None
@pytest.mark.asyncio
 async def test_traversal_in_meta_path_is_422(client, db_session, maker_user, validate_says):
    validate_says()
    task, _ = await make_task(db_session, maker_user)
    evil = {"schema_version": 2, "tools": [], "meta": {"dxf": {"d": {"path": "../../evil.dxf"}}}}
    resp = await upload(client, maker_user, task, recipe=evil, files=[("evil.dxf", b"x")])
    assert resp.status_code == 422
    assert not (settings.upload_path.parent / "evil.dxf").exists()
@pytest.mark.asyncio
 async def test_wrong_task_type_is_422(client, db_session, maker_user, validate_says):
    validate_says()
    task, _ = await make_task(db_session, maker_user, task_type="measure")
    resp = await upload(client, maker_user, task, files=GOOD_FILES)
    assert resp.status_code == 422
@pytest.mark.asyncio
 async def test_invalid_json_is_422(client, db_session, maker_user, validate_says):
    validate_says()
    task, _ = await make_task(db_session, maker_user)
    resp = await client.post(
        f"/api/vision/tasks/{task.id}/recipe", headers=auth_headers(maker_user),
        files=[("recipe", ("r.json", b"{nope", "application/json"))],
    )
    assert resp.status_code == 422
@pytest.mark.asyncio
 async def test_worker_not_ok_is_422_and_task_unchanged(client, db_session, maker_user, validate_says):
    validate_says(ok=False, errors=["meta file missing"])
    task, _ = await make_task(db_session, maker_user)
    task.vision_json = {"old": 1}
    task.vision_recipe_dir = "vision/recipes/old"
    await db_session.commit()
    resp = await upload(client, maker_user, task, files=GOOD_FILES)
    assert resp.status_code == 422
    assert resp.json()["detail"] == ["meta file missing"]
    await db_session.refresh(task)
    assert task.vision_json == {"old": 1} and task.vision_recipe_dir == "vision/recipes/old"
@pytest.mark.asyncio
 async def test_existing_folder_is_not_rewritten(client, db_session, maker_user, validate_says):
    validate_says()
    task, _ = await make_task(db_session, maker_user)
    first = (await upload(client, maker_user, task, files=GOOD_FILES)).json()
    marker = settings.upload_path / first["recipe_dir"] / "dxf" / "part.dxf"
    marker.write_bytes(b"CHANGED")
    second = (await upload(client, maker_user, task, files=GOOD_FILES)).json()
    assert second["recipe_dir"] == first["recipe_dir"]
    assert marker.read_bytes() == b"CHANGED"
@pytest.mark.asyncio
 async def test_upload_requires_maker(
    client, db_session, maker_user, measurement_tec_user, validate_says,
 ):
    validate_says()
    task, _ = await make_task(db_session, maker_user)
    resp = await upload(client, measurement_tec_user, task, files=GOOD_FILES)
    assert resp.status_code in (401, 403)
@pytest.mark.asyncio
 async def test_execute_dxf_compare_saves_files_and_extended_response(
    client, db_session, measurement_tec_user, monkeypatch,
 ):
    task, subtask = await make_task(db_session, measurement_tec_user)
    task.vision_json = RECIPE
    task.vision_recipe_dir = "vision/recipes/1/abc"
    subtask.vision_output = "cmp.points_in_pct"
    await db_session.commit()
    seen = {}
    overlay = [{"type": "contour", "role": "found", "rows": [1], "cols": [2], "tool_id": "cmp"}]
    async def fake_call(image_bytes, graph, **kwargs):
        seen.update(kwargs)
        return {
            "outputs": {"cmp.points_in_pct": float(subtask.nominal)},
            "failures": [], "engine_version": "e" * 40, "duration_ms": 5,
            "overlay": overlay, "outcome": "OK",
        }
    monkeypatch.setattr("src.backend.services.vision_service._call_worker", fake_call)
    resp = await client.post(
        "/api/vision/execute", headers=auth_headers(measurement_tec_user),
        files={"image": ("p.png", an_image(), "image/png")},
        data={"task_id": str(task.id)},
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert seen["recipe_dir"] == "vision/recipes/1/abc" and "calibration" not in seen
    assert body["outcome"] == "OK" and body["overlay"] == overlay and body["failures"] == []
    assert body["image_url"] == f"/api/vision/results/{body['vision_result_id']}/image"
    assert body["measurements"][0]["pass_fail"] == "pass"
    result = (await db_session.execute(select(VisionResult))).scalars().one()
    assert (settings.upload_path / result.image_path).read_bytes() == an_image()
    assert json.loads((settings.upload_path / result.overlay_path).read_text()) == overlay
    img = await client.get(body["image_url"], headers=auth_headers(measurement_tec_user))
    assert img.status_code == 200 and img.content == an_image()
@pytest.mark.asyncio
 async def test_execute_dxf_compare_out_of_tolerance_fails(
    client, db_session, measurement_tec_user, monkeypatch,
 ):
    task, subtask = await make_task(db_session, measurement_tec_user)
    task.vision_json = RECIPE
    subtask.vision_output = "cmp.points_in_pct"
    await db_session.commit()
    async def fake_call(image_bytes, graph, **kwargs):
        return {"outputs": {"cmp.points_in_pct": 99.0}, "failures": [],
                "engine_version": "e" * 40, "outcome": "OK"}
    monkeypatch.setattr("src.backend.services.vision_service._call_worker", fake_call)
    resp = await client.post(
        "/api/vision/execute", headers=auth_headers(measurement_tec_user),
        files={"image": ("p.png", an_image(), "image/png")},
        data={"task_id": str(task.id)},
    )
    assert resp.status_code == 200
    # il verdetto e' delle tolleranze del subtask, non di outcome di VisionSuite
    assert resp.json()["measurements"][0]["pass_fail"] == "fail"
    assert resp.json()["overlay"] == []
@pytest.mark.asyncio
 async def test_copy_on_write_carries_recipe_dir(client, db_session, maker_user):
    from src.backend.tests.test_vision_copy_on_write import (
        _current_task_and_subtask, _seed_measurement,
    )
    recipe = await create_test_recipe(db_session, maker_user.id)
    _, task, _ = await _current_task_and_subtask(db_session, recipe.id)
    task.task_type = "dxf_compare"
    task.vision_json = RECIPE
    task.vision_recipe_dir = "vision/recipes/1/abc"
    await db_session.flush()
    await _seed_measurement(db_session, recipe.id, maker_user.id)
    db_session.expunge_all()
    resp = await client.put(
        f"/api/recipes/{recipe.id}", headers=auth_headers(maker_user),
        json={"change_notes": "cow"},
    )
    assert resp.status_code == 200
    db_session.expunge_all()
    version, new_task, _ = await _current_task_and_subtask(db_session, recipe.id)
    assert version.version_number == 2
    assert new_task.vision_recipe_dir == "vision/recipes/1/abc" and new_task.vision_json == RECIPE
@pytest.mark.asyncio
 async def test_retry_with_missing_file_then_complete_succeeds(
    client, db_session, maker_user, monkeypatch,
 ):
    # worker finto: ok solo se il DXF esiste nella cartella della ricetta
    async def fake(graph, recipe_dir):
        present = (settings.upload_path / recipe_dir / "dxf" / "part.dxf").is_file()
        return {"ok": present, "errors": [] if present else ["dxf missing"], "outputs": OUTPUTS}
    monkeypatch.setattr("src.backend.services.vision_service.validate_recipe", fake)
    task, _ = await make_task(db_session, maker_user)
    first = await upload(client, maker_user, task, files=[])
    assert first.status_code == 422
    second = await upload(client, maker_user, task, files=GOOD_FILES)
    assert second.status_code == 200, second.text
@@ -39,7 +39,7 @@ async def recipe_parts(db_session, recipe_id: int):
@pytest.fixture
 def worker_says(monkeypatch):
    def _install(outputs: dict):
-        async def fake_call(image_bytes: bytes, graph: dict, **kwargs) -> dict:
+        async def fake_call(image_bytes: bytes, graph: dict) -> dict:
            return {"outputs": outputs, "failures": [],
                    "engine_version": "b" * 40, "duration_ms": 7}
        monkeypatch.setattr(
@@ -69,7 +69,7 @@ async def recipe_parts(db_session, recipe_id: int):
@pytest.fixture
 def worker_says(monkeypatch):
    def _install(outputs: dict):
-        async def fake_call(image_bytes: bytes, graph: dict, **kwargs) -> dict:
+        async def fake_call(image_bytes: bytes, graph: dict) -> dict:
            return {"outputs": outputs, "failures": [],
                    "engine_version": "c" * 40, "duration_ms": 5}
        monkeypatch.setattr(
@@ -1,8 +1,10 @@
 """MeasurementTec blueprint - recipe selection and measurement execution."""
 import urllib.parse
 import requests as http_requests
 from flask import (
-    Blueprint, abort, flash, jsonify, redirect, render_template,
+    Blueprint, Response, abort, flash, jsonify, redirect, render_template,
    request, session, url_for,
 )
 from flask_babel import gettext as _
@@ -820,3 +822,67 @@ def api_close_production(run_id: int):
 def api_get_file(file_path: str):
    """Proxy: Serve file from API server (browser can't send X-API-Key)."""
    return proxy_file(file_path)
 # ---------------------------------------------------------------------------
 # Route: vision execute (#46) - the browser sends an image, we relay it to
 # POST /api/vision/execute (multipart) because it cannot send X-API-Key itself.
 # ---------------------------------------------------------------------------
@measure_bp.route("/api/vision/execute", methods=["POST"])
@login_required
@role_required("MeasurementTec")
 def api_vision_execute():
    """Proxy: run the task's VisionSuite recipe on the uploaded image."""
    image = request.files.get("image")
    task_id = request.form.get("task_id")
    if image is None or not task_id:
        return jsonify({
            "error": True,
            "detail": _("Dati mancanti: task_id e image sono obbligatori"),
        }), 400
    # Same context fields as the manual flow; empty ones are left out so the
    # backend sees them as absent rather than as "".
    form = {"task_id": task_id}
    for key in ("task_type", "lot_number", "serial_number", "production_run_id"):
        val = request.form.get(key)
        if val:
            form[key] = val
    form.setdefault("lot_number", session.get("lot_number", ""))
    form.setdefault("serial_number", session.get("serial_number", ""))
    resp = api_client.post(
        "/api/vision/execute",
        data=form,
        files={"image": (
            image.filename or "capture.jpg",
            image.read(),
            image.mimetype or "application/octet-stream",
        )},
    )
    if resp.get("error"):
        status_code = resp.get("status_code", 500)
        body = {"error": True, "detail": resp.get("detail", _("Errore nell'esecuzione"))}
        if resp.get("session_expired"):
            body["session_expired"] = True
        return jsonify(body), status_code if status_code >= 400 else 500
    return jsonify(resp), 200
@measure_bp.route("/api/vision/results/<int:result_id>/image", methods=["GET"])
@login_required
@role_required("MeasurementTec")
 def api_vision_result_image(result_id: int):
    """Proxy: the image saved with a vision result (needs X-API-Key)."""
    base_url = Config.API_SERVER_URL.rstrip("/")
    resp = http_requests.get(
        f"{base_url}/api/vision/results/{result_id}/image",
        headers={"X-API-Key": session.get("api_key", "")},
        timeout=30,
    )
    if resp.status_code != 200:
        return Response(resp.text, status=resp.status_code)
    return Response(
        resp.content,
        content_type=resp.headers.get("content-type", "application/octet-stream"),
    )
@@ -0,0 +1,125 @@
 /**
 * Vision execute - Alpine component for camera_measure / dxf_compare tasks (#46).
 *
 * The operator takes (or picks) an image, `acquire(blob)` sends it to the Flask
 * proxy, and the result - quote table, outcome, failures, overlay - is shown on
 * the image. `acquire(blob)` is the single entry point on purpose: the agent
 * camera (#3) will call it with its own blob and reuse everything below.
 *
 * Config comes from window.__visionCfg (set by the template):
 *   executeUrl, imageUrlTpl (with a 0 placeholder for the result id), taskId,
 *   taskType, subtasks, labels.
 * Run context (lot, serial, production run) is read lazily through `ctx()`,
 * which the template binds to the parent task component.
 */
 function visionExecute() {
  // Overlay colour per role (VisionSuite roles: reference, roi, found).
  const ROLE_COLORS = { reference: '#2563eb', found: '#16a34a', roi: '#f59e0b' };
  return {
    busy: false,
    error: '',
    result: null,      // last execute response
    imageSrc: '',      // object URL of the acquired image (shown under the overlay)
    ctx: () => ({}),   // replaced by the template: {lot, serial, runId}
    // Entry point shared with the future agent camera.
    async acquire(blob) {
      if (!blob || this.busy) return;
      const cfg = window.__visionCfg || {};
      this.busy = true;
      this.error = '';
      this.result = null;
      // Show the picked image at once: the overlay is drawn on top when the result arrives.
      if (this.imageSrc) URL.revokeObjectURL(this.imageSrc);
      this.imageSrc = URL.createObjectURL(blob);
      try {
        const c = this.ctx() || {};
        const fd = new FormData();
        fd.append('task_id', cfg.taskId);
        fd.append('task_type', cfg.taskType || '');
        fd.append('image', blob, blob.name || 'capture.jpg');
        fd.append('lot_number', c.lot || '');
        fd.append('serial_number', c.serial || '');
        if (c.runId) fd.append('production_run_id', c.runId);
        const csrf = document.querySelector('meta[name=csrf-token]')?.content || '';
        const resp = await fetch(cfg.executeUrl, {
          method: 'POST', headers: { 'X-CSRFToken': csrf }, body: fd,
        });
        const body = await resp.json().catch(() => ({}));
        if (!resp.ok || body.error) {
          this.error = body.detail || (cfg.labels && cfg.labels.requestFailed) || 'Error';
          return;
        }
        this.result = body;
        // Draw once Alpine has rendered the canvas.
        this.$nextTick(() => this.drawOverlay());
      } catch (e) {
        this.error = (cfg.labels && cfg.labels.requestFailed) || 'Error';
      } finally {
        this.busy = false;
      }
    },
    // Input change: take the file and allow picking the same one again.
    onPick(ev) {
      const f = ev.target.files && ev.target.files[0];
      ev.target.value = '';
      if (f) this.acquire(f);
    },
    // Image shown: the saved one from the server when available, else the local one.
    get shownImage() {
      const cfg = window.__visionCfg || {};
      if (this.result && this.result.vision_result_id && cfg.imageUrlTpl) {
        return cfg.imageUrlTpl.replace('/0/', '/' + this.result.vision_result_id + '/');
      }
      return this.imageSrc;
    },
    // Quote rows: server measurements joined with the subtask names.
    get rows() {
      const cfg = window.__visionCfg || {};
      const subs = cfg.subtasks || [];
      return ((this.result && this.result.measurements) || []).map((m) => {
        const st = subs.find((s) => s.id === m.subtask_id) || {};
        return {
          label: (st.marker_number ? st.marker_number + '. ' : '') + (st.measurement_type || st.name || '#' + m.subtask_id),
          value: m.value,
          unit: st.unit || 'mm',
          status: m.pass_fail,
        };
      });
    },
    statusClass(s) {
      // Same palette as the measurement markers of the manual flow.
      if (s === 'pass') return 'bg-measure-pass/20 text-measure-pass border border-measure-pass/40';
      if (s === 'fail') return 'bg-red-100 dark:bg-red-900/30 text-measure-fail border border-measure-fail/40';
      if (s === 'warning') return 'bg-amber-100 dark:bg-amber-900/30 text-measure-warning border border-measure-warning/40';
      return 'bg-[var(--bg-secondary)] text-[var(--text-muted)] border border-[var(--border-color)]';
    },
    // Canvas is sized to the natural image, so overlay coordinates (image px)
    // are used as they are and CSS scales the whole thing to the displayed size.
    drawOverlay() {
      const img = this.$refs.img, cv = this.$refs.overlay;
      if (!img || !cv || !img.naturalWidth) return;
      cv.width = img.naturalWidth;
      cv.height = img.naturalHeight;
      const g = cv.getContext('2d');
      g.clearRect(0, 0, cv.width, cv.height);
      g.lineWidth = Math.max(2, cv.width / 400);
      g.lineJoin = 'round';
      for (const item of ((this.result && this.result.overlay) || [])) {
        const rows = item.rows || [], cols = item.cols || [];
        if (!rows.length || rows.length !== cols.length) continue;
        g.strokeStyle = ROLE_COLORS[item.role] || '#ef4444';
        g.beginPath();
        g.moveTo(cols[0], rows[0]);
        for (let i = 1; i < rows.length; i++) g.lineTo(cols[i], rows[i]);
        g.stroke();
      }
    },
  };
 }
@@ -191,7 +191,8 @@
  {# ================================================================
     MAIN — 3-column layout
     ================================================================ #}
-  <div class="flex-1 flex overflow-hidden">
+  {# #46: camera_measure / dxf_compare get the acquisition panel below instead of the manual layout #}
  <div class="flex-1 flex overflow-hidden" x-show="!isVisionTask">
    {# ──────────────────────────────────────────────
       LEFT SIDEBAR — Marker list (vertical, hidden when no subtasks)
@@ -451,6 +452,63 @@
  </div>
  {# ================================================================
     VISION PANEL (#46) - acquire an image, run the recipe server-side, show the
     quotes and the overlay. acquire(blob) is the entry point the agent camera
     (#3) will reuse.
     ================================================================ #}
  <div x-show="isVisionTask" x-cloak
       x-data="visionExecute()"
       x-init="ctx = () => ({ lot: lotNumber, serial: serialNumber, runId: productionRun ? productionRun.id : null })"
       class="flex-1 overflow-auto p-3 flex flex-col lg:flex-row gap-3">
    <div class="flex-1 min-w-0 flex flex-col gap-2">
      <div class="flex items-center gap-2 flex-wrap">
        {# On a tablet this opens the camera; on a PC it picks a file. «Riprova» reuses the same input. #}
        <label class="btn btn-primary text-sm gap-1.5 cursor-pointer" :class="busy ? 'opacity-50 pointer-events-none' : ''">
          <span x-text="result ? '{{ _('Riprova') }}' : '{{ _('Acquisisci immagine') }}'"></span>
          <input type="file" accept="image/*" capture="environment" class="hidden"
                 @change="onPick($event)" :disabled="busy">
        </label>
        <span x-show="busy" class="text-sm text-[var(--text-muted)]">{{ _('Elaborazione in corso...') }}</span>
      </div>
      <div x-show="error" class="p-2 rounded-lg bg-red-50 dark:bg-red-900/20 border border-red-200 dark:border-red-800 text-sm text-measure-fail" x-text="error"></div>
      <div x-show="shownImage" class="relative border border-[var(--border-color)] rounded-lg overflow-hidden bg-black/5">
        <img x-ref="img" :src="shownImage" @load="drawOverlay()" alt="" class="block w-full h-auto">
        <canvas x-ref="overlay" class="absolute inset-0 w-full h-full pointer-events-none"></canvas>
      </div>
      <div x-show="!shownImage" class="text-sm text-[var(--text-muted)]">
        {{ _('Nessuna immagine acquisita: premi «Acquisisci immagine».') }}
      </div>
    </div>
    <div class="lg:w-80 shrink-0 flex flex-col gap-2" x-show="result">
      <div class="text-sm font-bold text-[var(--text-primary)]">
        {{ _('Esito') }}: <span x-text="(result && result.outcome) || '-'"></span>
      </div>
      <table class="w-full text-sm" x-show="rows.length">
        <template x-for="r in rows" :key="r.label">
          <tr>
            <td class="py-1 pr-2 text-[var(--text-secondary)]" x-text="r.label"></td>
            <td class="py-1 pr-2 font-mono text-right" x-text="Number(r.value).toFixed(3) + ' ' + r.unit"></td>
            <td class="py-1"><span class="inline-block px-2 py-0.5 rounded text-xs font-bold" :class="statusClass(r.status)" x-text="r.status"></span></td>
          </tr>
        </template>
      </table>
      <div x-show="result && result.failures && result.failures.length"
           class="p-2 rounded-lg bg-amber-50 dark:bg-amber-900/20 border border-amber-300 dark:border-amber-700 text-xs">
        <div class="font-bold mb-1">{{ _('Strumenti falliti') }}</div>
        <template x-for="f in ((result && result.failures) || [])" :key="f.tool_id">
          <div><span class="font-mono" x-text="f.tool_name || f.tool_id"></span>: <span x-text="f.error"></span></div>
        </template>
      </div>
    </div>
  </div>
  {# ================================================================
     MEASUREMENT TIMER BANNER
     ================================================================ #}
@@ -912,6 +970,18 @@
 <script src="{{ url_for('static', filename='js/numpad.js') }}"></script>
 <script src="{{ url_for('static', filename='js/annotation-viewer.js') }}?v=6"></script>
 <script src="{{ url_for('static', filename='js/caliper.js') }}"></script>
 {# #46: acquisition + overlay for camera_measure / dxf_compare #}
 <script>
  window.__visionCfg = {
    executeUrl: '{{ url_for("measure.api_vision_execute") }}',
    imageUrlTpl: '{{ url_for("measure.api_vision_result_image", result_id=0) }}',
    taskId: {{ task.id|tojson }},
    taskType: {{ (task.task_type or '')|tojson }},
    subtasks: window.__taskSubtasks,
    labels: { requestFailed: {{ _('Richiesta non riuscita')|tojson }} },
  };
 </script>
 <script src="{{ url_for('static', filename='js/vision_execute.js') }}"></script>
 {# The clock's rules - how a delay reads, when the buzzer goes, where the
   measurement lives - are shared with the other screens rather than restated. #}
 <script src="{{ url_for('static', filename='js/production-clock.js') }}"></script>
@@ -1776,6 +1846,12 @@ function taskExecute() {
      return declared === 'measure' || declared === 'camera_measure';
    },
    /* #46: image-based tasks run a VisionSuite recipe on the server. */
    get isVisionTask() {
      const t = this.task.task_type;
      return t === 'camera_measure' || t === 'dxf_compare';
    },
    /* A measurement task with no quotes yet: the recipe is unfinished. Said out
       loud rather than silently rendering an empty screen. */
    get isMeasureTaskWithoutQuotes() {
@@ -0,0 +1,86 @@
 """gitea:Adriano/TieMeasureFlow#46 - operator side of the vision tasks.
 The backend of contract B is not merged yet: api_client is mocked.
 """
 import io
 from unittest.mock import MagicMock, patch
 from tests.conftest import api_get_router
 TASK = {
    "id": 12, "recipe_id": 3, "version_id": 5, "task_type": "camera_measure",
    "title": "Foto", "directive": None, "description": None, "file_path": None,
    "file_type": None, "annotations_json": None, "subtasks": [],
 }
 RECIPE = {
    "id": 3, "code": "REC-1", "name": "Ricetta",
    "measurement_interval_minutes": 30, "allow_manual_input": True,
    "current_version": {"id": 5, "version_number": 1},
 }
 def test_execute_page_has_acquire_input(logged_in_client, mock_api_client, monkeypatch):
    """camera_measure screen offers the file/camera input and the acquire entry point."""
    monkeypatch.setattr("blueprints.measure.Config.STATION_CODE", "ST-TEST")
    mock_api_client.get_station_recipes.return_value = {"items": [{"id": 3}]}
    mock_api_client.get.side_effect = api_get_router({
        "/api/tasks/12": TASK,
        "/api/recipes/3/tasks": [{"id": 12, "order_index": 0}],
        "/api/recipes/3": RECIPE,
    })
    html = logged_in_client.get("/measure/execute/12").get_data(as_text=True)
    assert 'capture="environment"' in html
    assert 'accept="image/*"' in html
    assert "visionExecute()" in html
    assert "/measure/api/vision/execute" in html
 def test_vision_execute_proxy_forwards_multipart(logged_in_client, mock_api_client):
    mock_api_client.post.return_value = {
        "vision_result_id": 7, "outcome": "ok", "overlay": [], "failures": [],
        "measurements": [{"subtask_id": 1, "value": 9.9, "pass_fail": "pass"}],
    }
    resp = logged_in_client.post(
        "/measure/api/vision/execute",
        data={
            "task_id": "12", "lot_number": "L1", "serial_number": "S1",
            "production_run_id": "4",
            "image": (io.BytesIO(b"\xff\xd8\xff fake"), "shot.jpg", "image/jpeg"),
        },
        content_type="multipart/form-data",
    )
    assert resp.status_code == 200
    assert resp.get_json()["vision_result_id"] == 7
    args, kwargs = mock_api_client.post.call_args
    assert args[0] == "/api/vision/execute"
    assert kwargs["data"]["task_id"] == "12"
    assert kwargs["data"]["lot_number"] == "L1"
    assert kwargs["data"]["production_run_id"] == "4"
    name, content, mime = kwargs["files"]["image"]
    assert name == "shot.jpg" and mime == "image/jpeg" and content.startswith(b"\xff\xd8")
 def test_vision_execute_proxy_requires_image(logged_in_client, mock_api_client):
    resp = logged_in_client.post("/measure/api/vision/execute", data={"task_id": "12"})
    assert resp.status_code == 400
    mock_api_client.post.assert_not_called()
 def test_vision_execute_proxy_relays_backend_error(logged_in_client, mock_api_client):
    mock_api_client.post.return_value = {"error": True, "status_code": 422, "detail": "bad"}
    resp = logged_in_client.post(
        "/measure/api/vision/execute",
        data={"task_id": "12", "image": (io.BytesIO(b"x"), "a.jpg", "image/jpeg")},
        content_type="multipart/form-data",
    )
    assert resp.status_code == 422
    assert resp.get_json()["detail"] == "bad"
 def test_vision_result_image_proxy(logged_in_client):
    fake = MagicMock(status_code=200, content=b"IMG", headers={"content-type": "image/jpeg"})
    with patch("blueprints.measure.http_requests.get", return_value=fake) as g:
        resp = logged_in_client.get("/measure/api/vision/results/7/image")
    assert resp.status_code == 200 and resp.data == b"IMG"
    assert g.call_args[0][0].endswith("/api/vision/results/7/image")
    assert g.call_args[1]["headers"]["X-API-Key"] == "test-api-key-12345"