feat(sim): rotating structured pattern for simulated X-ray-eye camera
The simulated cam_xeye (SimIDSCamera/_SimIDSBackend) served a single static Gaussian blob, cached at construction -- rotating lsamrot in a simulated session produced no visible motion, so the smear/composite rotation-center calibration aid had nothing to smear into an arc. Add an opt-in sim_rotation_coupling config (mirrors the existing sim_coarse_coupling idiom in sim_lamni.py's measured_positions()): when set, _SimIDSBackend regenerates a fresh frame on every get_image_data() call with an eccentric blob (make_rotating_test_pattern) tracking the coupled axis' live simulated Galil position instead of serving the cached static frame. Wired into LamNI's cam_xeye sim config only (host/port/axis matching lsamrot's own sim config); flomni's cam_xeye and any other SimIDSCamera user are unaffected since the new params default to None (byte-identical output to before, confirmed by test). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -324,6 +324,15 @@ cam_xeye:
|
||||
transpose: false
|
||||
force_monochrome: true
|
||||
m_n_colormode: 1
|
||||
# Makes the synthetic frame track lsamrot's live simulated angle (an
|
||||
# eccentric blob orbiting image center) instead of a static blob, so the
|
||||
# smear/composite rotation-center calibration aid can be exercised in
|
||||
# sim. host/port/axis match lsamrot's own sim config above.
|
||||
sim_rotation_coupling:
|
||||
galil_host: mpc2680.psi.ch
|
||||
galil_port: 8081
|
||||
axis: C
|
||||
sim_feature_radius_px: 220
|
||||
enabled: true
|
||||
onFailure: buffer
|
||||
readOnly: false
|
||||
|
||||
@@ -21,6 +21,8 @@ from bec_lib.logger import bec_logger
|
||||
|
||||
from csaxs_bec.devices.ids_cameras.ids_camera import IDSCamera
|
||||
from csaxs_bec.devices.omny.webcam_viewer import WebcamViewer
|
||||
from csaxs_bec.devices.sim.sim_galil import SimGalilState
|
||||
from csaxs_bec.devices.sim.sim_socket import SimStateRegistry
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
@@ -40,6 +42,43 @@ def make_test_pattern(height: int, width: int, rgb: bool) -> np.ndarray:
|
||||
return frame
|
||||
|
||||
|
||||
def make_rotating_test_pattern(
|
||||
height: int,
|
||||
width: int,
|
||||
rgb: bool,
|
||||
angle_deg: float,
|
||||
radius_px: float,
|
||||
sigma: float | None = None,
|
||||
amplitude: float = 180.0,
|
||||
) -> np.ndarray:
|
||||
"""Create a synthetic camera frame (uint8) with an eccentric gaussian blob
|
||||
orbiting the image center, for testing rotation-dependent features (e.g.
|
||||
the smear/composite rotation-center calibration aid) in simulation.
|
||||
|
||||
angle_deg follows the standard math convention (0 = +x/right, increasing
|
||||
counter-clockwise); since image rows grow downward, the y term is negated
|
||||
to keep that sense on screen. The fixed crosshair marks image center (the
|
||||
FZP/beam reference), distinct from the orbiting feature -- unlike
|
||||
make_test_pattern, this frame is meant to be regenerated fresh per call
|
||||
as the angle changes, not cached.
|
||||
"""
|
||||
yy, xx = np.mgrid[0:height, 0:width]
|
||||
cy, cx = height / 2.0, width / 2.0
|
||||
if sigma is None:
|
||||
sigma = min(height, width) / 12.0
|
||||
angle_rad = np.deg2rad(angle_deg)
|
||||
fy = cy - radius_px * np.sin(angle_rad)
|
||||
fx = cx + radius_px * np.cos(angle_rad)
|
||||
blob = amplitude * np.exp(-(((yy - fy) ** 2) + ((xx - fx) ** 2)) / (2.0 * sigma**2))
|
||||
frame = 20.0 + blob
|
||||
frame[int(cy) - 1 : int(cy) + 2, :] = 230.0 # horizontal crosshair line
|
||||
frame[:, int(cx) - 1 : int(cx) + 2] = 230.0 # vertical crosshair line
|
||||
frame = frame.clip(0, 255).astype(np.uint8)
|
||||
if rgb:
|
||||
frame = np.repeat(frame[:, :, np.newaxis], 3, axis=2)
|
||||
return frame
|
||||
|
||||
|
||||
def add_frame_noise(frame: np.ndarray, noise_std: float) -> np.ndarray:
|
||||
"""Return a copy of the frame with gaussian noise, clipped to uint8."""
|
||||
if noise_std <= 0:
|
||||
@@ -90,15 +129,42 @@ class _SimIDSSensor:
|
||||
|
||||
|
||||
class _SimIDSBackend:
|
||||
"""Drop-in replacement for `base_integration.camera.Camera` serving synthetic frames."""
|
||||
"""Drop-in replacement for `base_integration.camera.Camera` serving synthetic frames.
|
||||
|
||||
def __init__(self, width: int, height: int, rgb: bool, noise_std: float = 0.0):
|
||||
By default serves a single static frame (built once, cached forever), same
|
||||
as always. If `rotation_coupling` is given (a dict with `galil_host`,
|
||||
`galil_port`, `axis` -- same shape/idiom as SimRtLamniState's
|
||||
coarse_coupling), a fresh frame is generated on every get_image_data()
|
||||
call with an eccentric feature tracking that axis' live simulated angle,
|
||||
for testing rotation-dependent features (e.g. the smear/composite
|
||||
rotation-center calibration aid) in simulation.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
width: int,
|
||||
height: int,
|
||||
rgb: bool,
|
||||
noise_std: float = 0.0,
|
||||
rotation_coupling: dict | None = None,
|
||||
feature_radius_px: float | None = None,
|
||||
):
|
||||
self.cam = _SimIDSSensor(width, height)
|
||||
self.force_monochrome = False
|
||||
self._connected = False
|
||||
self._rgb = rgb
|
||||
self._noise_std = float(noise_std)
|
||||
self._frame = make_test_pattern(height, width, rgb=rgb)
|
||||
self._width = width
|
||||
self._height = height
|
||||
self._rotation_coupling = rotation_coupling
|
||||
self._feature_radius_px = (
|
||||
float(feature_radius_px)
|
||||
if feature_radius_px is not None
|
||||
else min(width, height) / 4.0
|
||||
)
|
||||
self._frame = None if rotation_coupling is not None else make_test_pattern(
|
||||
height, width, rgb=rgb
|
||||
)
|
||||
|
||||
def on_connect(self):
|
||||
self._connected = True
|
||||
@@ -106,8 +172,26 @@ class _SimIDSBackend:
|
||||
def on_disconnect(self):
|
||||
self._connected = False
|
||||
|
||||
def _current_angle_deg(self) -> float:
|
||||
coupling = self._rotation_coupling
|
||||
galil = SimStateRegistry.get(
|
||||
SimGalilState, coupling["galil_host"], coupling["galil_port"]
|
||||
)
|
||||
axis_num = ord(str(coupling["axis"]).lower()) - 97
|
||||
ax = galil.axis(axis_num)
|
||||
return ax.position() / ax.stppermm
|
||||
|
||||
def get_image_data(self) -> np.ndarray:
|
||||
frame = self._frame
|
||||
if self._rotation_coupling is None:
|
||||
frame = self._frame
|
||||
else:
|
||||
frame = make_rotating_test_pattern(
|
||||
self._height,
|
||||
self._width,
|
||||
rgb=self._rgb,
|
||||
angle_deg=self._current_angle_deg(),
|
||||
radius_px=self._feature_radius_px,
|
||||
)
|
||||
if self.force_monochrome and frame.ndim == 3:
|
||||
frame = frame[:, :, 0]
|
||||
return add_frame_noise(frame, self._noise_std)
|
||||
@@ -136,6 +220,8 @@ class SimIDSCamera(IDSCamera):
|
||||
channels: int | None = None, # legacy OMNY config keys, accepted for compatibility
|
||||
sim_shape=(1024, 1280),
|
||||
sim_noise_std=0.0,
|
||||
sim_rotation_coupling: dict | None = None,
|
||||
sim_feature_radius_px: float | None = None,
|
||||
**kwargs,
|
||||
):
|
||||
if camera_ID is not None and not camera_id:
|
||||
@@ -154,6 +240,11 @@ class SimIDSCamera(IDSCamera):
|
||||
**kwargs,
|
||||
)
|
||||
self.cam = _SimIDSBackend(
|
||||
int(sim_shape[1]), int(sim_shape[0]), rgb=not force_monochrome, noise_std=sim_noise_std
|
||||
int(sim_shape[1]),
|
||||
int(sim_shape[0]),
|
||||
rgb=not force_monochrome,
|
||||
noise_std=sim_noise_std,
|
||||
rotation_coupling=sim_rotation_coupling,
|
||||
feature_radius_px=sim_feature_radius_px,
|
||||
)
|
||||
self.cam.force_monochrome = self._force_monochrome
|
||||
|
||||
@@ -0,0 +1,111 @@
|
||||
"""Unit tests for the simulated IDS camera frame source (_SimIDSBackend /
|
||||
make_rotating_test_pattern / SimIDSCamera's optional rotation coupling)."""
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from csaxs_bec.devices.sim.sim_cameras import (
|
||||
SimIDSCamera,
|
||||
_SimIDSBackend,
|
||||
make_rotating_test_pattern,
|
||||
make_test_pattern,
|
||||
)
|
||||
from csaxs_bec.devices.sim.sim_galil import SimGalilState
|
||||
from csaxs_bec.devices.sim.sim_socket import SimStateRegistry
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_sim_state():
|
||||
"""Avoid cross-test leakage of shared (host, port)-keyed simulation state."""
|
||||
SimStateRegistry.reset()
|
||||
yield
|
||||
SimStateRegistry.reset()
|
||||
|
||||
|
||||
def test_sim_ids_backend_default_matches_static_pattern():
|
||||
"""No rotation_coupling (the default) -> byte-identical to today's static frame.
|
||||
|
||||
Regression guard: this is what keeps flomni's cam_xeye and other existing
|
||||
SimIDSCamera users completely unaffected by the new opt-in feature.
|
||||
"""
|
||||
backend = _SimIDSBackend(width=64, height=48, rgb=True)
|
||||
expected = make_test_pattern(48, 64, rgb=True)
|
||||
assert np.array_equal(backend.get_image_data(), expected)
|
||||
# calling it again returns the exact same cached frame, not a fresh one
|
||||
assert np.array_equal(backend.get_image_data(), expected)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("angle_deg", [30.0, 120.0, 210.0, 300.0])
|
||||
def test_make_rotating_test_pattern_blob_position(angle_deg):
|
||||
height, width = 200, 300
|
||||
radius_px = 60.0
|
||||
frame = make_rotating_test_pattern(
|
||||
height, width, rgb=False, angle_deg=angle_deg, radius_px=radius_px
|
||||
)
|
||||
|
||||
cy, cx = height / 2.0, width / 2.0
|
||||
angle_rad = np.deg2rad(angle_deg)
|
||||
expected_y = cy - radius_px * np.sin(angle_rad)
|
||||
expected_x = cx + radius_px * np.cos(angle_rad)
|
||||
|
||||
# mask out the fixed center crosshair so it can't win the argmax
|
||||
masked = frame.astype(np.int32)
|
||||
masked[int(cy) - 2 : int(cy) + 3, :] = 0
|
||||
masked[:, int(cx) - 2 : int(cx) + 3] = 0
|
||||
py, px = np.unravel_index(np.argmax(masked), masked.shape)
|
||||
|
||||
assert abs(py - expected_y) <= 2
|
||||
assert abs(px - expected_x) <= 2
|
||||
|
||||
|
||||
def test_sim_ids_backend_rotation_coupling_tracks_live_angle():
|
||||
host, port, axis_letter = "test-galil.example", 9000, "C"
|
||||
galil = SimStateRegistry.get(SimGalilState, host, port)
|
||||
ax = galil.axis(ord(axis_letter.lower()) - 97)
|
||||
ax.stppermm = 1000.0
|
||||
ax.pos_steps = 0.0 # 0 deg
|
||||
|
||||
coupling = {"galil_host": host, "galil_port": port, "axis": axis_letter}
|
||||
backend = _SimIDSBackend(
|
||||
width=200, height=200, rgb=False, rotation_coupling=coupling, feature_radius_px=50.0
|
||||
)
|
||||
|
||||
frame0 = backend.get_image_data()
|
||||
assert np.array_equal(
|
||||
frame0, make_rotating_test_pattern(200, 200, rgb=False, angle_deg=0.0, radius_px=50.0)
|
||||
)
|
||||
|
||||
# move the simulated axis and confirm the frame is regenerated fresh, not cached
|
||||
ax.pos_steps = 90.0 * ax.stppermm
|
||||
frame90 = backend.get_image_data()
|
||||
assert np.array_equal(
|
||||
frame90, make_rotating_test_pattern(200, 200, rgb=False, angle_deg=90.0, radius_px=50.0)
|
||||
)
|
||||
assert not np.array_equal(frame0, frame90)
|
||||
|
||||
|
||||
def test_sim_ids_camera_threads_rotation_coupling_through():
|
||||
"""SimIDSCamera's sim_rotation_coupling/sim_feature_radius_px reach _SimIDSBackend."""
|
||||
host, port, axis_letter = "test-galil-2.example", 9100, "C"
|
||||
galil = SimStateRegistry.get(SimGalilState, host, port)
|
||||
ax = galil.axis(ord(axis_letter.lower()) - 97)
|
||||
ax.stppermm = 1000.0
|
||||
ax.pos_steps = 0.0
|
||||
|
||||
camera = SimIDSCamera(
|
||||
name="test_sim_cam",
|
||||
camera_id=1,
|
||||
prefix="test:",
|
||||
scan_info=None,
|
||||
m_n_colormode=1,
|
||||
bits_per_pixel=24,
|
||||
live_mode=False,
|
||||
sim_shape=(64, 64),
|
||||
sim_rotation_coupling={"galil_host": host, "galil_port": port, "axis": axis_letter},
|
||||
sim_feature_radius_px=20.0,
|
||||
)
|
||||
|
||||
frame0 = camera.cam.get_image_data()
|
||||
ax.pos_steps = 180.0 * ax.stppermm
|
||||
frame180 = camera.cam.get_image_data()
|
||||
assert not np.array_equal(frame0, frame180)
|
||||
Reference in New Issue
Block a user