feat(devices): camera integration

xil_bec/device_configs/x09lb_basic.yaml

@@ -45,6 +45,24 @@ bm1_try:
     - optics
     - beam_monitor
 
+###################################
+##      Beam Monitor Cameras     ##
+###################################
+
+bm1_cam:
+  readoutPriority: async
+  description: Beam Monitor 1 camera
+  deviceClass: xil_bec.devices.cameras.beam_monitor_cam.BeamMonitorCam
+  deviceConfig:
+    prefix: "X09LB-OP-BM1CAM:"
+  onFailure: retry
+  enabled: true
+  softwareTrigger: false
+  deviceTags:
+    - optics
+    - beam_monitor
+    - camera
+
 ###################################
 ##         Diode Readout         ##
 ###################################

xil_bec/devices/cameras/__init__.py

@@ -0,0 +1,2 @@
+from xil_bec.devices.cameras.beam_monitor_cam import BeamMonitorCam
+from xil_bec.devices.cameras.xil_base_cam import XILBaseCamera

xil_bec/devices/cameras/beam_monitor_cam.py

@@ -0,0 +1,49 @@
+"""Beam monitor cameras at XIL, served by the X09LB-CPCL-BMCAMS IOC."""
+
+from __future__ import annotations
+
+from ophyd import ADBase
+from ophyd import ADComponent as ADCpt
+from ophyd import Component as Cpt
+from ophyd_devices import EpicsSignalRO
+from ophyd_devices.devices.areadetector.cam import CamBase
+from ophyd_devices.devices.areadetector.plugins import ImagePlugin_V35
+
+from xil_bec.devices.cameras.xil_base_cam import XILBaseCamera
+
+
+class BeamMonitorImagePlugin(ImagePlugin_V35):
+    """Image plugin without ophyd's derived shaped_image signal.
+
+    shaped_image reshapes ArrayData on every monitor update and raises whenever the size
+    readbacks are still zero — which is the normal state of an idle camera, so it fills the
+    device server log with callback tracebacks. XILBaseCamera reshapes the image itself.
+    """
+
+    shaped_image = None
+
+
+class BeamMonitorCamBase(ADBase):
+    """areaDetector layout of the beam monitor cameras.
+
+    Uses the driver-independent CamBase: it covers the standard cam1 parameters every
+    areaDetector driver provides. Swap it for the matching driver class (e.g.
+    AravisDetectorCam, ProsilicaDetectorCam) to also expose that driver's own PVs.
+    """
+
+    cam_detector_state_string = Cpt(EpicsSignalRO, suffix="cam1:DetectorState_RBV", string=True)
+
+    _default_configuration_attrs = [
+        "cam1.acquire_time",
+        "cam1.detector_state",
+        "cam_detector_state_string",
+        "cam1.gain",
+        "cam1.model",
+    ]
+
+    cam1 = ADCpt(CamBase, "cam1:")
+    image1 = ADCpt(BeamMonitorImagePlugin, "image1:")
+
+
+class BeamMonitorCam(XILBaseCamera, BeamMonitorCamBase):
+    """Beam monitor camera at XIL. IOC prefix: X09LB-OP-BM1CAM:"""

xil_bec/devices/cameras/xil_base_cam.py

@@ -0,0 +1,130 @@
+"""Base class for camera integration at XIL."""
+
+from __future__ import annotations
+
+import threading
+from typing import TYPE_CHECKING
+
+import numpy as np
+from bec_lib.logger import bec_logger
+from ophyd import Component as Cpt
+from ophyd_devices import DeviceStatus, PreviewSignal, PSIDeviceBase, StatusBase
+from typeguard import typechecked
+
+if TYPE_CHECKING:  # pragma: no cover
+    from bec_lib.devicemanager import ScanInfo
+    from ophyd_devices.devices.areadetector.plugins import ImagePlugin_V35
+
+logger = bec_logger.logger
+
+
+class XILBaseCamera(PSIDeviceBase):
+    """Base class for XIL cameras.
+
+    Streams the areaDetector image plugin to BEC as a preview signal, so the camera shows
+    up live in the GUI without being part of the scan data.
+    """
+
+    USER_ACCESS = ["live_mode"]
+
+    preview = Cpt(PreviewSignal, name="preview", ndim=2, doc="Preview signal for the camera.")
+
+    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.image1: "ImagePlugin_V35"
+        self._update_frequency = 1  # Hz
+        self._live_mode = False
+        self._live_mode_event = None
+        self._task_status = None
+
+    @property
+    def live_mode(self) -> bool:
+        """Whether the camera is currently streaming previews to BEC."""
+        return self._live_mode
+
+    @live_mode.setter
+    @typechecked
+    def live_mode(self, value: bool) -> None:
+        """Enable or disable the preview stream.
+
+        Args:
+            value (bool): True to enable live mode, False to disable.
+        """
+        if value == self._live_mode:
+            return
+        self._live_mode = value
+        if value:
+            self._start_live_mode()
+        else:
+            self._stop_live_mode()
+
+    def _start_live_mode(self) -> None:
+        """Start the preview stream, replacing a stream that is already running."""
+        if self._live_mode_event is not None:
+            self._live_mode_event.set()
+            self._live_mode_event = None
+            if self._task_status is not None:
+                self.task_handler.kill_task(task_status=self._task_status)
+                self._task_status = None
+
+        self._live_mode_event = threading.Event()
+        self._task_status = self.task_handler.submit_task(task=self.emit_to_bec)
+
+    def _stop_live_mode(self) -> None:
+        """Stop the preview stream."""
+        if self._live_mode_event is not None:
+            self._live_mode_event.set()
+            self._live_mode_event = None
+
+    def _reshape_image(self, value, width: int, height: int) -> np.ndarray | None:
+        """Reshape the flat ArrayData waveform into a 2D image.
+
+        The waveform is allocated for the largest supported frame, so it can be longer than
+        the current image; anything beyond width*height is padding. A waveform that is too
+        short means the plugin and the size readbacks disagree — usually a colour image on a
+        camera configured as mono — and cannot be shown as a 2D preview.
+        """
+        if value is None or not width or not height:
+            return None
+        data = np.asarray(value)
+        expected = width * height
+        if data.size < expected:
+            logger.warning(
+                f"{self.name}: image1 returned {data.size} points for a {width}x{height} frame; "
+                "check the camera's ColorMode."
+            )
+            return None
+        return data[:expected].reshape(height, width)
+
+    def emit_to_bec(self) -> None:
+        """Push images to BEC until live mode is stopped."""
+        while not self._live_mode_event.wait(1 / self._update_frequency):
+            image = self._reshape_image(
+                self.image1.array_data.get(),
+                self.image1.array_size.width.get(),
+                self.image1.array_size.height.get(),
+            )
+            if image is None:
+                continue
+            self.preview.put(image)
+
+    ########################################
+    #  Beamline Specific Implementations   #
+    ########################################
+
+    def on_init(self) -> None:
+        """Called when the device is initialized, before any signal is connected."""
+
+    def on_connected(self) -> None:
+        """Called once all signals are connected."""
+        self.live_mode = True
+
+    def on_stage(self) -> DeviceStatus | StatusBase | None:
+        """Called while staging the device."""
+
+    def on_unstage(self) -> DeviceStatus | StatusBase | None:
+        """Called while unstaging the device."""
+
+    def on_stop(self) -> None:
+        """Called when the device is stopped."""
+        self.live_mode = False

xil_bec/scans/__init__.py

@@ -0,0 +1 @@
+from .xil_template_scan import XilTemplateScan