feat(omny): save reference images at sample-transfer confirm checkpoints

Mirrors flomni's ftransfer_confirm()/save_reference_image() pattern, adapted
for OMNY's dual-camera-pair setup: whichever camera pair is currently shown
live (cam200+cam203 for parking/shuttle, cam201+cam202 for the sample
stage -- see _omnycam_parking()/_omnycam_samplestage() in gui_tools.py) gets
saved together into one HDF5 file at each manual confirm checkpoint in
_otransfer_confirm(), via the new _otransfer_save_reference_images().

X-ray eye alignment image/fit saving is left untouched -- it already exists
as shared code (x_ray_eye_align.py) but has known pre-existing OMNY/LamNI
naming bugs unrelated to this change.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018Mupip8x7Hean6E3FFvpbj

csaxs_bec/bec_ipython_client/plugins/omny/gui_tools.py

@@ -96,6 +96,12 @@ class OMNYGuiTools:
 
     def _omnycam_parking(self):
         self.omnygui_show_omnycam_parking()
+        # Recorded so omny_sample_transfer_mixin.py's _otransfer_confirm()
+        # knows which camera pair to save a reference image from -- OMNY
+        # always shows exactly one pair (parking/shuttle or sample stage)
+        # depending on where the gripper currently is, unlike flomni's
+        # fixed single gripper camera.
+        self._omnycam_active_pair = ("cam200", "cam203")
 
     def omnygui_show_omnycam_parking(self):
         self.omnygui_show_gui()
@@ -136,6 +142,7 @@ class OMNYGuiTools:
 
     def _omnycam_samplestage(self):
         self.omnygui_show_omnycam_samplestage()
+        self._omnycam_active_pair = ("cam201", "cam202")
 
     def omnygui_show_omnycam_samplestage(self):
         self.omnygui_show_gui()

csaxs_bec/bec_ipython_client/plugins/omny/omny_sample_transfer_mixin.py

@@ -1,6 +1,8 @@
+import os
 import time
 import builtins
 
+import h5py
 import numpy as np
 
 from rich import box
@@ -740,6 +742,7 @@ class OMNYSampleTransferMixin:
         return bool(float(dev.otransy.controller.socket_put_and_receive("MGgrstat")))
 
     def otransfer_put_sample(self, pin_position: int):
+        self.transfer_step = 0
         self._otransfer_check_sensor_connected()
         if not self._otransfer_check_free_slot_available_at_position(pin_position, 0):
             raise OMNYTransferError(f"position {pin_position} is not free")
@@ -760,7 +763,7 @@ class OMNYSampleTransferMixin:
 
         print("The mount process started.")
 
-        self._otransfer_confirm()
+        self._otransfer_confirm(step_name="put")
         self._otransfer_controller_disable_mount_mode()
         # update the new sample status in storage
         self._otransfer_check_free_slot_available_at_position(pin_position, 1)
@@ -789,7 +792,7 @@ class OMNYSampleTransferMixin:
             self._otransfer_shield_close(pin_position)
         dev.omny_samples.show_all()
 
-    def _otransfer_confirm(self):
+    def _otransfer_confirm(self, step_name: str = ""):
         mntprgs = 1
 
         while mntprgs != 0:
@@ -820,10 +823,48 @@ class OMNYSampleTransferMixin:
                     text = "All OK? Please confirm"
 
                 if self.OMNYTools.yesno(text, "y"):
+                    self.transfer_step += 1
+                    self._otransfer_save_reference_images(file_suffix=f"{step_name}_{self.transfer_step}")
                     dev.otransy.controller.socket_put_confirmed("confirm=1")
                 else:
                     raise OMNYTransferError("The sample mount progress was aborted by user.")
 
+    def _otransfer_save_reference_images(self, file_suffix: str = "") -> None:
+        """Save the currently-active camera pair's last images to one HDF5
+        file -- OMNY's equivalent of flomni's ``save_reference_image()``,
+        called at each manual confirm checkpoint (see ``_otransfer_confirm``).
+
+        Unlike flomni (one fixed camera, ``cam_flomni_gripper``), OMNY
+        always shows exactly one *pair* of cameras live at a time --
+        parking/shuttle (cam200+cam203) or sample stage (cam201+cam202) --
+        depending on where the gripper currently is (see
+        ``_omnycam_parking()``/``_omnycam_samplestage()`` in gui_tools.py,
+        which record that pair on ``self._omnycam_active_pair``). Both
+        images from the active pair go into one file, keyed by camera name,
+        rather than flomni's single "reference_image" dataset.
+
+        IDSCamera's own ``get_last_image()`` is used directly (same as
+        ``x_ray_eye_align.py``'s ``update_frame()`` already does for
+        cam_xeye), rather than flomni's BEC-preview-redis-endpoint pull --
+        cam200-203 are IDSCamera, not the WebcamViewer class flomni's
+        cameras use.
+        """
+        pair = getattr(self, "_omnycam_active_pair", None)
+        if not pair:
+            print("No active OMNY camera pair known; skipping reference-image save.")
+            return
+        suffix = f"_{file_suffix}_" if file_suffix else "_"
+        timestamp = time.strftime("%Y%m%d_%H%M%S")
+        file = os.path.expanduser(
+            f"~/data/raw/logs/omny_sample_transfer_images/omny_sample_transfer_reference_image{suffix}{timestamp}.h5"
+        )
+        os.makedirs(os.path.dirname(file), exist_ok=True)
+        with h5py.File(file, "w") as f:
+            for cam_name in pair:
+                image = getattr(dev, cam_name).get_last_image()
+                if image is not None:
+                    f.create_dataset(cam_name, data=image)
+
     def _otransfer_check_sensor_connected(self):
         sensorvoltage = float(dev.otransy.controller.socket_put_and_receive("MG@AN[8001]"))
         gripper_sensorvoltagetarget = self._get_user_param_safe(
@@ -835,6 +876,7 @@ class OMNYSampleTransferMixin:
             )
 
     def otransfer_get_sample(self, pin_position: int):
+        self.transfer_step = 0
         self._omnycam_parking()
         if not self.OMNYTools.yesno(
             "Please confirm that currently there is no sample in the gripper. It would be dropped!",
@@ -875,7 +917,7 @@ class OMNYSampleTransferMixin:
 
         print("The Unmount process started.")
         time.sleep(1)
-        self._otransfer_confirm()
+        self._otransfer_confirm(step_name="get")
 
         self._otransfer_controller_disable_mount_mode()
         # update database