diff --git a/csaxs_bec/devices/vision/vision_interface.py b/csaxs_bec/devices/vision/vision_interface.py index bf8bb3e2..3240958b 100644 --- a/csaxs_bec/devices/vision/vision_interface.py +++ b/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: diff --git a/csaxs_bec/devices/vision/vision_toolkit.py b/csaxs_bec/devices/vision/vision_toolkit.py index f1ed3be5..45054f50 100644 --- a/csaxs_bec/devices/vision/vision_toolkit.py +++ b/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. diff --git a/tests/tests_devices/test_vision_interface.py b/tests/tests_devices/test_vision_interface.py index 02284e32..97d3fec7 100644 --- a/tests/tests_devices/test_vision_interface.py +++ b/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) diff --git a/tests/tests_devices/test_vision_toolkit.py b/tests/tests_devices/test_vision_toolkit.py index 47bd6a46..ea1c0900 100644 --- a/tests/tests_devices/test_vision_toolkit.py +++ b/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