feat(sim): simulated camera with negative values

ophyd_devices/__init__.py

@@ -1,5 +1,5 @@
 from .devices.sls_devices import SLSInfo, SLSOperatorMessages
-from .sim.sim_camera import SimCamera
+from .sim.sim_camera import SimCamera, SimNegativeCamera
 from .sim.sim_monitor import SimMonitor, SimMonitorAsync
 
 SynAxisMonitor = SimMonitor

ophyd_devices/sim/__init__.py

@@ -1,4 +1,4 @@
-from .sim_camera import SimCamera
+from .sim_camera import SimCamera, SimNegativeCamera
 from .sim_flyer import SimFlyer
 
 SynFlyer = SimFlyer

ophyd_devices/sim/sim_camera.py

@@ -6,7 +6,7 @@ from ophyd import Component as Cpt
 from ophyd import Device, Kind, StatusBase
 
 from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
-from ophyd_devices.sim.sim_data import SimulatedDataCamera
+from ophyd_devices.sim.sim_data import SimulatedDataCamera, SimulatedDataNegativeCamera
 from ophyd_devices.sim.sim_signals import ReadOnlySignal, SetableSignal
 from ophyd_devices.sim.sim_utils import H5Writer
 from ophyd_devices.utils.bec_signals import FileEventSignal, PreviewSignal
@@ -154,3 +154,18 @@ class SimCamera(PSIDeviceBase, SimCameraControl):
         """Stop the camera acquisition."""
         self.task_handler.shutdown()
         self.on_unstage()
+
+
+class SimNegativeCamera(SimCamera):
+    """A simulated 2D camera that emits signed images with negative pixel values."""
+
+    sim_cls = SimulatedDataNegativeCamera
+    BIT_DEPTH = np.int16
+
+    image = Cpt(
+        ReadOnlySignal,
+        name="image",
+        value=np.empty(SimCameraControl.SHAPE, dtype=BIT_DEPTH),
+        compute_readback=True,
+        kind=Kind.omitted,
+    )

ophyd_devices/sim/sim_data.py

@@ -72,6 +72,12 @@ DEFAULT_PARAMS_HOT_PIXEL = {
     "hot_pixel_values": np.array([1e3, 1e4, 1e3]),
 }
 
+DEFAULT_PARAMS_NEGATIVE_PIXEL = {
+    "negative_pixel_count": 20,
+    "negative_pixel_min": -100,
+    "negative_pixel_max": -1,
+}
+
 
 def _safeint(val: float) -> int:
     if isnan(val):
@@ -825,3 +831,36 @@ class SimulatedDataCamera(SimulatedDataBase):
                     if v[coord[0], coord[1]] / maximum > 0.5:
                         v[coord[0], coord[1]] = value
         return v
+
+
+class SimulatedDataNegativeCamera(SimulatedDataCamera):
+    """Simulated 2D camera data with signed negative pixel values."""
+
+    def _get_additional_params(self) -> None:
+        params = super()._get_additional_params()
+        params.update(deepcopy(DEFAULT_PARAMS_NEGATIVE_PIXEL))
+        return params
+
+    def compute_sim_state(self, signal_name: str, compute_readback: bool) -> None:
+        super().compute_sim_state(signal_name=signal_name, compute_readback=compute_readback)
+        value = np.asarray(self.sim_state[signal_name]["value"], dtype=self.bit_depth)
+        value = self._add_negative_pixels(value)
+        self.update_sim_state(signal_name, value)
+
+    def _add_negative_pixels(self, value: np.ndarray) -> np.ndarray:
+        pixel_count = int(self.params["negative_pixel_count"])
+        if pixel_count <= 0 or value.size == 0:
+            return value
+
+        negative_min = int(self.params["negative_pixel_min"])
+        negative_max = int(self.params["negative_pixel_max"])
+        if negative_min > negative_max:
+            raise SimulatedDataException(
+                "negative_pixel_min must be smaller than or equal to negative_pixel_max."
+            )
+
+        indices = np.random.choice(value.size, size=min(pixel_count, value.size), replace=False)
+        negative_values = np.random.randint(negative_min, negative_max + 1, size=len(indices))
+        flat_value = value.reshape(-1)
+        flat_value[indices] = negative_values.astype(self.bit_depth)
+        return value

tests/test_simulation.py

@@ -23,7 +23,7 @@ from ophyd_devices.interfaces.protocols.bec_protocols import (
     BECPositionerProtocol,
     BECSignalProtocol,
 )
-from ophyd_devices.sim.sim_camera import SimCamera
+from ophyd_devices.sim.sim_camera import SimCamera, SimNegativeCamera
 from ophyd_devices.sim.sim_data import _safeint
 from ophyd_devices.sim.sim_flyer import SimFlyer
 from ophyd_devices.sim.sim_frameworks.h5_image_replay_proxy import H5ImageReplayProxy
@@ -359,6 +359,16 @@ def test_camera_readback(camera, amplitude, noise_multiplier):
         assert (camera.image.get() <= (amplitude + noise_multiplier + 1)).all()
 
 
+def test_negative_camera_readback_contains_negative_pixels():
+    """Test that SimNegativeCamera emits signed images with negative values."""
+    camera = SimNegativeCamera(name="eiger_negative", device_manager=DMMock())
+    image = camera.image.get()
+
+    assert image.dtype == np.int16
+    assert image.shape == camera.SHAPE
+    assert (image < 0).sum() == camera.sim.params["negative_pixel_count"]
+
+
 def test_positioner_move(positioner):
     """Test the move method of SimPositioner."""
     positioner.move(0).wait()