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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018Mupip8x7Hean6E3FFvpbj

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

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

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)