feat(vision): add Layer-2 SampleVisionMonitor + hardware-free testing setup

SampleVisionMonitor (Layer 2): per-tag, per-camera in-memory reference store + best_reference_match() scoring across a configured VisionInterface's cameras, publishing a worst-camera diagnostic overlay. Adds VisionInterface.get_raw_frame()/publish_diagnostic_image() as the supported hooks Layer 2 needs from Layer 1. Testing setup, per request: DummyVisionCamera (a settable, get_last_image()-only camera stand-in -- no pyueye/Aravis, so it proves the pipeline is genuinely camera-agnostic, not just IDSCamera-shaped), device_configs/simulated_omny/vision_test.yaml (two dummy cameras + one VisionInterface + one SampleVisionMonitor, loadable into any running session via bec.config.add_to_session() -- no session/main.yaml wiring needed), and AI_docs/VISION_PIPELINE_TESTING_HOWTO.md walking through it. Found and documented while verifying the howto's own examples against real code (not asserted from reading): register_to_reference()'s ECC alignment has no bound on corrected translation, so pure positional drift is currently invisible to best_reference_match() -- a same-shaped object that has simply moved registers back into alignment and scores as unchanged. Missing/deformed defects are unaffected. Flagged in both the howto and the plan doc's implementation-status section as a known gap needing a real design decision, not fixed in this pass. 24 new tests (7 SampleVisionMonitor, 2 new VisionInterface methods, 7 DummyVisionCamera, 1 shared test-pattern helper -- also deduplicated into vision_toolkit.make_square_test_pattern(), reused by three test files and DummyVisionCamera itself). Full suite: 351 passed. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018Mupip8x7Hean6E3FFvpbj
2026-10-26 14:15:45 +01:00
1 parent 5b2b098158
commit 224a10dace
@@ -15,9 +15,27 @@ this plan grows out of — see its "full gripper sample-transfer chain... untest
  `device_manager.devices.get()`, runtime-selectable mode (`raw`/`grayscale`/`edges`/`clahe`), own
  `diagnostic_image` `PreviewSignal`, `refresh()`/`get_last_processed_image()`. Not yet wired into
  any device config (no YAML entry, no real camera exercised) or into a GUI dock.
- **Layer 2 (`SampleVisionMonitor`) — not started.** Next step.
- **CLI wrapper methods (`omny.ovision_*`), GUI mode-picker, `x_ray_eye_align.py` integration — not
-  started.**
+- **Layer 2 (`csaxs_bec/devices/vision/sample_vision_monitor.py`, `SampleVisionMonitor`) — done,
+  tested (7 tests).** Per-tag, per-camera in-memory reference store + `best_reference_match()`
+  scoring across a `VisionInterface`'s cameras, worst-camera diagnostic overlay published via
+  `publish_diagnostic_image()`. Not yet wired into any OMNY confirm gate.
+- **Testing tools — done.** `DummyVisionCamera` (`csaxs_bec/devices/vision/dummy_vision_camera.py`,
+  a settable, hardware-free `get_last_image()` stand-in) +
+  `device_configs/simulated_omny/vision_test.yaml` (two dummy cameras + one `VisionInterface` + one
+  `SampleVisionMonitor`, loadable into any running session via `bec.config.add_to_session(...)`) +
+  `AI_docs/VISION_PIPELINE_TESTING_HOWTO.md` (step-by-step walkthrough).
+- **⚠ Known limitation, found via the above testing tools, not yet fixed**: `register_to_reference()`
+  has no bound on how much translation its ECC alignment will correct, so a same-shaped object that
+  has simply *moved* is registered right back into alignment and scores as unchanged — pure
+  positional drift is currently invisible to `best_reference_match()`. This is backwards for OMNY's
+  actual need ("the pin isn't where it should be" is exactly the kind of thing a confirm gate must
+  catch). Missing/deformed defects are unaffected and verified working. See
+  `VISION_PIPELINE_TESTING_HOWTO.md`'s "Known limitation" section for the reproduction and a sketch
+  of two possible fixes (bound the tolerated translation; or run a supplementary un-registered
+  comparison) — needs a real design decision before Phase 0 data collection can be trusted for
+  drift-type defects specifically.
+- **CLI wrapper methods (`omny.ovision_*`), GUI mode-picker, `x_ray_eye_align.py` integration,
+  confirm-gate wiring — not started.**

 ## Motivation

@@ -0,0 +1,146 @@
+# Testing the vision pipeline without a camera (2026-09-06)
+
+How to exercise `VisionInterface`/`SampleVisionMonitor` (Layers 1/2 of
+`MACHINE_VISION_AUTOCONFIRM_PLAN.md`) end to end -- including a realistic diagnostic-overlay check
+-- with no OMNY hardware and no real camera attached. Everything below runs against
+`csaxs_bec/device_configs/simulated_omny/vision_test.yaml`, a small standalone config with two
+`DummyVisionCamera` instances (a settable, hardware-free camera stand-in -- see
+`csaxs_bec/devices/vision/dummy_vision_camera.py`) wired to one `VisionInterface` and one
+`SampleVisionMonitor`.
+
+## 1. Load the test devices into a running session
+
+Start any existing BEC session (a bare simulated session is enough -- this doesn't need OMNY's own
+devices at all) and merge the extra config in, without touching or replacing anything already
+loaded:
+
+```python
+bec.config.add_to_session(
+    "csaxs_bec/device_configs/simulated_omny/vision_test.yaml"
+)
+```
+
+(`bec.config` is `bec_lib.config_helper.ConfigHelperUser` -- `add_to_session()` merges devices from
+a file into the *current* session; it does not touch existing devices, so it's safe to run against
+whatever session is already up.)
+
+After it returns, `dev.vision_cam_a`, `dev.vision_cam_b`, `dev.vision_test_interface`, and
+`dev.vision_test_monitor` should all be available, same as any other device.
+
+## 2. Sanity check: set a frame and look at it
+
+```python
+dev.vision_cam_a.set_test_pattern()  # synthetic white square on black, centered
+dev.vision_cam_a.get_last_image().shape  # (200, 200, 3)
+
+dev.vision_test_interface.get_raw_frame("vision_cam_a")  # same array, via the interface
+```
+
+## 3. Try the processing modes
+
+```python
+dev.vision_test_interface.set_mode("edges")
+dev.vision_test_interface.refresh("vision_cam_a")  # publishes to diagnostic_image
+dev.vision_test_interface.diagnostic_image.get().data.shape  # (200, 200) -- 2D edge map
+
+dev.vision_test_interface.set_mode("raw")
+```
+
+There's no GUI dock wired up for `diagnostic_image` yet (see the plan doc's open questions) -- to
+actually *see* the published frame today, either pull `.data` from `diagnostic_image.get()` and
+save it (e.g. `cv2.imwrite("/tmp/check.png", ...)`) or point a `bec_widgets` `Image` widget at
+`vision_test_interface`/`diagnostic_image` manually, the same way any other device's preview signal
+is wired into a dock.
+
+## 4. The actual scoring workflow (this is the Phase-0 use case)
+
+This is what `SampleVisionMonitor` gives OMNY's confirm gates: does the current view still look
+like what was saved before at this step.
+
+```python
+# Set both cameras to a "known-good" frame and save it as the reference for a tag
+# (in real use the tag would be e.g. f"{step_name}_{pin_position}" -- here just pick anything).
+dev.vision_cam_a.set_test_pattern(offset=(0, 0))
+dev.vision_cam_b.set_test_pattern(offset=(0, 0))
+dev.vision_test_monitor.save_reference("test_tag")
+
+# Score against an identical frame -- should be a low (near-zero) score on both cameras.
+dev.vision_test_monitor.score("test_tag")
+# {'tag': 'test_tag', 'overall_score': ~0.0, 'worst_camera': 'vision_cam_a' or 'vision_cam_b',
+#  'per_camera': {'vision_cam_a': {'score': ~0.0, 'reference_count': 1, 'converged': True}, ...}}
+
+# Now simulate a real problem on one camera -- the shape genuinely changes (not just moves --
+# see "Known limitation: pure translation is currently invisible" below for why that distinction
+# matters), e.g. the pin/square is no longer there at all.
+import numpy as np
+dev.vision_cam_b.set_image(np.zeros((200, 200, 3), dtype=np.uint8))
+dev.vision_test_monitor.score("test_tag")
+# overall_score jumps, worst_camera == "vision_cam_b" -- and diagnostic_image on
+# vision_test_interface now shows vision_cam_b's (empty) frame with no contour to draw.
+```
+
+A camera with nothing saved under a tag yet scores `inf`, not an error -- try
+`dev.vision_test_monitor.score("never_saved")` (after the cameras already have *some* frame set,
+e.g. from step 2/3 above -- a camera with no frame at all raises instead, since that's a real setup
+error, not "no references yet") to see this (expected state, not a bug).
+
+## Known limitation, found while writing this howto: pure translation is currently invisible
+
+Verified directly, not a hypothetical: shifting the square (as the very first draft of this howto's
+step 4 did, `set_test_pattern(offset=(30, 0))`) does **not** raise the score at all --
+`best_reference_match()` scores it at `0.0`, identical to an unchanged frame. Reproduce:
+
+```python
+from csaxs_bec.devices.vision import vision_toolkit as vt
+current = vt.make_square_test_pattern(offset=(0, 0))
+shifted = vt.make_square_test_pattern(offset=(30, 0))
+vt.best_reference_match(current, [shifted]).best_score  # 0.0, not "different"
+```
+
+Root cause: `register_to_reference()`'s ECC alignment has no bound on how much translation it will
+correct -- it estimates and removes *any* translation between the two frames before the contour
+comparison ever runs, so a same-shaped object that has simply moved is registered right back into
+alignment and then, correctly by the scoring step's own logic, looks identical. This is exactly
+backwards for OMNY's actual use case: registration is meant to compensate for small *camera* mount
+drift, not to silently erase genuine *sample* position changes -- and "the pin isn't where it should
+be" is one of the more important things a confirm gate needs to catch, not blind itself to.
+
+**Not fixed here** -- this needs a real design decision (e.g. cap `register_to_reference()`'s
+tolerated translation to a small camera-drift-sized bound and treat anything larger as itself part
+of the score, or run a supplementary un-registered comparison alongside the registered one) before
+Phase 0 data collection can be trusted for drift-type defects specifically. Missing/deformed defects
+(the `np.zeros(...)` case above, and `test_score_flags_a_camera_that_no_longer_matches_its_reference`
+in `test_sample_vision_monitor.py`) are unaffected and already verified working.
+
+## 5. A more realistic manual test: a real photo instead of a synthetic square
+
+`DummyVisionCamera.load_image_file(path)` reads an image file from disk (jpg/png/...) -- e.g. a
+phone photo of an actual gripper/pin, transferred to whatever machine the BEC session is running
+on:
+
+```python
+dev.vision_cam_a.load_image_file("/path/to/photo_of_gripper_ok.jpg")
+dev.vision_test_monitor.save_reference("manual_check")
+
+dev.vision_cam_a.load_image_file("/path/to/photo_of_gripper_moved.jpg")
+dev.vision_test_monitor.score("manual_check")
+```
+
+This is the best available proxy for real hardware behavior right now: take two photos of the same
+physical setup (one "normal", one with something deliberately off -- pin rotated, missing, shifted)
+and see whether the contour-matching score and the published overlay actually track the difference
+a human would notice. Useful for sanity-checking `edge_low_threshold`/`edge_high_threshold`
+(`vision_toolkit.detect_edges()`'s defaults, `50`/`150`) against real lighting/background, which the
+synthetic square can't tell you anything about.
+
+## What this does and doesn't prove
+
+Proves: the Layer 0/1/2 code paths work correctly end to end (registration, contour scoring,
+multi-reference best-match, diagnostic publishing) against arbitrary images, camera-agnostic (a
+`DummyVisionCamera` is not an `IDSCamera`/`AlliedVisionAravisCamera` and has none of their code).
+
+Does **not** prove: anything about real OMNY hardware, real exposure/lighting variation at the
+beamline, or whether the confirm gates in `omny_sample_transfer_mixin.py` actually get wired up to
+call this -- that wiring hasn't been written yet (see the plan doc's implementation-status section).
+Real-camera/real-lighting validation still needs either real photos (step 5) or, eventually, a real
+OMNY session.
@@ -0,0 +1,61 @@
+# Standalone, hand-maintained config for testing the vision pipeline
+# (AI_docs/MACHINE_VISION_AUTOCONFIRM_PLAN.md) without real camera hardware.
+#
+# Not part of simulated_omny.yaml (which is regenerated from ptycho_omny.yaml -- see
+# SIMULATED_ENDSTATIONS.md) and not wired into any session by default. Load it into an
+# already-running session with:
+#
+#   bec.config.add_to_session("<path-to-this-file>/vision_test.yaml")
+#
+# See AI_docs/VISION_PIPELINE_TESTING_HOWTO.md for the full walkthrough.
+
+vision_cam_a:
+  description: Dummy vision-pipeline test camera A -- no hardware. Set its image via
+    dev.vision_cam_a.set_image()/set_test_pattern()/load_image_file().
+  deviceClass: csaxs_bec.devices.vision.dummy_vision_camera.DummyVisionCamera
+  deviceConfig: {}
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: on_request
+  deviceTags:
+    - vision_test
+
+vision_cam_b:
+  description: Dummy vision-pipeline test camera B -- no hardware. Second camera of the
+    simulated pair, mirroring OMNY's cam200+cam203/cam201+cam202 confirm-gate camera pairs.
+  deviceClass: csaxs_bec.devices.vision.dummy_vision_camera.DummyVisionCamera
+  deviceConfig: {}
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: on_request
+  deviceTags:
+    - vision_test
+
+vision_test_interface:
+  description: VisionInterface (Layer 1) wired to the two dummy test cameras above.
+  deviceClass: csaxs_bec.devices.vision.vision_interface.VisionInterface
+  deviceConfig:
+    camera_names:
+      - vision_cam_a
+      - vision_cam_b
+    mode: raw
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: on_request
+  deviceTags:
+    - vision_test
+
+vision_test_monitor:
+  description: SampleVisionMonitor (Layer 2) scoring against vision_test_interface.
+  deviceClass: csaxs_bec.devices.vision.sample_vision_monitor.SampleVisionMonitor
+  deviceConfig:
+    vision_interface_name: vision_test_interface
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: on_request
+  deviceTags:
+    - vision_test
@@ -0,0 +1,79 @@
+"""A camera-contract stand-in with no hardware backing.
+
+For exercising :class:`~csaxs_bec.devices.vision.vision_interface.VisionInterface` and
+:class:`~csaxs_bec.devices.vision.sample_vision_monitor.SampleVisionMonitor` without a real camera
+-- see ``AI_docs/VISION_PIPELINE_TESTING_HOWTO.md`` for the step-by-step walkthrough this device is
+meant to support. Implements only the shared ``get_last_image()`` contract; the "current frame" is
+whatever was last set via :meth:`set_image`, :meth:`set_test_pattern`, or :meth:`load_image_file`,
+all callable directly from a BEC session with no hardware attached.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import cv2
+import numpy as np
+from ophyd import Component as Cpt
+from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
+from ophyd_devices.utils.bec_signals import PreviewSignal
+
+from csaxs_bec.devices.vision import vision_toolkit as vt
+
+if TYPE_CHECKING:
+    from bec_lib.devicemanager import ScanInfo
+
+
+class DummyVisionCameraError(Exception):
+    pass
+
+
+class DummyVisionCamera(PSIDeviceBase):
+    """A settable, hardware-free camera for vision-pipeline testing."""
+
+    image = Cpt(
+        PreviewSignal,
+        name="image",
+        ndim=2,
+        doc="Live preview of the currently-set still image, if any.",
+        num_rotation_90=0,
+        transpose=False,
+    )
+
+    USER_ACCESS = ["get_last_image", "set_image", "set_test_pattern", "load_image_file"]
+
+    def __init__(
+        self, *, name: str, prefix: str = "", scan_info: "ScanInfo | None" = None, **kwargs
+    ):
+        super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs)
+        self._image: np.ndarray | None = None
+
+    def get_last_image(self) -> np.ndarray | None:
+        return self._image
+
+    def set_image(self, image: np.ndarray) -> None:
+        """Set the currently-returned frame directly from an array."""
+        self._image = np.asarray(image)
+        self.image.put(self._image)
+
+    def set_test_pattern(
+        self, size: int = 200, square_side: int = 60, offset: tuple[int, int] = (0, 0)
+    ) -> None:
+        """Set the current frame to a synthetic square-on-black test pattern -- ``offset`` moves
+        the square, for exercising registration/scoring without any external image file."""
+        self.set_image(
+            vt.make_square_test_pattern(size=size, square_side=square_side, offset=offset)
+        )
+
+    def load_image_file(self, path: str) -> None:
+        """Load the current frame from an image file on disk (e.g. a phone photo of a real
+        gripper/pin) -- the most realistic way to manually test the scoring pipeline without a
+        real camera attached.
+
+        Converted to RGB to match ``get_last_image()``'s convention elsewhere (``IDSCamera``/
+        ``AlliedVisionAravisCamera`` both hand back RGB, not OpenCV's native BGR).
+        """
+        image = cv2.imread(path, cv2.IMREAD_COLOR)
+        if image is None:
+            raise DummyVisionCameraError(f"{self.name}: could not read image file '{path}'.")
+        self.set_image(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))
@@ -0,0 +1,134 @@
+"""Sample-transfer confirm/scoring application, built on Layers 0/1.
+
+Layer 2 of ``AI_docs/MACHINE_VISION_AUTOCONFIRM_PLAN.md``: composes a configured
+:class:`~csaxs_bec.devices.vision.vision_interface.VisionInterface` (for camera access and
+diagnostic-image publishing) with a per-tag, per-camera reference-image store and
+:func:`vision_toolkit.best_reference_match` scoring, so a caller (e.g. OMNY's
+``omny_sample_transfer_mixin.py`` confirm gates) can ask "does the current camera view match what
+we've seen before at this step" without doing any ``cv2`` work itself.
+
+Phase 0 (see the plan doc's staged rollout) only needs this to compute and expose a score; nothing
+here changes any confirm behavior on its own -- that stays entirely in the caller.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import numpy as np
+from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
+
+from csaxs_bec.devices.vision import vision_toolkit as vt
+
+if TYPE_CHECKING:
+    from bec_lib.devicemanager import ScanInfo
+
+
+class SampleVisionMonitorError(Exception):
+    pass
+
+
+class SampleVisionMonitor(PSIDeviceBase):
+    """Reference-tagged contour-match scoring for a camera pair, via a configured VisionInterface.
+
+    Reference images are opaque-tag-keyed and held in memory only, for this device's process
+    lifetime -- persisting/curating a "golden" reference set is an open question in the design doc
+    ("Reference corpus curation"), not resolved or implemented here.
+    """
+
+    USER_ACCESS = ["save_reference", "score", "reference_tags", "clear_reference"]
+
+    def __init__(
+        self,
+        *,
+        name: str,
+        vision_interface_name: str,
+        prefix: str = "",
+        scan_info: "ScanInfo | None" = None,
+        **kwargs,
+    ):
+        """Initialize the sample vision monitor.
+
+        Args:
+            name (str): Name of the device.
+            vision_interface_name (str): Name of the ``VisionInterface`` device this monitor scores
+                against, resolved through the device manager at call time (same lazy-lookup pattern
+                as ``VisionInterface``'s own camera dependency).
+        """
+        super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs)
+        self._vision_interface_name = vision_interface_name
+        # tag -> camera_name -> list of reference frames.
+        self._references: dict[str, dict[str, list[np.ndarray]]] = {}
+
+    def _resolve_vision_interface(self):
+        if self.device_manager is None or not hasattr(self.device_manager, "devices"):
+            raise SampleVisionMonitorError(f"{self.name}: device manager is not available.")
+        interface = self.device_manager.devices.get(self._vision_interface_name, None)
+        if interface is None:
+            raise SampleVisionMonitorError(
+                f"{self.name}: configured vision interface '{self._vision_interface_name}' was"
+                " not found in the current device config."
+            )
+        return interface
+
+    def save_reference(self, tag: str) -> None:
+        """Snapshot the current raw frame from every camera behind the configured
+        ``VisionInterface`` and store it as a reference under ``tag``."""
+        interface = self._resolve_vision_interface()
+        per_camera_references = self._references.setdefault(tag, {})
+        for camera_name in interface.camera_names:
+            frame = interface.get_raw_frame(camera_name)
+            per_camera_references.setdefault(camera_name, []).append(frame)
+
+    def reference_tags(self) -> list[str]:
+        return list(self._references)
+
+    def clear_reference(self, tag: str) -> None:
+        self._references.pop(tag, None)
+
+    def score(self, tag: str) -> dict:
+        """Register+score the current frame of every camera against the references stored under
+        ``tag``, publish a diagnostic contour overlay for the worst-scoring camera, and return a
+        summary a caller can turn into a confirm-dialog decision.
+
+        A camera with no stored references yet scores ``inf`` -- an expected state (nothing saved
+        under this tag yet), not an error; see :func:`vision_toolkit.best_reference_match`.
+        """
+        interface = self._resolve_vision_interface()
+        per_camera_references = self._references.get(tag, {})
+
+        per_camera_summary: dict[str, dict] = {}
+        worst_camera_name: str | None = None
+        worst_match: vt.MatchResult | None = None
+        worst_frame: np.ndarray | None = None
+
+        for camera_name in interface.camera_names:
+            current_frame = interface.get_raw_frame(camera_name)
+            match = vt.best_reference_match(
+                current_frame, per_camera_references.get(camera_name, [])
+            )
+            per_camera_summary[camera_name] = {
+                "score": match.best_score,
+                "reference_count": len(per_camera_references.get(camera_name, [])),
+                "converged": match.best.registration.converged if match.best is not None else None,
+            }
+            if worst_match is None or match.best_score > worst_match.best_score:
+                worst_camera_name = camera_name
+                worst_match = match
+                worst_frame = current_frame
+
+        overall_score = max(
+            (entry["score"] for entry in per_camera_summary.values()), default=float("inf")
+        )
+
+        if worst_match is not None and worst_match.best is not None and worst_frame is not None:
+            current_contour = vt.largest_contour(vt.detect_edges(worst_frame))
+            overlay = vt.draw_contour_overlay(worst_frame, current_contour)
+            interface.publish_diagnostic_image(overlay)
+
+        return {
+            "tag": tag,
+            "overall_score": overall_score,
+            "worst_camera": worst_camera_name,
+            "per_camera": per_camera_summary,
+        }
@@ -55,7 +55,15 @@ class VisionInterface(PSIDeviceBase):
        kind=Kind.config,
    )

-    USER_ACCESS = ["camera_names", "set_mode", "get_mode", "get_last_processed_image", "refresh"]
+    USER_ACCESS = [
+        "camera_names",
+        "set_mode",
+        "get_mode",
+        "get_last_processed_image",
+        "get_raw_frame",
+        "publish_diagnostic_image",
+        "refresh",
+    ]

    def __init__(
        self,
@@ -141,6 +149,22 @@ class VisionInterface(PSIDeviceBase):
        frame = self._get_camera_frame(resolved_name)
        return self._process(frame, self.get_mode())

+    def get_raw_frame(self, camera_name: str | None = None) -> np.ndarray:
+        """Pull the latest unprocessed frame from ``camera_name`` (default: the first configured
+        camera), bypassing the active mode -- for a caller (e.g. ``SampleVisionMonitor``) that
+        needs the raw frame regardless of what this interface's diagnostic view currently shows.
+        """
+        resolved_name = camera_name if camera_name is not None else self._camera_names[0]
+        return self._get_camera_frame(resolved_name)
+
+    def publish_diagnostic_image(self, image: np.ndarray) -> None:
+        """Publish an externally-computed image (e.g. a ``SampleVisionMonitor``'s diagnostic
+        overlay) through this interface's ``diagnostic_image`` channel -- the supported way for
+        another device-server-side object to reuse this interface's GUI-facing signal, mirroring
+        ``IDSCamera.push_preview_image()``/``push_smear_preview()``.
+        """
+        self.diagnostic_image.put(image)
+
    def refresh(self, camera_name: str | None = None) -> np.ndarray:
        """Pull, process, and publish the latest frame through ``diagnostic_image``.

@@ -149,5 +173,5 @@ class VisionInterface(PSIDeviceBase):
        needed by anything using this device yet.
        """
        processed = self.get_last_processed_image(camera_name)
-        self.diagnostic_image.put(processed)
+        self.publish_diagnostic_image(processed)
        return processed
@@ -47,6 +47,22 @@ class MatchResult:
        return self.best.contour_score if self.best is not None else math.inf


+def make_square_test_pattern(
+    size: int = 200, square_side: int = 60, offset: tuple[int, int] = (0, 0)
+) -> np.ndarray:
+    """A synthetic RGB frame: a white filled square on a black background.
+
+    For exercising the toolkit/devices without a real camera -- ``offset`` shifts the square from
+    center, e.g. to simulate a pin that has drifted out of position between a reference and the
+    current frame.
+    """
+    image = np.zeros((size, size, 3), dtype=np.uint8)
+    top_left = (size // 2 - square_side // 2 + offset[0], size // 2 - square_side // 2 + offset[1])
+    bottom_right = (top_left[0] + square_side, top_left[1] + square_side)
+    cv2.rectangle(image, top_left, bottom_right, color=(255, 255, 255), thickness=-1)
+    return image
+
+
 def to_grayscale(image: np.ndarray) -> np.ndarray:
    """Convert an image to single-channel grayscale.

@@ -0,0 +1,65 @@
+"""Unit tests for the hardware-free test camera (csaxs_bec/devices/vision/dummy_vision_camera.py)."""
+
+import numpy as np
+import pytest
+
+from csaxs_bec.devices.vision import vision_toolkit as vt
+from csaxs_bec.devices.vision.dummy_vision_camera import DummyVisionCamera, DummyVisionCameraError
+
+
+@pytest.fixture
+def camera():
+    return DummyVisionCamera(name="dummy_cam_test")
+
+
+def test_get_last_image_is_none_before_anything_is_set(camera):
+    assert camera.get_last_image() is None
+
+
+def test_set_image_stores_and_publishes(camera):
+    image = np.zeros((10, 10, 3), dtype=np.uint8)
+    camera.set_image(image)
+    assert np.array_equal(camera.get_last_image(), image)
+    published = camera.image.get()
+    assert np.array_equal(published.data, image)
+
+
+def test_set_test_pattern_produces_a_square_image(camera):
+    camera.set_test_pattern(size=50, square_side=10)
+    frame = camera.get_last_image()
+    assert frame.shape == (50, 50, 3)
+    assert frame.sum() > 0
+
+
+def test_set_test_pattern_offset_moves_the_square(camera):
+    camera.set_test_pattern(size=50, square_side=10, offset=(0, 0))
+    centered = camera.get_last_image()
+    camera.set_test_pattern(size=50, square_side=10, offset=(15, 0))
+    shifted = camera.get_last_image()
+    assert not np.array_equal(centered, shifted)
+
+
+def test_set_test_pattern_matches_vision_toolkit_helper(camera):
+    camera.set_test_pattern(size=40, square_side=8, offset=(2, -2))
+    expected = vt.make_square_test_pattern(size=40, square_side=8, offset=(2, -2))
+    assert np.array_equal(camera.get_last_image(), expected)
+
+
+def test_load_image_file_missing_file_raises(camera, tmp_path):
+    with pytest.raises(DummyVisionCameraError):
+        camera.load_image_file(str(tmp_path / "does_not_exist.png"))
+
+
+def test_load_image_file_reads_and_converts_to_rgb(camera, tmp_path):
+    import cv2
+
+    bgr_image = np.zeros((10, 10, 3), dtype=np.uint8)
+    bgr_image[:, :, 0] = 200  # blue channel, in BGR storage order
+    file_path = tmp_path / "test_frame.png"
+    cv2.imwrite(str(file_path), bgr_image)
+
+    camera.load_image_file(str(file_path))
+    frame = camera.get_last_image()
+    assert frame.shape == (10, 10, 3)
+    # BGR blue (channel 0) should now show up as RGB's channel 2.
+    assert frame[0, 0, 2] > frame[0, 0, 0]
@@ -0,0 +1,126 @@
+"""Unit tests for the Layer-2 sample vision monitor device
+(csaxs_bec/devices/vision/sample_vision_monitor.py)."""
+
+import numpy as np
+import pytest
+from bec_server.device_server.tests.utils import DMMock
+
+from csaxs_bec.devices.vision import vision_toolkit as vt
+from csaxs_bec.devices.vision.sample_vision_monitor import (
+    SampleVisionMonitor,
+    SampleVisionMonitorError,
+)
+from csaxs_bec.devices.vision.vision_interface import VisionInterface
+
+
+class FakeCamera:
+    """Minimal stand-in for a camera device -- only get_last_image(), settable per test."""
+
+    def __init__(self, image: np.ndarray | None):
+        self.image = image
+
+    def get_last_image(self) -> np.ndarray | None:
+        return self.image
+
+
+def _square_image(size: int = 60, offset: int = 0) -> np.ndarray:
+    return vt.make_square_test_pattern(size=size, square_side=20, offset=(offset, offset))
+
+
+def _make_monitor(camera_names=("cam_a", "cam_b"), images=None) -> SampleVisionMonitor:
+    dm = DMMock()
+    images = images or {}
+    cameras = {name: FakeCamera(images.get(name)) for name in camera_names}
+    for name, camera in cameras.items():
+        dm.devices[name] = camera
+    interface = VisionInterface(
+        name="vision_interface_test", camera_names=list(camera_names), device_manager=dm
+    )
+    dm.devices["vision_interface_test"] = interface
+    monitor = SampleVisionMonitor(
+        name="vision_monitor_test", vision_interface_name="vision_interface_test", device_manager=dm
+    )
+    dm.devices["vision_monitor_test"] = monitor
+    monitor._cameras = cameras  # convenience handle for tests to mutate live frames
+    return monitor
+
+
+def test_missing_vision_interface_raises():
+    dm = DMMock()
+    monitor = SampleVisionMonitor(
+        name="vision_monitor_test", vision_interface_name="does_not_exist", device_manager=dm
+    )
+    with pytest.raises(SampleVisionMonitorError):
+        monitor.score("some_tag")
+
+
+def test_score_with_no_saved_references_is_infinite_but_does_not_raise():
+    monitor = _make_monitor(images={"cam_a": _square_image(), "cam_b": _square_image()})
+    result = monitor.score("put_mount_start_1")
+    assert result["overall_score"] == float("inf")
+    for camera_summary in result["per_camera"].values():
+        assert camera_summary["score"] == float("inf")
+        assert camera_summary["reference_count"] == 0
+
+
+def test_save_reference_then_score_against_identical_frame_scores_low():
+    image = _square_image()
+    monitor = _make_monitor(images={"cam_a": image, "cam_b": image})
+    monitor.save_reference("put_mount_start_1")
+
+    result = monitor.score("put_mount_start_1")
+
+    assert result["overall_score"] < float("inf")
+    for camera_summary in result["per_camera"].values():
+        assert camera_summary["reference_count"] == 1
+        assert camera_summary["score"] == pytest.approx(0.0, abs=1e-3)
+
+
+def test_score_flags_a_camera_that_no_longer_matches_its_reference():
+    matching_image = _square_image()
+    monitor = _make_monitor(images={"cam_a": matching_image, "cam_b": matching_image})
+    monitor.save_reference("put_mount_start_1")
+
+    # cam_b's live view changes to something with no square at all.
+    monitor._cameras["cam_b"].image = np.zeros((60, 60, 3), dtype=np.uint8)
+
+    result = monitor.score("put_mount_start_1")
+
+    assert result["worst_camera"] == "cam_b"
+    assert result["per_camera"]["cam_b"]["score"] > result["per_camera"]["cam_a"]["score"]
+    assert result["overall_score"] == result["per_camera"]["cam_b"]["score"]
+
+
+def test_score_publishes_a_diagnostic_overlay_for_the_worst_camera():
+    image = _square_image()
+    monitor = _make_monitor(images={"cam_a": image, "cam_b": image})
+    monitor.save_reference("put_mount_start_1")
+    monitor.score("put_mount_start_1")
+
+    interface = monitor._resolve_vision_interface()
+    published = interface.diagnostic_image.get()
+    assert published.data.shape[:2] == image.shape[:2]
+
+
+def test_reference_tags_and_clear_reference():
+    monitor = _make_monitor(images={"cam_a": _square_image(), "cam_b": _square_image()})
+    assert monitor.reference_tags() == []
+
+    monitor.save_reference("tag_a")
+    monitor.save_reference("tag_b")
+    assert set(monitor.reference_tags()) == {"tag_a", "tag_b"}
+
+    monitor.clear_reference("tag_a")
+    assert monitor.reference_tags() == ["tag_b"]
+
+    # Clearing an unknown tag is a no-op, not an error.
+    monitor.clear_reference("does_not_exist")
+
+
+def test_save_reference_accumulates_multiple_references_per_tag():
+    monitor = _make_monitor(images={"cam_a": _square_image(), "cam_b": _square_image()})
+    monitor.save_reference("tag_a")
+    monitor.save_reference("tag_a")
+
+    result = monitor.score("tag_a")
+    assert result["per_camera"]["cam_a"]["reference_count"] == 2
@@ -118,3 +118,18 @@ def test_refresh_second_configured_camera_by_name():
    )
    result = interface.refresh(camera_name="cam_b")
    assert np.array_equal(result, image_b)
+
+
+def test_get_raw_frame_bypasses_active_mode():
+    image = _rgb_image()
+    interface = _make_interface(images={"cam_a": image}, mode="edges")
+    result = interface.get_raw_frame()
+    assert np.array_equal(result, image)
+
+
+def test_publish_diagnostic_image_makes_an_externally_computed_image_available():
+    interface = _make_interface(images={"cam_a": _rgb_image()})
+    overlay = np.full((5, 5), 42, dtype=np.uint8)
+    interface.publish_diagnostic_image(overlay)
+    published = interface.diagnostic_image.get()
+    assert np.array_equal(published.data, overlay)
@@ -10,12 +10,14 @@ from csaxs_bec.devices.vision import vision_toolkit as vt
 def _square_image(
    size: int = 100, square_side: int = 30, offset: tuple[int, int] = (0, 0)
 ) -> np.ndarray:
-    """A synthetic RGB frame: a white filled square on a black background."""
-    image = np.zeros((size, size, 3), dtype=np.uint8)
-    top_left = (size // 2 - square_side // 2 + offset[0], size // 2 - square_side // 2 + offset[1])
-    bottom_right = (top_left[0] + square_side, top_left[1] + square_side)
-    cv2.rectangle(image, top_left, bottom_right, color=(255, 255, 255), thickness=-1)
-    return image
+    return vt.make_square_test_pattern(size=size, square_side=square_side, offset=offset)
+
+
+def test_make_square_test_pattern_shape_and_offset():
+    centered = vt.make_square_test_pattern(size=40, square_side=10, offset=(0, 0))
+    shifted = vt.make_square_test_pattern(size=40, square_side=10, offset=(5, 0))
+    assert centered.shape == (40, 40, 3)
+    assert not np.array_equal(centered, shifted)


 def test_to_grayscale_passes_through_2d():