feat(vision): add per-camera ROI cropping to VisionInterface

set_roi()/get_roi() (plus a roi constructor argument -- a single tuple for
every configured camera, or a per-camera-name mapping) crop a camera's frame
to (x, y, width, height) before any mode processing, publishing, or scoring
happens. Applied centrally in _get_camera_frame(), the one method both
get_raw_frame() and get_last_processed_image()/refresh() go through, so
SampleVisionMonitor.save_reference()/score() pick it up automatically with
no changes needed there -- both already call get_raw_frame(), never the
camera directly. An out-of-bounds ROI raises VisionInterfaceError rather
than silently clamping.

New vision_toolkit.crop_to_roi() is the underlying pure-function primitive.

Also declares device_manager explicitly in VisionInterface's own __init__,
same bug and fix as SampleVisionMonitor's previous commit -- the device
server never injected it into either class before.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01C9zgK28uPoZWJFkJWV4P1G

csaxs_bec/devices/vision/vision_interface.py

@@ -2,9 +2,9 @@
 
 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.
+latest frame -- raw / grayscale / edge-detected / CLAHE-normalized, optionally cropped first to a
+per-camera region of interest, 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
@@ -20,10 +20,11 @@ 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
+    from bec_lib.devicemanager import DeviceManagerBase, ScanInfo
 
 VisionMode = Literal["raw", "grayscale", "edges", "clahe"]
 _VALID_MODES: tuple[VisionMode, ...] = ("raw", "grayscale", "edges", "clahe")
+RoiType = tuple[int, int, int, int]
 
 
 class VisionInterfaceError(Exception):
@@ -59,6 +60,8 @@ class VisionInterface(PSIDeviceBase):
         "camera_names",
         "set_mode",
         "get_mode",
+        "set_roi",
+        "get_roi",
         "get_last_processed_image",
         "get_raw_frame",
         "publish_diagnostic_image",
@@ -72,7 +75,9 @@ class VisionInterface(PSIDeviceBase):
         camera_names: list[str],
         prefix: str = "",
         scan_info: "ScanInfo | None" = None,
+        device_manager: "DeviceManagerBase | None" = None,
         mode: VisionMode = "raw",
+        roi: "RoiType | dict[str, RoiType] | None" = None,
         **kwargs,
     ):
         """Initialize the vision interface.
@@ -86,13 +91,30 @@ class VisionInterface(PSIDeviceBase):
                 single-camera setup is just a length-1 list.
             prefix (str): Prefix for the device.
             scan_info (ScanInfo | None): Scan information for the device.
+            device_manager (DeviceManagerBase | None): Device manager, needed to resolve
+                ``camera_names`` at call time -- must be declared explicitly here (not just
+                inherited via ``**kwargs``) since the device server only injects it into classes
+                whose own ``__init__`` signature names it.
             mode (VisionMode): Initial processing mode.
+            roi (RoiType | dict[str, RoiType] | None): Region of interest to crop from each
+                camera's frame before any further processing -- see :meth:`set_roi`. Either a
+                single ``(x, y, width, height)`` applied to every configured camera, a
+                ``{camera_name: (x, y, width, height)}`` mapping for per-camera ROIs, or ``None``
+                (the default) to use each camera's full frame.
         """
-        super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs)
+        super().__init__(
+            name=name, prefix=prefix, scan_info=scan_info, device_manager=device_manager, **kwargs
+        )
         if not camera_names:
             raise VisionInterfaceError(f"{name}: camera_names must be a non-empty list.")
         self._camera_names = list(camera_names)
+        self._rois: dict[str, RoiType] = {}
         self.set_mode(mode)
+        if isinstance(roi, dict):
+            for camera_name, camera_roi in roi.items():
+                self.set_roi(camera_roi, camera_name=camera_name)
+        elif roi is not None:
+            self.set_roi(roi)
 
     @property
     def camera_names(self) -> list[str]:
@@ -108,12 +130,54 @@ class VisionInterface(PSIDeviceBase):
     def get_mode(self) -> str:
         return self.mode_signal.get()
 
-    def _resolve_camera(self, camera_name: str):
+    def _check_camera_name(self, camera_name: str) -> None:
         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}."
             )
+
+    def set_roi(self, roi: "RoiType | None", camera_name: str | None = None) -> None:
+        """Set (or clear) the region of interest cropped from a camera's frame before any further
+        processing -- affects :meth:`get_raw_frame`/:meth:`get_last_processed_image`/:meth:`refresh`,
+        and therefore also what ``SampleVisionMonitor.save_reference()``/``score()`` see, uniformly
+        (both go through this interface's frame-fetching, not the camera directly).
+
+        Args:
+            roi (RoiType | None): ``(x, y, width, height)`` in pixels, top-left origin -- matching
+                the convention OpenCV and ``bec_widgets``' image ROI widgets use
+                (``RectangularROI.get_coordinates()``), so coordinates picked interactively in a
+                GUI can be passed straight through. ``None`` clears the ROI, going back to the
+                camera's full frame. Not validated against the camera's actual resolution until the
+                next frame is pulled -- the camera may not have produced one yet at configuration
+                time.
+            camera_name (str | None): Which configured camera to set this for. ``None`` (the
+                default) applies it to every camera configured on this interface.
+        """
+        targets = [camera_name] if camera_name is not None else self._camera_names
+        for target in targets:
+            self._check_camera_name(target)
+        for target in targets:
+            if roi is None:
+                self._rois.pop(target, None)
+            else:
+                self._rois[target] = tuple(roi)
+
+    def get_roi(self, camera_name: str | None = None) -> "RoiType | dict[str, RoiType] | None":
+        """Current ROI configuration.
+
+        Args:
+            camera_name (str | None): If given, that camera's ROI (``None`` if it has none set).
+                If omitted, the full ``{camera_name: roi}`` mapping of every camera that currently
+                has one set.
+        """
+        if camera_name is None:
+            return dict(self._rois)
+        self._check_camera_name(camera_name)
+        return self._rois.get(camera_name)
+
+    def _resolve_camera(self, camera_name: str):
+        self._check_camera_name(camera_name)
         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)
@@ -129,6 +193,12 @@ class VisionInterface(PSIDeviceBase):
         image = camera.get_last_image()
         if image is None:
             raise VisionInterfaceError(f"{self.name}: camera '{camera_name}' has no image yet.")
+        roi = self._rois.get(camera_name)
+        if roi is not None:
+            try:
+                image = vt.crop_to_roi(image, roi)
+            except ValueError as exc:
+                raise VisionInterfaceError(f"{self.name}: {exc}") from exc
         return image
 
     def _process(self, image: np.ndarray, mode: str) -> np.ndarray:

csaxs_bec/devices/vision/vision_toolkit.py

@@ -115,6 +115,30 @@ def make_square_test_pattern(
     return image
 
 
+def crop_to_roi(image: np.ndarray, roi: tuple[int, int, int, int]) -> np.ndarray:
+    """Crop ``image`` to a rectangular region of interest.
+
+    Args:
+        image: Array to crop (grayscale or multi-channel; only the first two axes are cropped).
+        roi: ``(x, y, width, height)`` in pixels, top-left origin -- the same convention OpenCV
+            and ``bec_widgets``' image ROI widgets use (``RectangularROI.get_coordinates()``), so
+            coordinates picked interactively in a GUI can be passed straight through.
+
+    Raises:
+        ValueError: if width/height aren't positive, or the ROI falls even partially outside
+            ``image``'s bounds -- silently clamping would make a caller believe it's always
+            comparing the same region when the configured ROI no longer fits the camera's actual
+            frame size.
+    """
+    x, y, width, height = roi
+    if width <= 0 or height <= 0:
+        raise ValueError(f"ROI width/height must be positive, got {roi}.")
+    image_height, image_width = image.shape[:2]
+    if x < 0 or y < 0 or x + width > image_width or y + height > image_height:
+        raise ValueError(f"ROI {roi} is out of bounds for image of shape {image.shape[:2]}.")
+    return image[y : y + height, x : x + width]
+
+
 def to_grayscale(image: np.ndarray) -> np.ndarray:
     """Convert an image to single-channel grayscale.
 

tests/tests_devices/test_vision_interface.py

@@ -127,6 +127,98 @@ def test_get_raw_frame_bypasses_active_mode():
     assert np.array_equal(result, image)
 
 
+def test_get_roi_defaults_to_none_for_every_camera():
+    interface = _make_interface(camera_names=["cam_a", "cam_b"])
+    assert interface.get_roi("cam_a") is None
+    assert interface.get_roi() == {}
+
+
+def test_set_roi_crops_frames_from_get_raw_frame():
+    image = _rgb_image(size=20)
+    interface = _make_interface(images={"cam_a": image})
+    interface.set_roi((5, 5, 8, 6))
+    result = interface.get_raw_frame()
+    assert result.shape == (6, 8, 3)
+    assert np.array_equal(result, image[5:11, 5:13])
+
+
+def test_set_roi_affects_processed_image_and_refresh():
+    image = _rgb_image(size=20)
+    interface = _make_interface(images={"cam_a": image}, mode="grayscale")
+    interface.set_roi((0, 0, 10, 8))
+    result = interface.refresh()
+    assert result.shape == (8, 10)
+    published = interface.diagnostic_image.get()
+    assert np.array_equal(published.data, result)
+
+
+def test_set_roi_none_clears_it():
+    image = _rgb_image(size=20)
+    interface = _make_interface(images={"cam_a": image})
+    interface.set_roi((0, 0, 10, 8))
+    interface.set_roi(None)
+    result = interface.get_raw_frame()
+    assert result.shape == image.shape
+
+
+def test_set_roi_per_camera_only_affects_that_camera():
+    image_a = _rgb_image(size=20)
+    image_b = _rgb_image(size=20)
+    interface = _make_interface(
+        camera_names=["cam_a", "cam_b"], images={"cam_a": image_a, "cam_b": image_b}
+    )
+    interface.set_roi((0, 0, 10, 8), camera_name="cam_a")
+    assert interface.get_raw_frame("cam_a").shape == (8, 10, 3)
+    assert interface.get_raw_frame("cam_b").shape == image_b.shape
+    assert interface.get_roi() == {"cam_a": (0, 0, 10, 8)}
+
+
+def test_set_roi_without_camera_name_applies_to_all_configured_cameras():
+    image_a = _rgb_image(size=20)
+    image_b = _rgb_image(size=20)
+    interface = _make_interface(
+        camera_names=["cam_a", "cam_b"], images={"cam_a": image_a, "cam_b": image_b}
+    )
+    interface.set_roi((0, 0, 10, 8))
+    assert interface.get_raw_frame("cam_a").shape == (8, 10, 3)
+    assert interface.get_raw_frame("cam_b").shape == (8, 10, 3)
+
+
+def test_set_roi_unknown_camera_name_raises():
+    interface = _make_interface(images={"cam_a": _rgb_image()})
+    with pytest.raises(VisionInterfaceError):
+        interface.set_roi((0, 0, 5, 5), camera_name="cam_not_configured")
+
+
+def test_get_raw_frame_out_of_bounds_roi_raises_vision_interface_error():
+    interface = _make_interface(images={"cam_a": _rgb_image(size=20)})
+    interface.set_roi((15, 15, 10, 10))
+    with pytest.raises(VisionInterfaceError):
+        interface.get_raw_frame()
+
+
+def test_roi_constructor_argument_single_tuple_applies_to_all_cameras():
+    image = _rgb_image(size=20)
+    interface = _make_interface(
+        camera_names=["cam_a", "cam_b"],
+        images={"cam_a": image, "cam_b": image},
+        roi=(0, 0, 10, 8),
+    )
+    assert interface.get_raw_frame("cam_a").shape == (8, 10, 3)
+    assert interface.get_raw_frame("cam_b").shape == (8, 10, 3)
+
+
+def test_roi_constructor_argument_dict_is_per_camera():
+    image = _rgb_image(size=20)
+    interface = _make_interface(
+        camera_names=["cam_a", "cam_b"],
+        images={"cam_a": image, "cam_b": image},
+        roi={"cam_a": (0, 0, 10, 8)},
+    )
+    assert interface.get_raw_frame("cam_a").shape == (8, 10, 3)
+    assert interface.get_raw_frame("cam_b").shape == image.shape
+
+
 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)

tests/tests_devices/test_vision_toolkit.py

@@ -20,6 +20,32 @@ def test_make_square_test_pattern_shape_and_offset():
     assert not np.array_equal(centered, shifted)
 
 
+def test_crop_to_roi_returns_expected_subregion():
+    image = np.zeros((20, 20), dtype=np.uint8)
+    image[5:10, 3:8] = 42
+    cropped = vt.crop_to_roi(image, (3, 5, 5, 5))
+    assert cropped.shape == (5, 5)
+    assert np.all(cropped == 42)
+
+
+def test_crop_to_roi_preserves_channels():
+    image = _square_image(size=20, square_side=6)
+    cropped = vt.crop_to_roi(image, (0, 0, 10, 8))
+    assert cropped.shape == (8, 10, 3)
+
+
+def test_crop_to_roi_rejects_non_positive_size():
+    image = np.zeros((10, 10), dtype=np.uint8)
+    with pytest.raises(ValueError):
+        vt.crop_to_roi(image, (0, 0, 0, 5))
+
+
+def test_crop_to_roi_rejects_out_of_bounds():
+    image = np.zeros((10, 10), dtype=np.uint8)
+    with pytest.raises(ValueError):
+        vt.crop_to_roi(image, (5, 5, 10, 10))
+
+
 def test_to_grayscale_passes_through_2d():
     gray = np.zeros((10, 10), dtype=np.uint8)
     assert vt.to_grayscale(gray) is gray