feat(omny): add real EPICS owner + measured-status channel for sample storage

Gives OMNY sample storage a dedicated owner + measured-status channel
(paired with the new SAMPLE_DB IOC record in OMNY/OMNY.git, one per sample
position: VAL=status enum none/started/completed, DESC=owner), replacing
the pack_desc/DESC-packing hack the widget briefly used for OMNY owner
support. FlOMNI is untouched this pass -- deferred.

- OMNYSampleStorage: new sample_status/sample_owners Dcpt/Cpt groups;
  set_sample_slot/set_sample_in_gripper/set_sample_in_samplestage take an
  owner param and reset status on every place/rename/clear; new
  set_sample_status/get_sample_status/get_sample_owner/
  get_sample_name_and_owner (+ gripper/stage equivalents); show_all()/
  help() updated.
- SimOMNYSampleStorage: seeding clears the new owner/status signals too.
- omny_sample_transfer_mixin.py: otransfer_* wrappers for all the new
  methods, existing set_sample_slot/set_sample_in_gripper/
  set_sample_in_samplestage wrappers pass owner through.
- sample_storage.py widget: OMNY cells read/write owner+status via the new
  signals instead of packing; the "✓ measured"/"◐ started" badge now works
  for OMNY too, plus a manual "Set measured status…" context-menu action.

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

csaxs_bec/bec_ipython_client/plugins/omny/omny_sample_transfer_mixin.py

@@ -1077,20 +1077,20 @@ class OMNYSampleTransferMixin:
     def otransfer_is_sample_in_gripper(self) -> bool:
         return dev.omny_samples.is_sample_in_gripper()
 
-    def otransfer_set_sample_slot(self, container: str, slot_nr: int, name: str) -> bool:
-        return dev.omny_samples.set_sample_slot(container, slot_nr, name)
+    def otransfer_set_sample_slot(self, container: str, slot_nr: int, name: str, owner: str = "") -> bool:
+        return dev.omny_samples.set_sample_slot(container, slot_nr, name, owner)
 
     def otransfer_unset_sample_slot(self, container: str, slot_nr: int) -> bool:
         return dev.omny_samples.unset_sample_slot(container, slot_nr)
 
-    def otransfer_set_sample_in_gripper(self, name: str) -> bool:
-        return dev.omny_samples.set_sample_in_gripper(name)
+    def otransfer_set_sample_in_gripper(self, name: str, owner: str = "") -> bool:
+        return dev.omny_samples.set_sample_in_gripper(name, owner)
 
     def otransfer_unset_sample_in_gripper(self) -> bool:
         return dev.omny_samples.unset_sample_in_gripper()
 
-    def otransfer_set_sample_in_samplestage(self, name: str) -> bool:
-        return dev.omny_samples.set_sample_in_samplestage(name)
+    def otransfer_set_sample_in_samplestage(self, name: str, owner: str = "") -> bool:
+        return dev.omny_samples.set_sample_in_samplestage(name, owner)
 
     def otransfer_unset_sample_in_samplestage(self) -> bool:
         return dev.omny_samples.unset_sample_in_samplestage()
@@ -1122,6 +1122,36 @@ class OMNYSampleTransferMixin:
     def otransfer_search_shuttle_in_slot(self, shuttle: str) -> int:
         return dev.omny_samples.search_shuttle_in_slot(shuttle)
 
+    def otransfer_get_sample_owner(self, container: str, slot_nr: int) -> str:
+        return dev.omny_samples.get_sample_owner(container, slot_nr)
+
+    def otransfer_get_sample_name_and_owner(self, container: str, slot_nr: int) -> tuple:
+        return dev.omny_samples.get_sample_name_and_owner(container, slot_nr)
+
+    def otransfer_set_sample_status(self, container: str, slot_nr: int, status: str) -> bool:
+        return dev.omny_samples.set_sample_status(container, slot_nr, status)
+
+    def otransfer_get_sample_status(self, container: str, slot_nr: int) -> str:
+        return dev.omny_samples.get_sample_status(container, slot_nr)
+
+    def otransfer_get_sample_owner_in_gripper(self) -> str:
+        return dev.omny_samples.get_sample_owner_in_gripper()
+
+    def otransfer_get_sample_owner_in_samplestage(self) -> str:
+        return dev.omny_samples.get_sample_owner_in_samplestage()
+
+    def otransfer_set_sample_status_in_gripper(self, status: str) -> bool:
+        return dev.omny_samples.set_sample_status_in_gripper(status)
+
+    def otransfer_get_sample_status_in_gripper(self) -> str:
+        return dev.omny_samples.get_sample_status_in_gripper()
+
+    def otransfer_set_sample_status_in_samplestage(self, status: str) -> bool:
+        return dev.omny_samples.set_sample_status_in_samplestage(status)
+
+    def otransfer_get_sample_status_in_samplestage(self) -> str:
+        return dev.omny_samples.get_sample_status_in_samplestage()
+
     # def _otransfer_modify_sample(_ocontainer, _oposition, _ostatus, _oname):
     #   #_ocontainer is A, B, C for the pin containers
     #   #it can also be FZP, OSA and omny

csaxs_bec/bec_widgets/widgets/sample_storage/sample_storage.py

@@ -9,13 +9,13 @@ different setups, detected at construction time (see ``SETUP_PROFILES``,
 ``_detect_setup()``) — a BEC session is always exactly one setup, never
 both at once:
 
-Both setups require every sample slot to carry a name *and* an owner,
-packed into one EPICS DESC field (see ``sample_desc_codec``) since neither
-backend has a separate owner PV.
+Both setups require every sample slot to carry a name *and* an owner, but
+via different backend mechanisms:
 
 FlOMNI (``dev.flomni_samples``)
     A flat magazine: slots 1-20, the sample stage (slot 0) and the gripper
-    (slot 100).
+    (slot 100). Name and owner are packed into one EPICS DESC field (see
+    ``sample_desc_codec``) since this backend has no separate owner PV.
 
 OMNY (``dev.omny_samples``)
     A richer structure: three shuttles (A/B/C) of 6 slots each, a sparse
@@ -23,9 +23,13 @@ OMNY (``dev.omny_samples``)
     separate sample stage and gripper, and a 6-slot "parking" carousel
     recording which shuttle letter is docked at each physical parking
     position (parking slots are a shuttle-letter assignment, not a sample,
-    so they carry no name/owner). Whichever shuttle's parking slot matches
-    the ``oparkz`` motor's current readback is "active", i.e. at the
-    transfer-ready position (see ``oparkz_active_parking_slot()``).
+    so they carry no name/owner/status). Unlike FlOMNI, OMNY has a
+    dedicated owner channel (and a measured-status channel -- see
+    ``OMNY_SampleStorage.STATUS_NONE``/``STATUS_STARTED``/
+    ``STATUS_COMPLETED``) next to each sample slot's existing placed/name
+    channels, so no packing is needed there. Whichever shuttle's parking
+    slot matches the ``oparkz`` motor's current readback is "active", i.e.
+    at the transfer-ready position (see ``oparkz_active_parking_slot()``).
 
 A LamNI session has neither device; the widget shows a "no known setup
 detected" message in that case.
@@ -127,6 +131,14 @@ OMNY_PARKING_SLOTS = tuple(range(1, 7))
 # slots (which use EMPTY_NAME = "-"), parking slots are cleared to "none".
 PARKING_EMPTY_SENTINEL = "none"
 
+# Mirrors OMNYSampleStorage.STATUS_NONE/STATUS_STARTED/STATUS_COMPLETED's
+# int<->str mapping, duplicated locally rather than reaching into that
+# class's own (private) _status_str_to_val()/_status_val_to_str()
+# classmethods -- this widget already duplicates other device-side
+# conventions the same way (see EMPTY_NAME/PARKING_EMPTY_SENTINEL above).
+OMNY_STATUS_OPTIONS = ("none", "started", "completed")
+OMNY_STATUS_STR_TO_VAL = {"none": 0, "started": 1, "completed": 2}
+
 # colours
 # An occupied slot's name label intentionally has no fixed foreground colour
 # here (see _SlotCell.set_state()) -- it was previously hardcoded to
@@ -388,10 +400,17 @@ class _SlotCell(QFrame):
         menu = QMenu(self)
         if self._occupied:
             act_rename = menu.addAction("Change name…")
+            # OMNY-only (tuple key): measured status has a real EPICS
+            # channel there (see OMNY_SampleStorage), unlike FlOMNI's
+            # scan-lifecycle-driven redis log, which isn't manually
+            # editable from here.
+            act_status = menu.addAction("Set measured status…") if isinstance(self._slot, tuple) else None
             act_clear = menu.addAction("Clear (set empty)")
             chosen = menu.exec_(self.mapToGlobal(pos))
             if chosen == act_rename:
                 self._owner.action_change_name(self._slot, self._name, self._sample_owner)
+            elif act_status is not None and chosen == act_status:
+                self._owner.action_set_status(self._slot)
             elif chosen == act_clear:
                 self._owner.action_clear(self._slot)
         else:
@@ -620,10 +639,11 @@ class OMNY_SampleStorage(BECWidget, QWidget):
         """Parse a dev.omny_samples.read(cached=True) dict into
         {key: (occupied, name, owner, measured_status)} for every
         sample-kind slot -- stage, gripper, the 10 fixed O positions, and
-        the 18 shuttle slots. Like FlOMNI, OMNY has no separate owner PV, so
-        the name+owner pair is packed into the one raw name field (see
-        ``sample_desc_codec``); measured_status is always None -- OMNY has
-        no measured-sample-log equivalent (see module docstring).
+        the 18 shuttle slots. Owner and measured status each have their own
+        dedicated OMNYSampleStorage signal (added alongside the existing
+        placed/name signals -- see that class's ``STATUS_NONE``/
+        ``STATUS_STARTED``/``STATUS_COMPLETED``), unlike FlOMNI which packs
+        name+owner into one field via ``sample_desc_codec``.
         """
 
         def _val(key, default=None):
@@ -632,32 +652,43 @@ class OMNY_SampleStorage(BECWidget, QWidget):
                 return entry.get("value", default)
             return default
 
-        def _name_owner(raw):
-            name, owner = unpack_desc(raw if raw else EMPTY_NAME)
-            return (name if name else EMPTY_NAME), owner
+        def _status_str(raw) -> Optional[str]:
+            # mirrors OMNYSampleStorage._status_val_to_str(), without a live
+            # device round-trip: this widget only has the bulk-read dict.
+            try:
+                val = int(round(float(raw)))
+            except (TypeError, ValueError):
+                return None
+            return {1: "started", 2: "completed"}.get(val)  # 0 ("none") -> None: no badge
 
         state: dict[tuple, tuple[bool, str, str, Optional[str]]] = {}
 
         used = bool(_val("omny_samples_sample_in_samplestage", 0))
-        name, owner = _name_owner(_val("omny_samples_sample_in_samplestage_name", EMPTY_NAME))
-        state[("stage",)] = (used, name, owner, None)
+        name = _val("omny_samples_sample_in_samplestage_name", EMPTY_NAME) or EMPTY_NAME
+        owner = _val("omny_samples_sample_in_samplestage_owner", "") or ""
+        status = _status_str(_val("omny_samples_sample_in_samplestage_status", 0))
+        state[("stage",)] = (used, name, owner, status)
 
         used = bool(_val("omny_samples_sample_in_gripper", 0))
-        name, owner = _name_owner(_val("omny_samples_sample_in_gripper_name", EMPTY_NAME))
-        state[("gripper",)] = (used, name, owner, None)
+        name = _val("omny_samples_sample_in_gripper_name", EMPTY_NAME) or EMPTY_NAME
+        owner = _val("omny_samples_sample_in_gripper_owner", "") or ""
+        status = _status_str(_val("omny_samples_sample_in_gripper_status", 0))
+        state[("gripper",)] = (used, name, owner, status)
 
         for n in OMNY_FIXED_POSITIONS:
             used = bool(_val(f"omny_samples_sample_placed_sample{n}", 0))
-            name, owner = _name_owner(_val(f"omny_samples_sample_names_sample{n}", EMPTY_NAME))
-            state[("O", n)] = (used, name, owner, None)
+            name = _val(f"omny_samples_sample_names_sample{n}", EMPTY_NAME) or EMPTY_NAME
+            owner = _val(f"omny_samples_sample_owners_sample{n}", "") or ""
+            status = _status_str(_val(f"omny_samples_sample_status_sample{n}", 0))
+            state[("O", n)] = (used, name, owner, status)
 
         for letter in OMNY_SHUTTLE_LETTERS:
             for n in OMNY_SHUTTLE_SLOTS:
                 used = bool(_val(f"omny_samples_sample_shuttle_{letter}_placed_sample{n}", 0))
-                name, owner = _name_owner(
-                    _val(f"omny_samples_sample_shuttle_{letter}_names_sample{n}", EMPTY_NAME)
-                )
-                state[("shuttle", letter, n)] = (used, name, owner, None)
+                name = _val(f"omny_samples_sample_shuttle_{letter}_names_sample{n}", EMPTY_NAME) or EMPTY_NAME
+                owner = _val(f"omny_samples_sample_shuttle_{letter}_owners_sample{n}", "") or ""
+                status = _status_str(_val(f"omny_samples_sample_shuttle_{letter}_status_sample{n}", 0))
+                state[("shuttle", letter, n)] = (used, name, owner, status)
 
         return state
 
@@ -796,29 +827,41 @@ class OMNY_SampleStorage(BECWidget, QWidget):
     def _write_slot_omny(self, key: tuple, used: int, name: str, owner: str = "") -> bool:
         """Write (used, name, owner) to an OMNY sample-kind slot -- stage,
         gripper, a fixed O position, or a shuttle slot -- mirroring the
-        direct-signal style of ``_write_slot()``, including packing
-        name+owner into the one raw name field the same way (see
-        ``sample_desc_codec``). No measured-sample-log integration exists
-        for OMNY (see module docstring), so unlike ``_write_slot`` this
-        never offers to print a P-touch label for the previous occupant.
+        direct-signal style of ``_write_slot()``. Unlike FlOMNI, OMNY has a
+        dedicated owner signal (added alongside placed/name -- see
+        ``OMNYSampleStorage``), so name and owner are written separately,
+        no packing needed. Measured status is reset to "none" on every
+        write here, mirroring ``OMNYSampleStorage.set_sample_slot()``'s/
+        ``unset_sample_slot()``'s own behavior -- a fresh name (even a
+        rename) or a cleared slot isn't mid-measurement. No P-touch
+        label-printing offer for the previous occupant (FlOMNI-only, see
+        module docstring).
         """
         try:
-            packed_name = pack_desc(name, owner)
+            status_none = self._samples.STATUS_NONE
             kind = key[0]
             if kind == "stage":
                 self._samples.sample_in_samplestage.set(used)
-                self._samples.sample_in_samplestage_name.set(packed_name)
+                self._samples.sample_in_samplestage_name.set(name)
+                self._samples.sample_in_samplestage_owner.set(owner)
+                self._samples.sample_in_samplestage_status.set(status_none)
             elif kind == "gripper":
                 self._samples.sample_in_gripper.set(used)
-                self._samples.sample_in_gripper_name.set(packed_name)
+                self._samples.sample_in_gripper_name.set(name)
+                self._samples.sample_in_gripper_owner.set(owner)
+                self._samples.sample_in_gripper_status.set(status_none)
             elif kind == "O":
                 _, n = key
                 getattr(self._samples.sample_placed, f"sample{n}").set(used)
-                getattr(self._samples.sample_names, f"sample{n}").set(packed_name)
+                getattr(self._samples.sample_names, f"sample{n}").set(name)
+                getattr(self._samples.sample_owners, f"sample{n}").set(owner)
+                getattr(self._samples.sample_status, f"sample{n}").set(status_none)
             elif kind == "shuttle":
                 _, letter, n = key
                 getattr(getattr(self._samples, f"sample_shuttle_{letter}_placed"), f"sample{n}").set(used)
-                getattr(getattr(self._samples, f"sample_shuttle_{letter}_names"), f"sample{n}").set(packed_name)
+                getattr(getattr(self._samples, f"sample_shuttle_{letter}_names"), f"sample{n}").set(name)
+                getattr(getattr(self._samples, f"sample_shuttle_{letter}_owners"), f"sample{n}").set(owner)
+                getattr(getattr(self._samples, f"sample_shuttle_{letter}_status"), f"sample{n}").set(status_none)
             else:
                 raise ValueError(f"unknown OMNY slot key {key!r}")
             self._last_state.pop(key, None)
@@ -832,6 +875,39 @@ class OMNY_SampleStorage(BECWidget, QWidget):
             )
             return False
 
+    def _write_status_omny(self, key: tuple, status: str) -> bool:
+        """Write just the measured-status signal for an OMNY sample-kind
+        slot, leaving used/name/owner untouched -- the counterpart to
+        ``_write_slot_omny()`` resetting status on every name/owner write,
+        for the manual "Set measured status…" action (see
+        ``action_set_status``).
+        """
+        try:
+            val = OMNY_STATUS_STR_TO_VAL[status]
+            kind = key[0]
+            if kind == "stage":
+                self._samples.sample_in_samplestage_status.set(val)
+            elif kind == "gripper":
+                self._samples.sample_in_gripper_status.set(val)
+            elif kind == "O":
+                _, n = key
+                getattr(self._samples.sample_status, f"sample{n}").set(val)
+            elif kind == "shuttle":
+                _, letter, n = key
+                getattr(getattr(self._samples, f"sample_shuttle_{letter}_status"), f"sample{n}").set(val)
+            else:
+                raise ValueError(f"unknown OMNY slot key {key!r}")
+            self._last_state.pop(key, None)
+            return True
+        except Exception as exc:
+            logger.warning(f"OMNY_SampleStorage: write of status for slot {key} failed: {exc}")
+            QMessageBox.critical(
+                self,
+                "Write failed",
+                f"Could not write status for slot {self._slot_title(key)} to the device:\n{exc}",
+            )
+            return False
+
     def _write_parking_omny(self, slot_nr: int, letter: Optional[str]) -> bool:
         """Assign (or clear) which shuttle letter is recorded as parked at
         OMNY parking slot `slot_nr` (1-6), mirroring
@@ -1186,6 +1262,32 @@ class OMNY_SampleStorage(BECWidget, QWidget):
         if self._write_sample(slot, 0, EMPTY_NAME):
             self.refresh()
 
+    def action_set_status(self, slot: tuple) -> None:
+        """Manually set an OMNY sample-kind slot's measured status. OMNY
+        only (tuple key) -- see ``_SlotCell._show_menu``. FlOMNI's status
+        stays a separate, scan-lifecycle-driven redis log (see module
+        docstring), not exposed for manual editing here.
+        """
+        if not isinstance(slot, tuple):
+            return
+        current = self._last_state.get(slot, (False, "", "", None))[3] or "none"
+        try:
+            current_index = OMNY_STATUS_OPTIONS.index(current)
+        except ValueError:
+            current_index = 0
+        status, ok = QInputDialog.getItem(
+            self,
+            "Set measured status",
+            f"Status for slot {self._slot_title(slot)}:",
+            OMNY_STATUS_OPTIONS,
+            current_index,
+            False,
+        )
+        if not ok:
+            return
+        if self._write_status_omny(slot, status):
+            self.refresh()
+
     def action_set_parking(self, letter: str, current_slot: Optional[int]) -> None:
         """Assign OMNY shuttle `letter` to a parking slot (1-6), via
         ``_ShuttleStatusLabel``'s context menu. Overwrites that slot's

csaxs_bec/devices/omny/omny_sample_storage.py

@@ -23,7 +23,17 @@ class OMNYSampleStorage(Device):
         "unset_sample_in_samplestage",
         "get_sample_name_in_samplestage",
         "get_sample_name_in_gripper",
+        "get_sample_owner_in_samplestage",
+        "get_sample_owner_in_gripper",
+        "set_sample_status_in_samplestage",
+        "get_sample_status_in_samplestage",
+        "set_sample_status_in_gripper",
+        "get_sample_status_in_gripper",
         "get_sample_name",
+        "get_sample_owner",
+        "get_sample_name_and_owner",
+        "set_sample_status",
+        "get_sample_status",
         "is_sample_in_samplestage",
         "set_shuttle_slot",
         "unset_shuttle_slot",
@@ -36,6 +46,18 @@ class OMNYSampleStorage(Device):
     SUB_VALUE = "value"
     _default_sub = SUB_VALUE
 
+    # Measured status, one of these three -- an "ai" record's VAL (see
+    # omny_samples_status.template in the OMNY/OMNY.git IOC repo), not a
+    # proper EPICS enum: this IOC's sample DB uses plain "ai" records
+    # uniformly even for boolean-ish fields (see the placed-flag records),
+    # so the new STATUS record follows the same convention rather than
+    # introducing an mbbo/mbbi record type used nowhere else here. The
+    # int<->str mapping lives entirely in this class.
+    STATUS_NONE = 0
+    STATUS_STARTED = 1
+    STATUS_COMPLETED = 2
+    _STATUS_LABELS = {STATUS_NONE: "none", STATUS_STARTED: "started", STATUS_COMPLETED: "completed"}
+
     sample_shuttle_A_placed = {
         f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_shuttle_A:{i}", {"auto_monitor": True}) for i in range(1, 7)
     }
@@ -97,6 +119,75 @@ class OMNYSampleStorage(Device):
     }
     sample_names = Dcpt(sample_names)
 
+    # Measured status (VAL) + owner (DESC) -- one new record per sample
+    # position, added alongside the existing placed/name records (see
+    # STATUS_NONE/STARTED/COMPLETED above and omny_samples_status.template
+    # in OMNY/OMNY.git). Parking slots have no equivalent -- they hold a
+    # shuttle letter, not a sample.
+    sample_shuttle_A_status = {
+        f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_shuttle_A:{i}:STATUS", {"auto_monitor": True})
+        for i in range(1, 7)
+    }
+    sample_shuttle_A_status = Dcpt(sample_shuttle_A_status)
+
+    sample_shuttle_B_status = {
+        f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_shuttle_B:{i}:STATUS", {"auto_monitor": True})
+        for i in range(1, 7)
+    }
+    sample_shuttle_B_status = Dcpt(sample_shuttle_B_status)
+
+    sample_shuttle_C_status = {
+        f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_shuttle_C:{i}:STATUS", {"auto_monitor": True})
+        for i in range(1, 7)
+    }
+    sample_shuttle_C_status = Dcpt(sample_shuttle_C_status)
+
+    sample_status = {
+        f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_omny:{i}:STATUS", {"auto_monitor": True})
+        for i in [10, 11, 12, 13, 14, 32, 33, 34, 100, 101]
+    }
+    sample_status = Dcpt(sample_status)
+
+    sample_shuttle_A_owners = {
+        f"sample{i}": (
+            EpicsSignal,
+            f"XOMNY-SAMPLE_DB_shuttle_A:{i}:STATUS.DESC",
+            {"string": True, "auto_monitor": True},
+        )
+        for i in range(1, 7)
+    }
+    sample_shuttle_A_owners = Dcpt(sample_shuttle_A_owners)
+
+    sample_shuttle_B_owners = {
+        f"sample{i}": (
+            EpicsSignal,
+            f"XOMNY-SAMPLE_DB_shuttle_B:{i}:STATUS.DESC",
+            {"string": True, "auto_monitor": True},
+        )
+        for i in range(1, 7)
+    }
+    sample_shuttle_B_owners = Dcpt(sample_shuttle_B_owners)
+
+    sample_shuttle_C_owners = {
+        f"sample{i}": (
+            EpicsSignal,
+            f"XOMNY-SAMPLE_DB_shuttle_C:{i}:STATUS.DESC",
+            {"string": True, "auto_monitor": True},
+        )
+        for i in range(1, 7)
+    }
+    sample_shuttle_C_owners = Dcpt(sample_shuttle_C_owners)
+
+    sample_owners = {
+        f"sample{i}": (
+            EpicsSignal,
+            f"XOMNY-SAMPLE_DB_omny:{i}:STATUS.DESC",
+            {"string": True, "auto_monitor": True},
+        )
+        for i in [10, 11, 12, 13, 14, 32, 33, 34, 100, 101]
+    }
+    sample_owners = Dcpt(sample_owners)
+
     sample_in_gripper = Cpt(
         EpicsSignal, name="sample_in_gripper", read_pv="XOMNY-SAMPLE_DB_omny:110.VAL", auto_monitor=True
     )
@@ -119,6 +210,28 @@ class OMNYSampleStorage(Device):
         auto_monitor=True
     )
 
+    sample_in_gripper_status = Cpt(
+        EpicsSignal, name="sample_in_gripper_status", read_pv="XOMNY-SAMPLE_DB_omny:110:STATUS", auto_monitor=True
+    )
+    sample_in_gripper_owner = Cpt(
+        EpicsSignal,
+        name="sample_in_gripper_owner",
+        read_pv="XOMNY-SAMPLE_DB_omny:110:STATUS.DESC",
+        string=True,
+        auto_monitor=True,
+    )
+
+    sample_in_samplestage_status = Cpt(
+        EpicsSignal, name="sample_in_samplestage_status", read_pv="XOMNY-SAMPLE_DB_omny:0:STATUS", auto_monitor=True
+    )
+    sample_in_samplestage_owner = Cpt(
+        EpicsSignal,
+        name="sample_in_samplestage_owner",
+        read_pv="XOMNY-SAMPLE_DB_omny:0:STATUS.DESC",
+        string=True,
+        auto_monitor=True,
+    )
+
     def __init__(self, prefix="", *, name, **kwargs):
         super().__init__(prefix, name=name, **kwargs)
         self.sample_shuttle_A_placed.sample1.subscribe(self._emit_value)
@@ -128,41 +241,114 @@ class OMNYSampleStorage(Device):
         self.wait_for_connection()
         self._run_subs(sub_type=self.SUB_VALUE, timestamp=timestamp, obj=self)
 
-    def set_sample_slot(self, container: str, slot_nr: int, name: str) -> bool:
+    @classmethod
+    def _status_str_to_val(cls, status: str) -> int:
+        status = (status or "none").strip().lower()
+        for val, label in cls._STATUS_LABELS.items():
+            if label == status:
+                return val
+        raise OMNYSampleStorageError(
+            f"Invalid status {status!r}; expected one of {sorted(cls._STATUS_LABELS.values())}."
+        )
+
+    @classmethod
+    def _status_val_to_str(cls, val) -> str:
+        try:
+            return cls._STATUS_LABELS.get(int(round(float(val))), "none")
+        except (TypeError, ValueError):
+            return "none"
+
+    def set_sample_slot(self, container: str, slot_nr: int, name: str, owner: str = "") -> bool:
         if slot_nr > 20:
             raise OMNYSampleStorageError(f"Invalid slot number {slot_nr}.")
 
+        status = self._status_str_to_val("none")
         if container == "A":
             getattr(self.sample_shuttle_A_placed, f"sample{slot_nr}").set(1)
             getattr(self.sample_shuttle_A_names, f"sample{slot_nr}").set(name)
+            getattr(self.sample_shuttle_A_owners, f"sample{slot_nr}").set(owner)
+            getattr(self.sample_shuttle_A_status, f"sample{slot_nr}").set(status)
         elif container == "B":
             getattr(self.sample_shuttle_B_placed, f"sample{slot_nr}").set(1)
             getattr(self.sample_shuttle_B_names, f"sample{slot_nr}").set(name)
+            getattr(self.sample_shuttle_B_owners, f"sample{slot_nr}").set(owner)
+            getattr(self.sample_shuttle_B_status, f"sample{slot_nr}").set(status)
         elif container == "C":
             getattr(self.sample_shuttle_C_placed, f"sample{slot_nr}").set(1)
             getattr(self.sample_shuttle_C_names, f"sample{slot_nr}").set(name)
+            getattr(self.sample_shuttle_C_owners, f"sample{slot_nr}").set(owner)
+            getattr(self.sample_shuttle_C_status, f"sample{slot_nr}").set(status)
         elif container == "O":
             getattr(self.sample_placed, f"sample{slot_nr}").set(1)
             getattr(self.sample_names, f"sample{slot_nr}").set(name)
-
+            getattr(self.sample_owners, f"sample{slot_nr}").set(owner)
+            getattr(self.sample_status, f"sample{slot_nr}").set(status)
 
     def unset_sample_slot(self, container: str, slot_nr: int) -> bool:
         if slot_nr > 20:
             raise OMNYSampleStorageError(f"Invalid slot number {slot_nr}.")
 
+        status = self._status_str_to_val("none")
         if container == "A":
             getattr(self.sample_shuttle_A_placed, f"sample{slot_nr}").set(0)
             getattr(self.sample_shuttle_A_names, f"sample{slot_nr}").set("-")
+            getattr(self.sample_shuttle_A_owners, f"sample{slot_nr}").set("")
+            getattr(self.sample_shuttle_A_status, f"sample{slot_nr}").set(status)
         elif container == "B":
             getattr(self.sample_shuttle_B_placed, f"sample{slot_nr}").set(0)
             getattr(self.sample_shuttle_B_names, f"sample{slot_nr}").set("-")
+            getattr(self.sample_shuttle_B_owners, f"sample{slot_nr}").set("")
+            getattr(self.sample_shuttle_B_status, f"sample{slot_nr}").set(status)
         elif container == "C":
             getattr(self.sample_shuttle_C_placed, f"sample{slot_nr}").set(0)
             getattr(self.sample_shuttle_C_names, f"sample{slot_nr}").set("-")
+            getattr(self.sample_shuttle_C_owners, f"sample{slot_nr}").set("")
+            getattr(self.sample_shuttle_C_status, f"sample{slot_nr}").set(status)
         elif container == "O":
             getattr(self.sample_placed, f"sample{slot_nr}").set(0)
             getattr(self.sample_names, f"sample{slot_nr}").set("-")
+            getattr(self.sample_owners, f"sample{slot_nr}").set("")
+            getattr(self.sample_status, f"sample{slot_nr}").set(status)
 
+    def set_sample_status(self, container: str, slot_nr: int, status: str) -> bool:
+        val = self._status_str_to_val(status)
+        if container == "A":
+            getattr(self.sample_shuttle_A_status, f"sample{slot_nr}").set(val)
+        elif container == "B":
+            getattr(self.sample_shuttle_B_status, f"sample{slot_nr}").set(val)
+        elif container == "C":
+            getattr(self.sample_shuttle_C_status, f"sample{slot_nr}").set(val)
+        elif container == "O":
+            getattr(self.sample_status, f"sample{slot_nr}").set(val)
+
+    def get_sample_status(self, container: str, slot_nr: int) -> str:
+        if container == "A":
+            val = getattr(self.sample_shuttle_A_status, f"sample{slot_nr}").get()
+        elif container == "B":
+            val = getattr(self.sample_shuttle_B_status, f"sample{slot_nr}").get()
+        elif container == "C":
+            val = getattr(self.sample_shuttle_C_status, f"sample{slot_nr}").get()
+        elif container == "O":
+            val = getattr(self.sample_status, f"sample{slot_nr}").get()
+        else:
+            return "none"
+        return self._status_val_to_str(val)
+
+    def get_sample_owner(self, container: str, slot_nr: int) -> str:
+        if container == "A":
+            val = getattr(self.sample_shuttle_A_owners, f"sample{slot_nr}").get()
+        elif container == "B":
+            val = getattr(self.sample_shuttle_B_owners, f"sample{slot_nr}").get()
+        elif container == "C":
+            val = getattr(self.sample_shuttle_C_owners, f"sample{slot_nr}").get()
+        elif container == "O":
+            val = getattr(self.sample_owners, f"sample{slot_nr}").get()
+        else:
+            return ""
+        return str(val) if val else ""
+
+    def get_sample_name_and_owner(self, container: str, slot_nr: int) -> tuple:
+        return self.get_sample_name(container, slot_nr), self.get_sample_owner(container, slot_nr)
 
     def set_shuttle_slot(self, container: str, slot_nr: int) -> bool:
         if slot_nr > 6:
@@ -176,21 +362,49 @@ class OMNYSampleStorage(Device):
         getattr(self.parking_placed, f"parking{slot_nr}").set(0)
         getattr(self.parking_names, f"parking{slot_nr}").set("none")
 
-    def set_sample_in_gripper(self, name: str) -> bool:
+    def set_sample_in_gripper(self, name: str, owner: str = "") -> bool:
         self.sample_in_gripper.set(1)
         self.sample_in_gripper_name.set(name)
+        self.sample_in_gripper_owner.set(owner)
+        self.sample_in_gripper_status.set(self._status_str_to_val("none"))
 
     def unset_sample_in_gripper(self) -> bool:
         self.sample_in_gripper.set(0)
         self.sample_in_gripper_name.set("-")
+        self.sample_in_gripper_owner.set("")
+        self.sample_in_gripper_status.set(self._status_str_to_val("none"))
 
-    def set_sample_in_samplestage(self, name: str) -> bool:
+    def set_sample_in_samplestage(self, name: str, owner: str = "") -> bool:
         self.sample_in_samplestage.set(1)
         self.sample_in_samplestage_name.set(name)
+        self.sample_in_samplestage_owner.set(owner)
+        self.sample_in_samplestage_status.set(self._status_str_to_val("none"))
 
     def unset_sample_in_samplestage(self) -> bool:
         self.sample_in_samplestage.set(0)
         self.sample_in_samplestage_name.set("-")
+        self.sample_in_samplestage_owner.set("")
+        self.sample_in_samplestage_status.set(self._status_str_to_val("none"))
+
+    def get_sample_owner_in_gripper(self) -> str:
+        val = self.sample_in_gripper_owner.get()
+        return str(val) if val else ""
+
+    def get_sample_owner_in_samplestage(self) -> str:
+        val = self.sample_in_samplestage_owner.get()
+        return str(val) if val else ""
+
+    def set_sample_status_in_gripper(self, status: str) -> bool:
+        self.sample_in_gripper_status.set(self._status_str_to_val(status))
+
+    def get_sample_status_in_gripper(self) -> str:
+        return self._status_val_to_str(self.sample_in_gripper_status.get())
+
+    def set_sample_status_in_samplestage(self, status: str) -> bool:
+        self.sample_in_samplestage_status.set(self._status_str_to_val(status))
+
+    def get_sample_status_in_samplestage(self) -> str:
+        return self._status_val_to_str(self.sample_in_samplestage_status.get())
 
     def is_sample_slot_used(self, container, slot_nr: int) -> bool:
         if container == "A":
@@ -247,6 +461,23 @@ class OMNYSampleStorage(Device):
                 returnvalue = i
         return returnvalue
 
+    def _describe_slot(self, container: str, slot_nr: int) -> str:
+        """Sample name, plus owner/measured-status suffix when occupied --
+        e.g. "sample1 (bob)  ◐ started". Used by show_all() so the
+        overview surfaces owner+status without changing get_sample_name()'s
+        own plain-name contract (still used elsewhere, e.g. transfers)."""
+        name = self.get_sample_name(container, slot_nr)
+        if not self.is_sample_slot_used(container, slot_nr):
+            return name
+        owner = self.get_sample_owner(container, slot_nr)
+        status = self.get_sample_status(container, slot_nr)
+        suffix = f" ({owner})" if owner else ""
+        if status == "started":
+            suffix += "  ◐ started"
+        elif status == "completed":
+            suffix += "  ✓ completed"
+        return name + suffix
+
     def show_all(self):
         t = PrettyTable()
         red = "\x1b[91m"
@@ -262,22 +493,32 @@ class OMNYSampleStorage(Device):
             field_names = [""]
             for ax in [1, 3, 5]:
                 row = []
-                row.extend([self.get_sample_name(ch, ax)])
+                row.extend([self._describe_slot(ch, ax)])
                 row.extend(str(ax))
                 row.extend(str(ax + 1))
-                row.extend([self.get_sample_name(ch, ax + 1)])
+                row.extend([self._describe_slot(ch, ax + 1)])
                 t.add_row(row)
             t.header = False
             t.vrules = FRAME
             print(t)
 
         if self.is_sample_in_samplestage():
-            print(f"\n\n  Sample stage:   {self.get_sample_name_in_samplestage()}")
+            name = self.get_sample_name_in_samplestage()
+            owner = self.get_sample_owner_in_samplestage()
+            status = self.get_sample_status_in_samplestage()
+            suffix = f" ({owner})" if owner else ""
+            if status == "started":
+                suffix += "  ◐ started"
+            elif status == "completed":
+                suffix += "  ✓ completed"
+            print(f"\n\n  Sample stage:   {name}{suffix}")
         else:
             print(f"\n\n  Sample stage:   no sample")
 
         if self.is_sample_in_gripper():
-            print(f"\n  Gripper:        {self.get_sample_name_in_gripper()}\n")
+            name = self.get_sample_name_in_gripper()
+            owner = self.get_sample_owner_in_gripper()
+            print(f"\n  Gripper:        {name}" + (f" ({owner})" if owner else "") + "\n")
         else:
             print(f"\n  Gripper:        no sample\n")
 
@@ -287,8 +528,7 @@ class OMNYSampleStorage(Device):
             row = []
             row.extend([f"Position {i:3d}"])
             if self.is_sample_slot_used("O", i):
-                name = self.get_sample_name("O", i)
-                row.extend([name])
+                row.extend([self._describe_slot("O", i)])
             else:
                 row.extend(["free"])
             t.add_row(row)
@@ -301,10 +541,14 @@ class OMNYSampleStorage(Device):
         print(" To get an overview use dev.omny_samples.show_all()")
         print(" Modify a slot:")
         print(" dev.omny_samples.unset_sample_slot('system',position)")
-        print(" dev.omny_samples.set_sample_slot('system',position,'name')")
+        print(" dev.omny_samples.set_sample_slot('system',position,'name','owner')")
         print("   system can be A, B, C, O")
-        print(" dev.omny_samples.set_sample_in_gripper('name') / unset_sample_in_gripper()")
-        print(" dev.omny_samples.set_sample_in_samplestage('name'), unset_sample_in_samplestage()")
+        print(" dev.omny_samples.get_sample_owner('system',position)")
+        print(" dev.omny_samples.set_sample_status('system',position,'status')")
+        print("   status is one of 'none', 'started', 'completed'")
+        print(" dev.omny_samples.get_sample_status('system',position)")
+        print(" dev.omny_samples.set_sample_in_gripper('name','owner') / unset_sample_in_gripper()")
+        print(" dev.omny_samples.set_sample_in_samplestage('name','owner'), unset_sample_in_samplestage()")
         print(" dev.omny_samples.set_shuttle_slot(container, slot_nr) / unset_shuttle_slot(slot_nr)")
         print(" dev.omny_samples.set_shuttle_slot('A',2)")
         print(" omny.otransfer_help()")

csaxs_bec/devices/sim/sim_omny.py

@@ -266,17 +266,41 @@ class SimOMNYSampleStorage(OMNYSampleStorage):
             # 0-20-only range check, but OMNY's own "O" fixed positions
             # legitimately include 32-34/100/101), which would raise here
             # for exactly the positions this needs to clear.
-            for shuttle_placed, shuttle_names in (
-                (storage.sample_shuttle_A_placed, storage.sample_shuttle_A_names),
-                (storage.sample_shuttle_B_placed, storage.sample_shuttle_B_names),
-                (storage.sample_shuttle_C_placed, storage.sample_shuttle_C_names),
+            # Same clearing for the owner/status signals added alongside
+            # placed/names -- untouched positions should read owner="" and
+            # status="none", not the mock-PV "0" default (an invalid status
+            # string, would fail _status_val_to_str's lookup silently back
+            # to "none" anyway, but owner would show a stray "0").
+            for shuttle_placed, shuttle_names, shuttle_owners, shuttle_status in (
+                (
+                    storage.sample_shuttle_A_placed,
+                    storage.sample_shuttle_A_names,
+                    storage.sample_shuttle_A_owners,
+                    storage.sample_shuttle_A_status,
+                ),
+                (
+                    storage.sample_shuttle_B_placed,
+                    storage.sample_shuttle_B_names,
+                    storage.sample_shuttle_B_owners,
+                    storage.sample_shuttle_B_status,
+                ),
+                (
+                    storage.sample_shuttle_C_placed,
+                    storage.sample_shuttle_C_names,
+                    storage.sample_shuttle_C_owners,
+                    storage.sample_shuttle_C_status,
+                ),
             ):
                 for slot_nr in shuttle_placed.component_names:
                     getattr(shuttle_placed, slot_nr).set(0)
                     getattr(shuttle_names, slot_nr).set("-")
+                    getattr(shuttle_owners, slot_nr).set("")
+                    getattr(shuttle_status, slot_nr).set(OMNYSampleStorage.STATUS_NONE)
             for slot_nr in storage.sample_placed.component_names:
                 getattr(storage.sample_placed, slot_nr).set(0)
                 getattr(storage.sample_names, slot_nr).set("-")
+                getattr(storage.sample_owners, slot_nr).set("")
+                getattr(storage.sample_status, slot_nr).set(OMNYSampleStorage.STATUS_NONE)
             for slot_nr in storage.parking_placed.component_names:
                 getattr(storage.parking_placed, slot_nr).set(0)
                 getattr(storage.parking_names, slot_nr).set("none")