diff --git a/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py b/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py index eef0837..81d897d 100644 --- a/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py +++ b/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py @@ -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 # ------------------------------------------------------------------ diff --git a/csaxs_bec/bec_ipython_client/plugins/LamNI/x_ray_eye_align.py b/csaxs_bec/bec_ipython_client/plugins/LamNI/x_ray_eye_align.py index ff189de..87e946a 100644 --- a/csaxs_bec/bec_ipython_client/plugins/LamNI/x_ray_eye_align.py +++ b/csaxs_bec/bec_ipython_client/plugins/LamNI/x_ray_eye_align.py @@ -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. diff --git a/tests/tests_bec_ipython_client/test_x_ray_eye_align.py b/tests/tests_bec_ipython_client/test_x_ray_eye_align.py index 2435dc3..5930066 100644 --- a/tests/tests_bec_ipython_client/test_x_ray_eye_align.py +++ b/tests/tests_bec_ipython_client/test_x_ray_eye_align.py @@ -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() \ No newline at end of file + 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) \ No newline at end of file