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 math
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"


class GraphBuildError(ValueError):
    """`Task.from_dict` refused the graph itself - a bad request, not a
    server fault (a missing "kind", an unknown schema version, a graph that
    is not even a dict). Kept apart from an error raised once the task is
    actually running: the worker maps this one to 422, everything else that
    can go wrong inside `task.run()` to 500."""


@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
    # A tool that "succeeded" but handed back NaN/inf (vs-measure emits NaN on
    # an empty slice - see caliper.py:112) or a bool (a subclass of int, so
    # `isinstance(x, (int, float))` alone would let it through as a quote).
    # Kept apart from `failures`: the tool did not raise, it just produced a
    # number the caller must not measure a part by. Value is the reason.
    invalid_outputs: dict[str, str] = field(default_factory=dict)


@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.

    Two phases, on purpose: building the task from `graph` can fail on
    nothing but the caller's input (a bad schema version, a tool with no
    "kind", a graph that is not a dict at all) - that is `GraphBuildError`,
    a 422 for the worker. Once the task exists, `task.run()` is the engine's
    own business; an exception from there is not the caller's mistake, and
    the worker must not call it one.
    """
    try:
        task = Task.from_dict(graph)
    except (KeyError, TypeError, AttributeError, ValueError) as exc:
        raise GraphBuildError(
            f"malformed graph: {type(exc).__name__}: {exc}"
        ) from exc

    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] = []
    invalid_outputs: dict[str, str] = {}
    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():
            key = f"{result.tool_id}.{name}"
            # bool is a subclass of int - a tool that outputs a flag must not
            # be read as a quote.
            if isinstance(value, bool) or not isinstance(value, (int, float)):
                continue
            value = float(value)
            if not math.isfinite(value):
                invalid_outputs[key] = "nan" if value != value else "infinite"
                continue
            outputs[key] = value

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