From 224a10dace196cc1682691f150fb7ce3d9a37ea3 Mon Sep 17 00:00:00 2001 From: x01dc Date: Sun, 6 Sep 2026 13:53:02 +0200 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_018Mupip8x7Hean6E3FFvpbj --- .../MACHINE_VISION_AUTOCONFIRM_PLAN.md | 24 ++- .../AI_docs/VISION_PIPELINE_TESTING_HOWTO.md | 146 ++++++++++++++++++ .../simulated_omny/vision_test.yaml | 61 ++++++++ .../devices/vision/dummy_vision_camera.py | 79 ++++++++++ .../devices/vision/sample_vision_monitor.py | 134 ++++++++++++++++ csaxs_bec/devices/vision/vision_interface.py | 28 +++- csaxs_bec/devices/vision/vision_toolkit.py | 16 ++ .../tests_devices/test_dummy_vision_camera.py | 65 ++++++++ .../test_sample_vision_monitor.py | 126 +++++++++++++++ tests/tests_devices/test_vision_interface.py | 15 ++ tests/tests_devices/test_vision_toolkit.py | 14 +- 11 files changed, 697 insertions(+), 11 deletions(-) create mode 100644 csaxs_bec/bec_ipython_client/plugins/omny/AI_docs/VISION_PIPELINE_TESTING_HOWTO.md create mode 100644 csaxs_bec/device_configs/simulated_omny/vision_test.yaml create mode 100644 csaxs_bec/devices/vision/dummy_vision_camera.py create mode 100644 csaxs_bec/devices/vision/sample_vision_monitor.py create mode 100644 tests/tests_devices/test_dummy_vision_camera.py create mode 100644 tests/tests_devices/test_sample_vision_monitor.py diff --git a/csaxs_bec/bec_ipython_client/plugins/omny/AI_docs/MACHINE_VISION_AUTOCONFIRM_PLAN.md b/csaxs_bec/bec_ipython_client/plugins/omny/AI_docs/MACHINE_VISION_AUTOCONFIRM_PLAN.md index 411d6f72..3e81c9d2 100644 --- a/csaxs_bec/bec_ipython_client/plugins/omny/AI_docs/MACHINE_VISION_AUTOCONFIRM_PLAN.md +++ b/csaxs_bec/bec_ipython_client/plugins/omny/AI_docs/MACHINE_VISION_AUTOCONFIRM_PLAN.md @@ -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 diff --git a/csaxs_bec/bec_ipython_client/plugins/omny/AI_docs/VISION_PIPELINE_TESTING_HOWTO.md b/csaxs_bec/bec_ipython_client/plugins/omny/AI_docs/VISION_PIPELINE_TESTING_HOWTO.md new file mode 100644 index 00000000..4afd3221 --- /dev/null +++ b/csaxs_bec/bec_ipython_client/plugins/omny/AI_docs/VISION_PIPELINE_TESTING_HOWTO.md @@ -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. diff --git a/csaxs_bec/device_configs/simulated_omny/vision_test.yaml b/csaxs_bec/device_configs/simulated_omny/vision_test.yaml new file mode 100644 index 00000000..e7384935 --- /dev/null +++ b/csaxs_bec/device_configs/simulated_omny/vision_test.yaml @@ -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("/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 diff --git a/csaxs_bec/devices/vision/dummy_vision_camera.py b/csaxs_bec/devices/vision/dummy_vision_camera.py new file mode 100644 index 00000000..b61eea1a --- /dev/null +++ b/csaxs_bec/devices/vision/dummy_vision_camera.py @@ -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)) diff --git a/csaxs_bec/devices/vision/sample_vision_monitor.py b/csaxs_bec/devices/vision/sample_vision_monitor.py new file mode 100644 index 00000000..a50b4c0c --- /dev/null +++ b/csaxs_bec/devices/vision/sample_vision_monitor.py @@ -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, + } diff --git a/csaxs_bec/devices/vision/vision_interface.py b/csaxs_bec/devices/vision/vision_interface.py index 51268d13..bf8bb3e2 100644 --- a/csaxs_bec/devices/vision/vision_interface.py +++ b/csaxs_bec/devices/vision/vision_interface.py @@ -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 diff --git a/csaxs_bec/devices/vision/vision_toolkit.py b/csaxs_bec/devices/vision/vision_toolkit.py index e87ba025..a8a11d21 100644 --- a/csaxs_bec/devices/vision/vision_toolkit.py +++ b/csaxs_bec/devices/vision/vision_toolkit.py @@ -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. diff --git a/tests/tests_devices/test_dummy_vision_camera.py b/tests/tests_devices/test_dummy_vision_camera.py new file mode 100644 index 00000000..8d7ac12f --- /dev/null +++ b/tests/tests_devices/test_dummy_vision_camera.py @@ -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] diff --git a/tests/tests_devices/test_sample_vision_monitor.py b/tests/tests_devices/test_sample_vision_monitor.py new file mode 100644 index 00000000..48ac4fcf --- /dev/null +++ b/tests/tests_devices/test_sample_vision_monitor.py @@ -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 diff --git a/tests/tests_devices/test_vision_interface.py b/tests/tests_devices/test_vision_interface.py index 531a87ba..02284e32 100644 --- a/tests/tests_devices/test_vision_interface.py +++ b/tests/tests_devices/test_vision_interface.py @@ -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) diff --git a/tests/tests_devices/test_vision_toolkit.py b/tests/tests_devices/test_vision_toolkit.py index 564e8f19..3582ef5c 100644 --- a/tests/tests_devices/test_vision_toolkit.py +++ b/tests/tests_devices/test_vision_toolkit.py @@ -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():