bec/csaxs_bec
0
0
Fork 2
Code Issues 17 Pull Requests 2 Actions 1 Packages Projects Releases Wiki Activity

feat(LamNI): experimental continuous-rotation smear aid for rotation-center calibration

Browse Source 
Adds an EXPERIMENTAL, operator-only alternative to the extended-sample
rotation-center click: instead of judging the center from a single
instant of live rotation, accumulate a max-projection composite while
lsamrot rotates continuously through a (tunable, default 360 deg)
sweep. Off-axis features smear into circular arcs; the common center
of curvature is the rotation axis, an easier target to click than one
live frame. No automatic circle fitting -- purely a visual aid feeding
the same _collect_click mechanism the production feature already uses.

Rotation is issued non-blocking via scans.mv() (new local wrapper,
sibling of this file's existing blocking umv()) and frames are grabbed
via get_last_image() in a single-threaded loop polling ScanReport.status
-- the same thing ScanReport.wait() does internally -- so no background
thread or device-server change is needed to grab frames while the motor
is actively moving. The composite is pushed to the GUI through the
camera's existing PreviewSignal (dev.cam_xeye.image), duty-cycling
live_mode_enabled off/on around each push so the background 5 Hz live
thread can't immediately overwrite it.

New entry point: lamni.xrayeye_rotation_center_calibration_smear_experimental().
Purely additive -- does not modify find_rotation_center()/_live_sweep()
or the two production entry points.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
x01dc
2026-07-24 16:08:01 +02:00
co-authored by Claude Sonnet 5
parent f49605317b
commit d0d3602ed1
3 changed files with 394 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py
+46
View File
@@ -524,6 +524,52 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
logger.warning(f"Failed to disable XRayEye DAP parameter forwarding: {gui_exc}")
raise exc
def xrayeye_rotation_center_calibration_smear_experimental(
self, keep_shutter_open: bool = False, sweep_deg: float = 360.0
):
"""EXPERIMENTAL: extended-sample rotation-center calibration aided by
a continuous-rotation max-projection ("star-trail") composite image,
instead of a single instant of live rotation.
Off-axis features smear into circular arcs while lsamrot rotates
continuously through sweep_deg; the common center of curvature of
those arcs is the rotation axis -- an easier target to click than
one live frame. This is purely a visual aid for the same manual
click used by xrayeye_rotation_center_calibration_extended(); there
is no automatic circle fitting.
Args:
keep_shutter_open: If True the shutter is left open between steps
so the sample remains visible in live view.
sweep_deg: angular range of the continuous smear sweep. Defaults
to a full 360 deg (complete arcs), but a smaller range may
already show enough curvature to locate the center.
Returns:
tuple: (new_lsamx_center, new_lsamy_center) in mm.
"""
aligner = XrayEyeAlignGUI(self.client, self)
try:
return aligner.find_rotation_center_smear_experimental(
keep_shutter_open=keep_shutter_open, sweep_deg=sweep_deg
)
except KeyboardInterrupt as exc:
print("Smear rotation-center calibration interrupted by user.")
try:
if not dev.cam_xeye.live_mode_enabled.get():
dev.cam_xeye.live_mode_enabled.put(True)
except Exception as cam_exc: # pylint: disable=broad-except
logger.warning(f"Failed to restore cam_xeye live mode: {cam_exc}")
try:
aligner.gui.hide_crosshair()
except Exception as gui_exc: # pylint: disable=broad-except
logger.warning(f"Failed to hide XRayEye alignment crosshair: {gui_exc}")
try:
aligner.gui.set_dap_params_forwarding(False)
except Exception as gui_exc: # pylint: disable=broad-except
logger.warning(f"Failed to disable XRayEye DAP parameter forwarding: {gui_exc}")
raise exc
# ------------------------------------------------------------------
# RT feedback / interferometer helpers
# ------------------------------------------------------------------
csaxs_bec/bec_ipython_client/plugins/LamNI/x_ray_eye_align.py
+244
View File
@@ -21,6 +21,10 @@ def umv(*args):
return scans.umv(*args, relative=False)
def mv(*args):
return scans.mv(*args, relative=False)
if TYPE_CHECKING:
from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni import LamNI
@@ -655,6 +659,246 @@ class XrayEyeAlign:
print(f"[rotation-center] at {self.get_tomo_angle():.1f} deg")
print("[rotation-center] live sweep done")
# --- EXPERIMENTAL: rotation-center smear mode -------------------------
# Max-projection "star-trail" composite aid for the extended-sample
# rotation-center click, built while lsamrot rotates continuously.
# Operator-visual only -- no automatic circle fitting. See
# find_rotation_center_smear_experimental() below for the entry point.
def _push_smear_composite(self, composite: np.ndarray, hold_display_s: float):
"""Briefly freeze the GUI's live view on the current composite.
live_mode_enabled.put(False) synchronously joins IDSCamera's
background capture thread before returning (see
IDSCamera._stop_live()'s thread.join()), so there is no race
between this push and the next raw frame the live thread would
otherwise publish to the same PreviewSignal/device_preview
channel the GUI is subscribed to.
"""
dev.cam_xeye.live_mode_enabled.put(False)
dev.cam_xeye.image.put(composite)
time.sleep(hold_display_s)
dev.cam_xeye.live_mode_enabled.put(True)
def _smear_sweep(
self,
sweep_deg: float = 360.0,
display_update_interval_s: float = 1.0,
hold_display_s: float = 0.4,
poll_interval_s: float = 0.15,
) -> np.ndarray:
"""Rotate lsamrot continuously through sweep_deg while accumulating
a max-projection composite of camera frames grabbed along the way.
Off-axis features smear into circular arcs as the sample rotates;
the common center of curvature of those arcs is the rotation axis
-- an easier target for the operator to click than a single live
frame. sweep_deg need not be a full 360 -- a smaller arc may
already show enough curvature to locate the center.
The rotation is issued non-blocking (mv(), the sibling of this
file's own blocking umv() wrapper) so frames can be grabbed via
get_last_image() in a plain single-threaded loop while the move
is still in progress on the server -- ScanReport.status is polled
directly, exactly what ScanReport.wait() does internally, so no
background thread is needed here.
"""
print(
f"[rotation-center][smear] EXPERIMENTAL feature: max-projection composite aid, "
"operator-visual only -- no automatic circle fitting."
)
if not dev.cam_xeye.live_mode_enabled.get():
dev.cam_xeye.live_mode_enabled.put(True)
self.gui.on_live_view_enabled(True)
dev.fsh.fshopen()
self._disable_rt_feedback()
start = self.get_tomo_angle()
target = start + sweep_deg
print(
f"[rotation-center][smear] starting continuous {sweep_deg:.1f} deg sweep: "
f"{start:.1f} -> {target:.1f} deg"
)
report = mv(self.device_manager.devices.lsamrot, target)
composite = None
last_frame = None
frames_seen = 0
last_push = time.time()
try:
while report.status != "COMPLETED":
frame = dev.cam_xeye.get_last_image()
if composite is None:
composite = frame.copy()
elif frame.shape != composite.shape:
print(
f"[rotation-center][smear] skipping frame with mismatched shape "
f"{frame.shape} != {composite.shape}"
)
else:
if last_frame is None or not np.array_equal(frame, last_frame):
frames_seen += 1
composite = np.maximum(composite, frame)
last_frame = frame
if time.time() - last_push > display_update_interval_s:
self._push_smear_composite(composite, hold_display_s)
last_push = time.time()
time.sleep(poll_interval_s)
except KeyboardInterrupt:
print("[rotation-center][smear] sweep interrupted -- cancelling in-flight rotation.")
report.cancel()
if not dev.cam_xeye.live_mode_enabled.get():
dev.cam_xeye.live_mode_enabled.put(True)
raise
self._push_smear_composite(composite, hold_display_s)
print(
f"[rotation-center][smear] sweep done -- {frames_seen} distinct frames over "
f"{sweep_deg:.1f} deg, now at {self.get_tomo_angle():.1f} deg"
)
return composite
def find_rotation_center_smear_experimental(
self, keep_shutter_open: bool = False, apply: bool = True, sweep_deg: float = 360.0
):
"""EXPERIMENTAL: extended-sample rotation-center calibration aided
by a continuous-rotation max-projection composite (see
_smear_sweep()), instead of a single instant of live rotation.
Otherwise identical in structure to find_rotation_center(sample_type="extended"):
FZP reference click, then iterate sweep -> click -> apply -> verify
-> accept/redo, then the same apply-confirmation/shutter-close
prompts. This is a separate, distinctly-named method -- it does not
modify find_rotation_center() or ROTATION_CENTER_SAMPLE_TYPES.
Args:
keep_shutter_open: passed through to update_frame(), see align().
apply: if True (default), ask for confirmation and then persist
the computed values via dev.lsamx/lsamy.update_user_parameter().
sweep_deg: angular range of the continuous smear sweep. Defaults
to a full 360 deg (complete arcs), but a smaller range may
already show enough curvature to locate the center.
Returns:
tuple: (new_lsamx_center, new_lsamy_center) in mm.
"""
self.lamni.lamnigui_show_xeyealign()
self.gui.set_dap_params_forwarding(False)
self._reset_init_values()
self.alignment_images = []
self.movement_buttons_enabled(False, False)
self.gui.enable_submit_button(False)
dev.omny_xray_gui.mvx.set(0)
dev.omny_xray_gui.mvy.set(0)
dev.omny_xray_gui.submit.set(0)
print(
"[rotation-center][smear] starting find_rotation_center_smear_experimental"
f"(sweep_deg={sweep_deg!r})"
)
self.send_message("Getting things ready. Please wait...")
self.lamni.losa_out()
self._ensure_at_configured_center()
self.lamni.lfzp_in(force_feedback_reset=True)
self.update_frame(keep_shutter_open)
fzp_x, fzp_y = self._collect_click(0, "Submit centre of FZP.", label="FZP reference")
self.send_message("Please wait - moving sample in...")
self.lamni.loptics_out()
self._disable_rt_feedback()
self.tomo_rotate(0)
self.update_frame(keep_shutter_open)
self.gui.set_crosshair_position(fzp_x / self.pixel_calibration, fzp_y / self.pixel_calibration)
self.gui.show_crosshair()
iteration = 0
cumulative_shift_x, cumulative_shift_y = 0.0, 0.0
while True:
iteration += 1
print(f"[rotation-center][smear] === iteration {iteration} ===")
self.send_message(
"Watch the sample rotate and locate the stationary rotation centre..."
)
self._smear_sweep(sweep_deg=sweep_deg)
self.update_frame(keep_shutter_open)
self.movement_buttons_enabled(True, True)
center_x, center_y = self._collect_click(
1,
"Submit the rotation centre (0 deg).",
label=f"rotation centre, iter {iteration}",
)
delta_x, delta_y = self._compute_shift_to_fzp(fzp_x, fzp_y, center_x, center_y)
cumulative_shift_x += delta_x
cumulative_shift_y += delta_y
print(
f"[rotation-center][smear] cumulative shift from originally-configured center: "
f"({cumulative_shift_x:.4f}, {cumulative_shift_y:.4f}) mm"
)
self._apply_rotation_center_shift(cumulative_shift_x, cumulative_shift_y)
self.send_message("Verifying alignment...")
self._live_sweep([45, 0])
self.update_frame(keep_shutter_open)
answer = (
input("Alignment acceptable -- stop here? [Y/n] (n = run another iteration): ")
.strip()
.lower()
)
if answer in ("", "y", "yes"):
print(
f"[rotation-center][smear] operator accepted alignment after "
f"{iteration} iteration(s)"
)
break
print("[rotation-center][smear] operator requested another iteration.")
self.gui.hide_crosshair()
new_lsamx = self.device_manager.devices.lsamx.readback.read()["lsamx"]["value"]
new_lsamy = self.device_manager.devices.lsamy.readback.read()["lsamy"]["value"]
old_lsamx = dev.lsamx.user_parameter.get("center")
old_lsamy = dev.lsamy.user_parameter.get("center")
print(
f"[rotation-center][smear] RESULT: new lsamx_center={new_lsamx:.4f} mm "
f"(was {old_lsamx}), new lsamy_center={new_lsamy:.4f} mm (was {old_lsamy})"
)
if apply:
answer = (
input(
f"Update lsamx/lsamy center user parameters to "
f"({new_lsamx:.4f}, {new_lsamy:.4f})? [Y/n]: "
)
.strip()
.lower()
)
if answer in ("", "y", "yes"):
dev.lsamx.update_user_parameter({"center": float(new_lsamx)})
dev.lsamy.update_user_parameter({"center": float(new_lsamy)})
print(
f"[rotation-center][smear] lsamx.user_parameter['center'] = "
f"{dev.lsamx.user_parameter.get('center')}, "
f"lsamy.user_parameter['center'] = {dev.lsamy.user_parameter.get('center')}"
)
else:
print(
"[rotation-center][smear] NOT updated -- config unchanged. Apply manually "
"with dev.lsamx.update_user_parameter({'center': ...}) / same for lsamy."
)
if keep_shutter_open:
answer = input("Close the shutter now? [Y/n]: ").strip().lower()
if answer in ("", "y", "yes"):
dev.fsh.fshclose()
self.gui.on_live_view_enabled(False)
print("Shutter closed.")
else:
print("Shutter left open.")
return (new_lsamx, new_lsamy)
# --- end EXPERIMENTAL: rotation-center smear mode ---------------------
def _ensure_at_configured_center(self, tol: float = 0.003):
"""Move lsamx/lsamy to their currently-configured center if they
aren't already there.
tests/tests_bec_ipython_client/test_x_ray_eye_align.py
+104 -1
View File
@@ -1,5 +1,6 @@
from unittest import mock
import numpy as np
import pytest
from bec_lib.device import DeviceBase
@@ -365,4 +366,106 @@ def test_find_rotation_center_extended_iterates_until_happy(bec_client_mock):
# one lamni_move_to_scan_center pair per iteration -> 2 iterations = 4 calls
assert align.scans.lamni_move_to_scan_center.call_count == 4
dev_mock.lsamx.update_user_parameter.assert_called_once()
dev_mock.lsamy.update_user_parameter.assert_called_once()
dev_mock.lsamy.update_user_parameter.assert_called_once()
# ----------------------------------------------------------------------
# EXPERIMENTAL: rotation-center smear mode (_smear_sweep)
# ----------------------------------------------------------------------
def _report_with_status_sequence(*statuses):
report = mock.MagicMock()
type(report).status = mock.PropertyMock(side_effect=list(statuses))
return report
def test_push_smear_composite_duty_cycles_live_mode(bec_client_mock):
client = bec_client_mock
align, dev_mock = _make_calibration_align(client)
composite = np.zeros((2, 2), dtype=np.uint8)
with mock.patch(f"{XRAY_EYE_ALIGN}.dev", dev_mock):
with mock.patch(f"{XRAY_EYE_ALIGN}.time.sleep"):
align._push_smear_composite(composite, hold_display_s=0.4)
# live view must be turned off before the composite is pushed (so the
# background live-mode thread can't overwrite it) and back on after.
assert [c.args[0] for c in dev_mock.cam_xeye.live_mode_enabled.put.call_args_list] == [
False,
True,
]
dev_mock.cam_xeye.image.put.assert_called_once_with(composite)
def test_smear_sweep_max_projection(bec_client_mock):
client = bec_client_mock
align, dev_mock = _make_calibration_align(client)
frame_a = np.array([[1, 5], [3, 2]], dtype=np.uint8)
frame_b = np.array([[4, 2], [3, 6]], dtype=np.uint8)
frame_c = np.array([[0, 1], [9, 0]], dtype=np.uint8)
dev_mock.cam_xeye.get_last_image.side_effect = [frame_a, frame_b, frame_c]
report = _report_with_status_sequence("RUNNING", "RUNNING", "RUNNING", "COMPLETED")
with mock.patch(f"{XRAY_EYE_ALIGN}.dev", dev_mock):
with mock.patch(f"{XRAY_EYE_ALIGN}.mv", return_value=report):
with mock.patch(f"{XRAY_EYE_ALIGN}.time.sleep"):
composite = align._smear_sweep()
expected = np.maximum(np.maximum(frame_a, frame_b), frame_c)
assert np.array_equal(composite, expected)
# exactly 3 frames were grabbed (get_last_image's side_effect list is
# exhausted) -- no extra grabs happened after status went COMPLETED.
assert dev_mock.cam_xeye.get_last_image.call_count == 3
def test_smear_sweep_issues_single_continuous_move(bec_client_mock):
client = bec_client_mock
align, dev_mock = _make_calibration_align(client)
dev_mock.cam_xeye.get_last_image.return_value = np.zeros((2, 2), dtype=np.uint8)
report = _report_with_status_sequence("COMPLETED")
with mock.patch(f"{XRAY_EYE_ALIGN}.dev", dev_mock):
with mock.patch(f"{XRAY_EYE_ALIGN}.mv", return_value=report) as mv_mock:
with mock.patch(f"{XRAY_EYE_ALIGN}.time.sleep"):
align._smear_sweep(sweep_deg=90.0)
# lsamrot readback starts at 0.0 (see _fake_positioner in _make_calibration_align)
mv_mock.assert_called_once_with(align.device_manager.devices.lsamrot, 90.0)
def test_smear_sweep_skips_frame_with_mismatched_shape(bec_client_mock, capsys):
client = bec_client_mock
align, dev_mock = _make_calibration_align(client)
frame_a = np.zeros((2, 2), dtype=np.uint8)
frame_bad = np.zeros((3, 3), dtype=np.uint8)
dev_mock.cam_xeye.get_last_image.side_effect = [frame_a, frame_bad]
report = _report_with_status_sequence("RUNNING", "RUNNING", "COMPLETED")
with mock.patch(f"{XRAY_EYE_ALIGN}.dev", dev_mock):
with mock.patch(f"{XRAY_EYE_ALIGN}.mv", return_value=report):
with mock.patch(f"{XRAY_EYE_ALIGN}.time.sleep"):
composite = align._smear_sweep()
assert np.array_equal(composite, frame_a)
assert "mismatched shape" in capsys.readouterr().out
def test_smear_sweep_cancels_report_on_keyboard_interrupt(bec_client_mock):
client = bec_client_mock
align, dev_mock = _make_calibration_align(client)
dev_mock.cam_xeye.live_mode_enabled.get.return_value = False
dev_mock.cam_xeye.get_last_image.side_effect = KeyboardInterrupt()
report = mock.MagicMock()
type(report).status = mock.PropertyMock(return_value="RUNNING")
with mock.patch(f"{XRAY_EYE_ALIGN}.dev", dev_mock):
with mock.patch(f"{XRAY_EYE_ALIGN}.mv", return_value=report):
with mock.patch(f"{XRAY_EYE_ALIGN}.time.sleep"):
with pytest.raises(KeyboardInterrupt):
align._smear_sweep()
report.cancel.assert_called_once()
dev_mock.cam_xeye.live_mode_enabled.put.assert_any_call(True)