TieMeasureFlow/src/vision/runner.py

"""Graph plus image in, numbers out.

The one implementation of vision execution: the worker beside the server and
the station agent both call this, and they must agree to the digit. No HTTP, no
database, no filesystem beyond reading the pinned submodule's commit.
"""
from __future__ import annotations

import os
import subprocess
import time
from dataclasses import dataclass, field
from functools import lru_cache
from pathlib import Path
from typing import Any

import numpy as np
from visionsuite.core.image import Image
from visionsuite.task import Task

_SUBMODULE = Path(__file__).resolve().parents[2] / "vendor" / "visionsuite"


@dataclass
class ToolFailure:
    tool_id: str
    tool_name: str
    error: str


@dataclass
class RunOutcome:
    outputs: dict[str, float] = field(default_factory=dict)
    failures: list[ToolFailure] = field(default_factory=list)
    engine_version: str = ""
    duration_ms: int = 0


@lru_cache(maxsize=1)
def _engine_version_from_git() -> str:
    """Read the pinned commit from the submodule checkout.

    Cached because the subprocess call is not free and the answer cannot
    change within a run. Kept separate from `engine_version()` so that the
    (cheap) environment-variable check below is never served a stale answer,
    and so a test can force this path to fail without poisoning the cache
    for real callers.
    """
    try:
        return subprocess.check_output(
            ["git", "-C", str(_SUBMODULE), "rev-parse", "HEAD"],
            text=True,
            stderr=subprocess.PIPE,
        ).strip()
    except (OSError, subprocess.CalledProcessError) as exc:
        raise RuntimeError(
            "engine_version: cannot determine which VisionSuite commit this "
            "process is running. VISION_ENGINE_VERSION is not set, and "
            f"`git -C {_SUBMODULE} rev-parse HEAD` failed ({exc}). "
            "In the vision worker container this is expected - it has no "
            "usable .git for the vendored submodule - so the image must be "
            "built with `--build-arg VISION_ENGINE_VERSION=<commit>` "
            "(see Dockerfile.vision). In a development checkout, make sure "
            "the vendor/visionsuite submodule is initialised. Refusing to "
            "guess: a station must not measure with an engine it cannot name."
        ) from exc


def engine_version() -> str:
    """The pinned VisionSuite commit: stamped on every number produced.

    Resolution order: the VISION_ENGINE_VERSION environment variable (set at
    container build time - Ruling R8, see Dockerfile.vision) if set and
    non-empty; otherwise `git rev-parse HEAD` on the vendored submodule,
    which works in a development checkout. Never falls back to a guess, an
    empty string, or "unknown" - a station running an engine it cannot name
    must refuse to measure rather than produce numbers nobody can trace.
    """
    stamped = os.environ.get("VISION_ENGINE_VERSION", "").strip()
    if stamped:
        return stamped
    return _engine_version_from_git()


def run_graph(
    image: Any,
    graph: dict,
    calibration: dict | None = None,
) -> RunOutcome:
    """Run a serialised vs-task graph against one image.

    A tool that fails does not stop the others - on a crooked piece an edge may
    simply not be there, and the other quotes are still wanted. Failures come
    back named, they are not swallowed.

    `image` is whatever the caller already has in hand: a plain numpy array is
    wrapped into the `visionsuite.core.image.Image` that vs-measure's tools
    actually read via `.pixels`; an `Image` passed in is used as-is.
    """
    task = Task.from_dict(graph)
    wrapped_image = image if isinstance(image, Image) else Image(pixels=np.asarray(image))
    context: dict[str, Any] = {"image": wrapped_image}
    if calibration is not None:
        context["calibration"] = calibration

    started = time.monotonic()
    report = task.run(context=context)
    elapsed_ms = int((time.monotonic() - started) * 1000)

    outputs: dict[str, float] = {}
    failures: list[ToolFailure] = []
    for result in report:
        if not result.ok:
            failures.append(ToolFailure(
                tool_id=result.tool_id,
                tool_name=result.tool_name,
                error=result.error or "unknown failure",
            ))
            continue
        for name, value in result.outputs.items():
            if isinstance(value, (int, float)):
                outputs[f"{result.tool_id}.{name}"] = float(value)

    return RunOutcome(
        outputs=outputs,
        failures=failures,
        engine_version=engine_version(),
        duration_ms=elapsed_ms,
    )