From 5b2b098158aa6ce6b38b0eb2ce3653c43f249dce Mon Sep 17 00:00:00 2001 From: x01dc Date: Sun, 6 Sep 2026 13:37:02 +0200 Subject: [PATCH] feat(vision): add Layer-1 VisionInterface device Generic camera-in/processed-image-out device: wraps one or more camera devices by configured name (device_manager.devices.get(), resolved lazily per call -- the ddg_1.py dependency-lookup pattern generalized to a configurable name), requiring only the shared get_last_image() contract -- proven against a plain FakeCamera stand-in in tests, not IDSCamera, to keep it honestly camera-agnostic. Runtime-selectable mode (raw/grayscale/edges/clahe, built on vision_toolkit) publishes through its own diagnostic_image PreviewSignal, decoupled from whichever camera backs it. 12 new tests; full suite (334 tests) still green. Also fixes a stale "rpc_access" attribute-name reference in the plan doc (the actual attribute is USER_ACCESS) and adds an implementation-status section. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_018Mupip8x7Hean6E3FFvpbj --- .../MACHINE_VISION_AUTOCONFIRM_PLAN.md | 18 ++- csaxs_bec/devices/vision/vision_interface.py | 153 ++++++++++++++++++ tests/tests_devices/test_vision_interface.py | 120 ++++++++++++++ 3 files changed, 289 insertions(+), 2 deletions(-) create mode 100644 csaxs_bec/devices/vision/vision_interface.py create mode 100644 tests/tests_devices/test_vision_interface.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 b73cb7a8..411d6f72 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 @@ -5,6 +5,20 @@ can pick this up without re-deriving it. Companion docs: `OPEN_ISSUES.md` (the l this plan grows out of — see its "full gripper sample-transfer chain... untested" item) and `SAMPLE_STORAGE_TEST_CHECKLIST.md`. +## Implementation status (branch `feat/omny_vision`) + +- **Layer 0 (`csaxs_bec/devices/vision/vision_toolkit.py`) — done, tested (16 tests).** Grayscale, + CLAHE, Canny edges, ECC registration (never raises on non-convergence), contour + extraction/scoring, multi-reference best-match, contour overlay drawing. +- **Layer 1 (`csaxs_bec/devices/vision/vision_interface.py`, `VisionInterface`) — done, tested (12 + tests).** Config-driven `camera_names`, lazy per-call resolution via + `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.** + ## Motivation Mirko's starting question: is there a Python package for comparing a live gripper/pin camera image @@ -58,9 +72,9 @@ post-hoc audit log today, with no comparison logic reading it back. - **`get_last_image() -> np.ndarray`** is already a shared, RPC-exposed contract across camera device classes, not an `IDSCamera` idiosyncrasy: both `IDSCamera` - (`csaxs_bec/devices/ids_cameras/ids_camera.py:234`, `rpc_access` line 72) and + (`csaxs_bec/devices/ids_cameras/ids_camera.py:234`, `USER_ACCESS` line 62) and `AlliedVisionAravisCamera` (`csaxs_bec/devices/allied_vision_cameras/allied_vision_aravis_camera.py:216`, - `rpc_access` line 62) implement it identically, and client-side code already calls it directly + `USER_ACCESS` line 62) implement it identically, and client-side code already calls it directly through the device-server RPC boundary — `LamNI/x_ray_eye_align.py:202,821` and `flomni/x_ray_eye_align.py:127` do exactly this (`dev.cam_xeye.get_last_image()`) regardless of which concrete camera class backs `cam_xeye` at a given beamline. This is the natural minimal diff --git a/csaxs_bec/devices/vision/vision_interface.py b/csaxs_bec/devices/vision/vision_interface.py new file mode 100644 index 00000000..51268d13 --- /dev/null +++ b/csaxs_bec/devices/vision/vision_interface.py @@ -0,0 +1,153 @@ +"""Generic camera-in, processed-image-out device. + +Layer 1 of ``AI_docs/MACHINE_VISION_AUTOCONFIRM_PLAN.md``: wraps one or more source cameras, +referenced by configured name rather than by class, and republishes a processed view of the +latest frame -- raw / grayscale / edge-detected / CLAHE-normalized, using the Layer-0 +:mod:`vision_toolkit` -- through its own ``diagnostic_image`` signal, independent of the source +camera's own live-acquisition channel. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Literal + +import numpy as np +from ophyd import Component as Cpt +from ophyd import Kind, Signal +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 + +VisionMode = Literal["raw", "grayscale", "edges", "clahe"] +_VALID_MODES: tuple[VisionMode, ...] = ("raw", "grayscale", "edges", "clahe") + + +class VisionInterfaceError(Exception): + pass + + +class VisionInterface(PSIDeviceBase): + """Camera-agnostic vision interface. + + Only depends on the configured camera(s) implementing ``get_last_image() -> np.ndarray | None`` + -- the contract already shared identically by ``IDSCamera`` and ``AlliedVisionAravisCamera`` -- + so it can be pointed at any current or future camera class via config alone, with no code + change here. + """ + + diagnostic_image = Cpt( + PreviewSignal, + name="diagnostic_image", + ndim=2, + doc="Processed view of the configured camera(s)' latest frame, published by refresh().", + num_rotation_90=0, + transpose=False, + ) + mode_signal = Cpt( + Signal, + name="mode_signal", + value="raw", + doc=f"Active processing mode -- one of {_VALID_MODES}.", + kind=Kind.config, + ) + + USER_ACCESS = ["camera_names", "set_mode", "get_mode", "get_last_processed_image", "refresh"] + + def __init__( + self, + *, + name: str, + camera_names: list[str], + prefix: str = "", + scan_info: "ScanInfo | None" = None, + mode: VisionMode = "raw", + **kwargs, + ): + """Initialize the vision interface. + + Args: + name (str): Name of the device. + camera_names (list[str]): Names of the camera device(s) this instance is a view onto, + resolved through the device manager at call time (not at construction, since + device build order across a config isn't guaranteed). A list, not a single name, + so a camera *pair* (e.g. OMNY's parking-view cam200+cam203) is a natural fit; a + single-camera setup is just a length-1 list. + prefix (str): Prefix for the device. + scan_info (ScanInfo | None): Scan information for the device. + mode (VisionMode): Initial processing mode. + """ + super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs) + if not camera_names: + raise VisionInterfaceError(f"{name}: camera_names must be a non-empty list.") + self._camera_names = list(camera_names) + self.set_mode(mode) + + @property + def camera_names(self) -> list[str]: + return list(self._camera_names) + + def set_mode(self, mode: str) -> None: + if mode not in _VALID_MODES: + raise VisionInterfaceError( + f"{self.name}: unknown mode '{mode}'; valid modes are {_VALID_MODES}." + ) + self.mode_signal.put(mode) + + def get_mode(self) -> str: + return self.mode_signal.get() + + def _resolve_camera(self, camera_name: str): + if camera_name not in self._camera_names: + raise VisionInterfaceError( + f"{self.name}: '{camera_name}' is not one of this device's configured cameras" + f" {self._camera_names}." + ) + if self.device_manager is None or not hasattr(self.device_manager, "devices"): + raise VisionInterfaceError(f"{self.name}: device manager is not available.") + camera = self.device_manager.devices.get(camera_name, None) + if camera is None: + raise VisionInterfaceError( + f"{self.name}: configured camera '{camera_name}' was not found in the current" + " device config." + ) + return camera + + def _get_camera_frame(self, camera_name: str) -> np.ndarray: + camera = self._resolve_camera(camera_name) + image = camera.get_last_image() + if image is None: + raise VisionInterfaceError(f"{self.name}: camera '{camera_name}' has no image yet.") + return image + + def _process(self, image: np.ndarray, mode: str) -> np.ndarray: + if mode == "raw": + return image + if mode == "grayscale": + return vt.to_grayscale(image) + if mode == "edges": + return vt.detect_edges(image) + if mode == "clahe": + return vt.apply_clahe(image) + raise VisionInterfaceError(f"{self.name}: unknown mode '{mode}'.") + + def get_last_processed_image(self, camera_name: str | None = None) -> np.ndarray: + """Pull the latest frame from ``camera_name`` (default: the first configured camera) and + apply the active mode's transform, without publishing it anywhere.""" + resolved_name = camera_name if camera_name is not None else self._camera_names[0] + frame = self._get_camera_frame(resolved_name) + return self._process(frame, self.get_mode()) + + def refresh(self, camera_name: str | None = None) -> np.ndarray: + """Pull, process, and publish the latest frame through ``diagnostic_image``. + + Returns the published array. Only ever republishes one camera's frame at a time -- a + composite covering every configured camera at once is a possible future extension, not + needed by anything using this device yet. + """ + processed = self.get_last_processed_image(camera_name) + self.diagnostic_image.put(processed) + return processed diff --git a/tests/tests_devices/test_vision_interface.py b/tests/tests_devices/test_vision_interface.py new file mode 100644 index 00000000..531a87ba --- /dev/null +++ b/tests/tests_devices/test_vision_interface.py @@ -0,0 +1,120 @@ +"""Unit tests for the Layer-1 vision interface device (csaxs_bec/devices/vision/vision_interface.py).""" + +import numpy as np +import pytest +from bec_server.device_server.tests.utils import DMMock + +from csaxs_bec.devices.vision.vision_interface import VisionInterface, VisionInterfaceError + + +class FakeCamera: + """Minimal stand-in for a camera device -- only the shared get_last_image() contract, + deliberately not an IDSCamera/AlliedVisionAravisCamera subclass, to prove VisionInterface + doesn't depend on either.""" + + def __init__(self, image: np.ndarray | None): + self._image = image + + def get_last_image(self) -> np.ndarray | None: + return self._image + + +def _make_interface(camera_names=("cam_a",), images=None, **kwargs) -> VisionInterface: + dm = DMMock() + images = images or {} + for camera_name in camera_names: + dm.devices[camera_name] = FakeCamera(images.get(camera_name)) + return VisionInterface( + name="vision_test", camera_names=list(camera_names), device_manager=dm, **kwargs + ) + + +def _rgb_image(size: int = 20) -> np.ndarray: + image = np.zeros((size, size, 3), dtype=np.uint8) + image[5:15, 5:15] = 255 + return image + + +def test_construction_requires_at_least_one_camera_name(): + dm = DMMock() + with pytest.raises(VisionInterfaceError): + VisionInterface(name="vision_test", camera_names=[], device_manager=dm) + + +def test_camera_names_returns_a_copy(): + interface = _make_interface(camera_names=["cam_a", "cam_b"]) + names = interface.camera_names + names.append("cam_c") + assert interface.camera_names == ["cam_a", "cam_b"] + + +def test_default_mode_is_raw(): + interface = _make_interface() + assert interface.get_mode() == "raw" + + +def test_set_mode_rejects_unknown_mode(): + interface = _make_interface() + with pytest.raises(VisionInterfaceError): + interface.set_mode("sepia") + + +def test_get_last_processed_image_raw_mode_returns_unmodified_frame(): + image = _rgb_image() + interface = _make_interface(images={"cam_a": image}) + result = interface.get_last_processed_image() + assert np.array_equal(result, image) + + +def test_get_last_processed_image_grayscale_mode(): + image = _rgb_image() + interface = _make_interface(images={"cam_a": image}, mode="grayscale") + result = interface.get_last_processed_image() + assert result.ndim == 2 + assert result.shape == image.shape[:2] + + +def test_get_last_processed_image_edges_mode(): + image = _rgb_image() + interface = _make_interface(images={"cam_a": image}, mode="edges") + result = interface.get_last_processed_image() + assert result.ndim == 2 + assert result.sum() > 0 + + +def test_get_last_processed_image_unknown_camera_name_raises(): + interface = _make_interface(camera_names=["cam_a"], images={"cam_a": _rgb_image()}) + with pytest.raises(VisionInterfaceError): + interface.get_last_processed_image(camera_name="cam_not_configured") + + +def test_get_last_processed_image_missing_camera_in_device_manager_raises(): + dm = DMMock() # cam_a never registered + interface = VisionInterface(name="vision_test", camera_names=["cam_a"], device_manager=dm) + with pytest.raises(VisionInterfaceError): + interface.get_last_processed_image() + + +def test_get_last_processed_image_no_frame_yet_raises(): + interface = _make_interface(images={"cam_a": None}) + with pytest.raises(VisionInterfaceError): + interface.get_last_processed_image() + + +def test_refresh_publishes_to_diagnostic_image_and_returns_it(): + image = _rgb_image() + interface = _make_interface(images={"cam_a": image}, mode="grayscale") + result = interface.refresh() + assert result.ndim == 2 + published = interface.diagnostic_image.get() + assert np.array_equal(published.data, result) + + +def test_refresh_second_configured_camera_by_name(): + image_a = _rgb_image() + image_b = np.zeros((20, 20, 3), dtype=np.uint8) + interface = _make_interface( + camera_names=["cam_a", "cam_b"], images={"cam_a": image_a, "cam_b": image_b} + ) + result = interface.refresh(camera_name="cam_b") + assert np.array_equal(result, image_b)