feat(flomni): sample storage on the new flOMNI soft IOC (FLOMNI-SAMPLE:<slot>:<field>)

The flOMNI sample storage moves from the XOMNY-SAMPLE_DB_flomni<n>:GET/.DESC
records (name and owner packed into one 40-char DESC via sample_desc_codec,
measurement status in the BEC global var "flomni_measured_samples" keyed by
that packed string) to one record per slot on the flOMNI soft IOC
(flOMNI repo, softioc/ioc_flomni.py):

    FLOMNI-SAMPLE:<slot>:USED / NAME (up to 256 chars) / OWNER /
                         STATUS (none|started|completed) / MEAS_DATE
    <slot> = STAGE (0), T01..T20 (1-20), GRIPPER (100)

No transition (decided by Mirko 2026-10-17).

- FlomniSampleStorage: same public methods (slot 100 = gripper), attribute
  names sample_placed/sample_names/sample_in_gripper[_name] kept; new
  sample_owners/sample_status/sample_meas_date (+ gripper equivalents),
  get_sample_status/meas_date/record, set_sample_measured(slot, status,
  date, owner_if_empty) and transfer_sample(from, to), which moves the whole
  record so the measurement status travels with the sample
- flomni plugin: transfers use transfer_sample(); tomo_scan marks the stage
  sample started/completed on the record; the label offer reads status,
  date and owner from the record; _MeasuredSampleLog and
  sample_get_measured_log_key removed
  Behaviour change: the label's account is the sample owner (was the active
  account at measurement time); set_sample_measured fills an empty owner
  with the active account
- sample storage widget: bulk read incl. owner/status, writes via
  set_sample_slot/unset_sample_slot; no global-var measured log
- webpage generator: name and owner from separate fields
- e2e harness and tomo test updated; nexus writer unchanged (sample name is
  now the plain name)

Tested against the live IOC on mpc2844 (ophyd): long names, transfers slot ->
gripper -> stage with status/date, owner_if_empty, show_all, clear, slot
range check, bulk read() keys.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TNtKEk4jAyjebMDQFwRHJQ

csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py

@@ -42,7 +42,6 @@ from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.omny_general_tools import
 )
 from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.ptouch_printer import PTouchLabelPrinter
 from csaxs_bec.devices.omny.galil.galil_ophyd import GalilError
-from csaxs_bec.devices.omny.sample_desc_codec import pack_desc, unpack_desc
 
 logger = bec_logger.logger
 
@@ -818,13 +817,11 @@ class FlomniSampleTransferMixin:
         self.ftransfer_controller_disable_mount_mode()
         self.ensure_gripper_up()
 
-        signal_name = getattr(dev.flomni_samples.sample_names, f"sample{position}")
-        name, owner = unpack_desc(signal_name.get())
-        self.flomni_modify_storage_non_interactive(100, 1, name, owner=owner)
-        self.flomni_modify_storage_non_interactive(position, 0, "-")
+        # the whole record (name, owner, measurement status/date) moves to the gripper
+        dev.flomni_samples.transfer_sample(position, 100)
 
     def ftransfer_show_all(self):
-        dev.flomni_samples.show_all(measured=self.measured_log.snapshot())
+        dev.flomni_samples.show_all()
 
     def ftransfer_put_sample(self, position: int):
         self.check_position_is_valid(position)
@@ -877,9 +874,8 @@ class FlomniSampleTransferMixin:
         self.ftransfer_controller_disable_mount_mode()
         self.ensure_gripper_up()
 
-        sample_name, sample_owner = unpack_desc(dev.flomni_samples.sample_in_gripper_name.get())
-        self.flomni_modify_storage_non_interactive(100, 0, "-")
-        self.flomni_modify_storage_non_interactive(position, 1, sample_name, owner=sample_owner)
+        # the whole record (name, owner, measurement status/date) moves from the gripper
+        dev.flomni_samples.transfer_sample(100, position)
 
         if position == 0:
             self.ftransfer_flomni_stage_in()
@@ -889,18 +885,7 @@ class FlomniSampleTransferMixin:
         """
         Get the name of the sample currently in the given position.
         """
-        signal_name = getattr(dev.flomni_samples.sample_names, f"sample{position}")
-        return unpack_desc(signal_name.get())[0]
-
-    def sample_get_measured_log_key(self, position: int = 0) -> str:
-        """Packed `name | owner` identity key for the sample at `position`,
-        for use with `self.measured_log` -- NOT the same as `sample_name`
-        (a `sample_get_name()` alias), which is name-only and would silently
-        under-key the measured log against a different sample of the same
-        name but a different owner.
-        """
-        name, owner = dev.flomni_samples.get_sample_name_and_owner(position)
-        return pack_desc(name, owner)
+        return dev.flomni_samples.get_sample_name(position)
 
     def ftransfer_sample_change(self, new_sample_position: int):
         self.check_tray_in()
@@ -1039,23 +1024,22 @@ class FlomniSampleTransferMixin:
         print offer would incorrectly fire on every ordinary transfer step
         instead of only on a genuine manual clear/replace.
         """
-        if not dev.flomni_samples.is_sample_slot_used(position):
-            return
-        prev_name, prev_owner = dev.flomni_samples.get_sample_name_and_owner(position)
-        prev_packed = pack_desc(prev_name, prev_owner)
-        record = self.measured_log.pop(prev_packed)
-        if record is None:
+        # measurement status/date/owner live on the sample record in the IOC
+        # (FLOMNI-SAMPLE:<slot>:STATUS/MEAS_DATE/OWNER) since 2026-10-17
+        record = dev.flomni_samples.get_sample_record(position)
+        if not record["used"] or record["status"] == "none":
             return
+        prev_name = record["name"]
         if self.OMNYTools.yesno(
             f"Sample '{prev_name}' was measured ({record['status']}) on"
-            f" {record['measured_date']}. Print a label?",
+            f" {record['meas_date']}. Print a label?",
             "y",
         ):
             self.label_printer.ensure_ready()
             printed = self.label_printer.print_label(
                 {
-                    self._PTOUCH_FIELD_ACCOUNT: record["account"],
-                    self._PTOUCH_FIELD_DATE: record["measured_date"],
+                    self._PTOUCH_FIELD_ACCOUNT: record["owner"],
+                    self._PTOUCH_FIELD_DATE: record["meas_date"],
                     self._PTOUCH_FIELD_SAMPLENAME: prev_name,
                 }
             )
@@ -1068,15 +1052,11 @@ class FlomniSampleTransferMixin:
     def flomni_modify_storage_non_interactive(
         self, position: int, used: int, name: str, owner: str = ""
     ):
-        packed_name = pack_desc(name, owner)
-        if position == 100:
-            dev.flomni_samples.sample_in_gripper.set(used).wait(timeout=5)
-            dev.flomni_samples.sample_in_gripper_name.set(packed_name).wait(timeout=5)
+        # position 100 = gripper; a new/renamed sample starts with measurement status "none"
+        if used:
+            dev.flomni_samples.set_sample_slot(position, name, owner=owner)
         else:
-            signal = getattr(dev.flomni_samples.sample_placed, f"sample{position}")
-            signal.set(used).wait(timeout=5)
-            signal_name = getattr(dev.flomni_samples.sample_names, f"sample{position}")
-            signal_name.set(packed_name).wait(timeout=5)
+            dev.flomni_samples.unset_sample_slot(position)
 
     def check_position_is_valid(self, position: int):
         if 0 <= position < 21:
@@ -1710,83 +1690,6 @@ class _ProgressProxy:
         return self._load()
 
 
-class _MeasuredSampleLog:
-    """Tracks which flomni samples have been measured, for the P-touch
-    label-printing feature (print an account/date/name label when a
-    measured sample is cleared or replaced in tray storage).
-
-    Keyed by the packed ``name | owner`` string (see pack_desc/unpack_desc
-    in sample_desc_codec.py) rather than by slot number, since that packed
-    string is the identity that already travels with a sample across
-    ftransfer_get_sample/ftransfer_put_sample transfers between tray
-    slots, the gripper, and the stage -- tracking by slot number would
-    require threading a flag through every one of those transfer calls.
-    Persisted as a BEC global var (same pattern as _ProgressProxy above)
-    so it's visible to any client session -- CLI or GUI widget -- and
-    survives a kernel restart.
-
-    Two statuses: "started" (a fresh tomogram began for this sample) and
-    "completed" (that tomogram's scan loop ran to completion without
-    error). Both count as "measured" for is_measured()/the print offer --
-    "started" without "completed" just means the tomogram may be
-    incomplete, which callers surface in the offer wording rather than
-    suppressing the offer entirely.
-    """
-
-    _GLOBAL_VAR_KEY = "flomni_measured_samples"
-
-    def __init__(self, client):
-        self._client = client
-
-    def _load(self) -> dict:
-        val = self._client.get_global_var(self._GLOBAL_VAR_KEY)
-        return val if val is not None else {}
-
-    def _save(self, data: dict) -> None:
-        self._client.set_global_var(self._GLOBAL_VAR_KEY, data)
-
-    def mark_measured(self, sample_key: str, account: str, date: str, status: str) -> None:
-        """Record `sample_key` as measured, overwriting any existing record.
-
-        Overwriting (rather than merging) means re-measuring a sample from
-        scratch correctly resets a stale "completed" back to "started".
-        """
-        data = self._load()
-        data[sample_key] = {"account": account, "measured_date": date, "status": status}
-        self._save(data)
-
-    def is_measured(self, sample_key: str) -> bool:
-        return sample_key in self._load()
-
-    def get(self, sample_key: str) -> dict | None:
-        return self._load().get(sample_key)
-
-    def snapshot(self) -> dict:
-        """One bulk fetch of the whole log, as a plain dict.
-
-        Used instead of `get` when the caller needs to hand the data to
-        `dev.flomni_samples.show_all()`, which runs via BEC's
-        device-server RPC mechanism -- every argument gets msgpack-
-        serialized, so a bound method (`self.measured_log.get`) can't be
-        passed directly; a plain dict can.
-        """
-        return self._load()
-
-    def pop(self, sample_key: str) -> dict | None:
-        """Remove and return the record for `sample_key`, if any.
-
-        Called once the print offer for that sample has been resolved
-        (printed or declined), so the log stays bounded to
-        currently-measured-but-not-yet-cleared samples instead of growing
-        unbounded over a beamtime.
-        """
-        data = self._load()
-        record = data.pop(sample_key, None)
-        if record is not None:
-            self._save(data)
-        return record
-
-
 class Flomni(
     TomoQueueMixin,
     TomoAngleFileMixin,
@@ -1843,7 +1746,6 @@ class Flomni(
         self.reconstructor = PtychoReconstructor(self.ptycho_reconstruct_foldername)
         self.tomo_id_manager = TomoIDManager()
         self.align = XrayEyeAlign(self.client, self)
-        self.measured_log = _MeasuredSampleLog(self.client)
         self.label_printer = PTouchLabelPrinter(host="BRN94DDF8AAB8EC.psi.ch", template_number=1)
         self.set_client(client)
 
@@ -2842,11 +2744,12 @@ class Flomni(
                 "test additional info",
                 "BEC",
             )
-            self.measured_log.mark_measured(
-                self.sample_get_measured_log_key(0),
-                bec.active_account or "",
+            # measurement status on the sample record (stage slot 0) in the IOC
+            dev.flomni_samples.set_sample_measured(
+                0,
+                "started",
                 datetime.date.today().strftime("%Y-%m"),
-                status="started",
+                owner_if_empty=bec.active_account or "",
             )
             self.write_pdf_report()
             # estimated_remaining_time/estimated_finish_time/accumulated_idle_time/
@@ -3035,11 +2938,12 @@ class Flomni(
         self._print_progress()
         self._log_tomogram_timing()
         self.OMNYTools.printgreenbold("Tomoscan finished")
-        self.measured_log.mark_measured(
-            self.sample_get_measured_log_key(0),
-            bec.active_account or "",
+        # measurement status on the sample record (stage slot 0) in the IOC
+        dev.flomni_samples.set_sample_measured(
+            0,
+            "completed",
             datetime.date.today().strftime("%Y-%m"),
-            status="completed",
+            owner_if_empty=bec.active_account or "",
         )
         timing_lines = self._build_timing_lines()
         for line in timing_lines[3:]:

csaxs_bec/bec_ipython_client/plugins/flomni/flomni_webpage_generator.py

@@ -43,7 +43,6 @@ from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.webpage_generator_base imp
     _safe_get,
     _safe_float,
 )
-from csaxs_bec.devices.omny.sample_desc_codec import unpack_desc
 
 
 class FlomniWebpageGenerator(WebpageGeneratorBase):
@@ -106,19 +105,16 @@ class FlomniWebpageGenerator(WebpageGeneratorBase):
         # ── Sample name ───────────────────────────────────────────────
         # slot 0 of flomni_samples is the flOMNI measurement stage itself
         # (see FlomniSampleStorage.show_all()'s own "flOMNI stage:" label
-        # for slot 0 -- not an arbitrary tray position). The raw DESC value
-        # is packed as "name | owner" (sample_desc_codec.pack_desc, added
-        # 7e8e807) -- unpack it here too, the same way
-        # FlomniSampleStorage.get_sample_name_and_owner() and show_all()
-        # already do, so the page doesn't show the raw delimiter-joined
-        # string as the "sample name". Owner is appended in parentheses,
-        # mirroring show_all()'s own "(owner: ...)" formatting.
-        raw_sample = _safe_get(dm, "flomni_samples.sample_names.sample0")
-        if raw_sample is None:
+        # for slot 0 -- not an arbitrary tray position). Name and owner are
+        # separate IOC fields since 2026-10-17 (FLOMNI-SAMPLE:STAGE:NAME /
+        # :OWNER); owner is appended in parentheses, mirroring show_all()'s
+        # own "(owner: ...)" formatting.
+        name = _safe_get(dm, "flomni_samples.sample_names.sample0")
+        if name is None:
             sample_name = "N/A"
         else:
-            name, owner = unpack_desc(raw_sample)
-            sample_name = f"{name} (owner: {owner})" if owner else name
+            owner = _safe_get(dm, "flomni_samples.sample_owners.sample0")
+            sample_name = f"{name} (owner: {owner})" if owner else str(name)
 
         # ── Temperatures ──────────────────────────────────────────────
         temperatures = {

csaxs_bec/bec_widgets/widgets/sample_storage/sample_storage.py

@@ -29,10 +29,14 @@ single source of truth also consulted by the transfer state machine.  The
 widget reads/writes it directly, mirroring
 ``Flomni.flomni_modify_storage_non_interactive()``:
 
-  slots 0–20 : sample_placed.sample{N}.set(used)
-               sample_names.sample{N}.set(name)
-  gripper    : sample_in_gripper.set(used)
-               sample_in_gripper_name.set(name)
+  writes     : set_sample_slot(N, name, owner) / unset_sample_slot(N)
+               (N = 0–20, 100 = gripper; device methods, all record fields)
+  reads      : one bulk read() of sample_placed / sample_names /
+               sample_owners / sample_status (gripper: sample_in_gripper*)
+
+Since 2026-10-17 every slot is one record on the flOMNI soft IOC
+(FLOMNI-SAMPLE:<slot>:USED/NAME/OWNER/STATUS/MEAS_DATE); the measurement
+status travels with the sample and is no longer kept in a BEC global var.
 
 Polling
 -------
@@ -69,7 +73,6 @@ from qtpy.QtWidgets import (
 )
 
 from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.ptouch_printer import PTouchLabelPrinter
-from csaxs_bec.devices.omny.sample_desc_codec import pack_desc, unpack_desc
 
 logger = bec_logger.logger
 
@@ -115,54 +118,6 @@ PTOUCH_FIELD_SAMPLENAME = "SAMPLENAME"
 PRINTER_NAME = "WSLA_X12SA"
 
 
-# ── measured-sample log ──────────────────────────────────────────────────────
-
-
-class _MeasuredSampleLog:
-    """Tracks which flomni samples have been measured, for the P-touch
-    label-printing feature.
-
-    Deliberately duplicated here rather than imported from
-    ``flomni.py``'s ``_MeasuredSampleLog`` -- pulling in that whole plugin
-    module (CLI-only dependencies, module-level ``bec``/``dev`` builtins
-    access) into this widget's gui-server process is a heavier and more
-    fragile coupling than re-declaring this ~15-line global-var wrapper,
-    and this widget already duplicates ``flomni_modify_storage_non_interactive()``
-    the same way (see module docstring). Both copies read/write the same
-    ``flomni_measured_samples`` global-var key, so they stay consistent in
-    effect despite being independent code, exactly like the existing
-    ``_write_slot``/``flomni_modify_storage_non_interactive`` duplication.
-    """
-
-    _GLOBAL_VAR_KEY = "flomni_measured_samples"
-
-    def __init__(self, client):
-        self._client = client
-
-    def _load(self) -> dict:
-        val = self._client.get_global_var(self._GLOBAL_VAR_KEY)
-        return val if val is not None else {}
-
-    def _save(self, data: dict) -> None:
-        self._client.set_global_var(self._GLOBAL_VAR_KEY, data)
-
-    def get(self, sample_key: str) -> Optional[dict]:
-        return self._load().get(sample_key)
-
-    def snapshot(self) -> dict:
-        """One bulk fetch of the whole log, for callers (like the poll
-        loop below) that need to look up many sample keys at once without
-        one redis round-trip per key."""
-        return self._load()
-
-    def pop(self, sample_key: str) -> Optional[dict]:
-        data = self._load()
-        record = data.pop(sample_key, None)
-        if record is not None:
-            self._save(data)
-        return record
-
-
 # ── slot cell widget ─────────────────────────────────────────────────────────
 
 
@@ -232,7 +187,7 @@ class _SlotCell(QFrame):
     ) -> None:
         """Update the cell's displayed state (called by the poll/refresh).
 
-        `measured_status` is "started"/"completed" (from `_MeasuredSampleLog`)
+        `measured_status` is "started"/"completed" (FLOMNI-SAMPLE:<slot>:STATUS)
         or None if this slot's current occupant has no measured record.
         """
         self._occupied = occupied
@@ -309,7 +264,6 @@ class OMNY_SampleStorage(BECWidget, QWidget):
         # last-seen (occupied, name, owner, measured_status) per slot, for
         # change-detection in refresh()
         self._last_state: dict[int, tuple[bool, str, str, Optional[str]]] = {}
-        self.measured_log = _MeasuredSampleLog(self.client)
         self.label_printer = PTouchLabelPrinter(
             host=PTOUCH_PRINTER_HOST, template_number=PTOUCH_TEMPLATE_NUMBER
         )
@@ -345,9 +299,8 @@ class OMNY_SampleStorage(BECWidget, QWidget):
     def _read_all_slots(self) -> dict[int, tuple[bool, str, str, Optional[str]]]:
         """Return {slot: (occupied, name, owner, measured_status)} for every
         slot 0–20 and the gripper from a SINGLE bulk
-        ``dev.flomni_samples.read()`` round-trip plus a single bulk
-        ``_MeasuredSampleLog.snapshot()`` (one redis round-trip, not one
-        per occupied slot).
+        ``dev.flomni_samples.read()`` round-trip (name, owner and
+        measurement status are separate IOC fields since 2026-10-17).
 
         The per-slot accessors (``is_sample_slot_used`` +
         ``sample_names.sample{N}.get()``) would be ~44 blocking device
@@ -358,8 +311,7 @@ class OMNY_SampleStorage(BECWidget, QWidget):
         ``flomni_samples_sample_names_sample{N}`` /
         ``flomni_samples_sample_in_gripper`` /
         ``flomni_samples_sample_in_gripper_name``), the same keys the CLI
-        iterated over in ftransfer_sample_change. The raw name signal packs
-        name+owner (see ``sample_desc_codec``), unpacked here.
+        iterated over in ftransfer_sample_change.
 
         Every underlying signal is an ``auto_monitor=True`` ``EpicsSignal``
         (see ``FlomniSampleStorage``), so BEC already keeps a fresh cached
@@ -375,40 +327,31 @@ class OMNY_SampleStorage(BECWidget, QWidget):
             # keep whatever is currently displayed rather than blanking out
             return {}
 
-        try:
-            measured = self.measured_log.snapshot()
-        except Exception as exc:
-            logger.warning(f"OMNY_SampleStorage: measured-log read failed: {exc}")
-            measured = {}
-
         def _val(key, default=None):
             entry = data.get(key)
             if isinstance(entry, dict):
                 return entry.get("value", default)
             return default
 
-        def _measured_status(used: bool, name: str, owner: str) -> Optional[str]:
-            if not used:
-                return None
-            record = measured.get(pack_desc(name, owner))
-            return record["status"] if record else None
+        def _status(used: bool, raw) -> Optional[str]:
+            status = str(raw) if raw is not None else ""
+            return status if used and status in ("started", "completed") else None
 
         for slot in (STAGE_SLOT, *STORAGE_SLOTS):
-            used = _val(f"flomni_samples_sample_placed_sample{slot}", 0)
-            raw_name = _val(f"flomni_samples_sample_names_sample{slot}", EMPTY_NAME)
-            name, owner = unpack_desc(raw_name if raw_name else EMPTY_NAME)
-            name = name if name else EMPTY_NAME
-            result[slot] = (bool(used), name, owner, _measured_status(bool(used), name, owner))
+            used = bool(_val(f"flomni_samples_sample_placed_sample{slot}", 0))
+            name = _val(f"flomni_samples_sample_names_sample{slot}", EMPTY_NAME) or EMPTY_NAME
+            owner = _val(f"flomni_samples_sample_owners_sample{slot}", "") or ""
+            status = _status(used, _val(f"flomni_samples_sample_status_sample{slot}"))
+            result[slot] = (used, str(name), str(owner), status)
 
-        g_used = _val("flomni_samples_sample_in_gripper", 0)
-        g_raw_name = _val("flomni_samples_sample_in_gripper_name", EMPTY_NAME)
-        g_name, g_owner = unpack_desc(g_raw_name if g_raw_name else EMPTY_NAME)
-        g_name = g_name if g_name else EMPTY_NAME
+        g_used = bool(_val("flomni_samples_sample_in_gripper", 0))
+        g_name = _val("flomni_samples_sample_in_gripper_name", EMPTY_NAME) or EMPTY_NAME
+        g_owner = _val("flomni_samples_sample_in_gripper_owner", "") or ""
         result[GRIPPER_SLOT] = (
-            bool(g_used),
-            g_name,
-            g_owner,
-            _measured_status(bool(g_used), g_name, g_owner),
+            g_used,
+            str(g_name),
+            str(g_owner),
+            _status(g_used, _val("flomni_samples_sample_in_gripper_status")),
         )
 
         return result
@@ -425,39 +368,32 @@ class OMNY_SampleStorage(BECWidget, QWidget):
         ``Flomni.flomni_modify_storage_non_interactive()``.
         """
         try:
-            if slot == GRIPPER_SLOT:
-                was_used = bool(self._samples.sample_in_gripper.get())
-                prev_name, prev_owner = unpack_desc(str(self._samples.sample_in_gripper_name.get()))
-            else:
-                was_used = self._samples.is_sample_slot_used(slot)
-                prev_name, prev_owner = self._samples.get_sample_name_and_owner(slot)
+            record = self._samples.get_sample_record(slot)
         except Exception as exc:
             logger.warning(
                 f"OMNY_SampleStorage: could not read previous state of slot {slot} for the"
                 f" measured-label check: {exc}"
             )
             return
-        if not was_used:
-            return
-        record = self.measured_log.pop(pack_desc(prev_name, prev_owner))
-        if record is None:
+        if not record["used"] or record["status"] not in ("started", "completed"):
             return
+        prev_name = record["name"]
         reply = QMessageBox.question(
             self,
             "Print label?",
             f"Sample '{prev_name}' was measured ({record['status']}) on"
-            f" {record['measured_date']}. Print a label?",
+            f" {record['meas_date']}. Print a label?",
             QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
         )
         if reply != QMessageBox.StandardButton.Yes:
             return
         fields = {
-            PTOUCH_FIELD_ACCOUNT: record["account"],
-            PTOUCH_FIELD_DATE: record["measured_date"],
+            PTOUCH_FIELD_ACCOUNT: record["owner"],
+            PTOUCH_FIELD_DATE: record["meas_date"],
             PTOUCH_FIELD_SAMPLENAME: prev_name,
         }
-        # Retry loop instead of a single attempt: the record was already
-        # popped from measured_log above, so if we gave up after one
+        # Retry loop instead of a single attempt: the slot record is
+        # overwritten right after this offer, so if we gave up after one
         # failure the label's account/date/name would be gone for good
         # even though the operator could just switch the printer on and
         # try again a moment later. Mirrors the CLI's ensure_ready() retry
@@ -487,13 +423,12 @@ class OMNY_SampleStorage(BECWidget, QWidget):
         """
         self._offer_print_label_for_previous_occupant(slot)
         try:
-            packed_name = pack_desc(name, owner)
-            if slot == GRIPPER_SLOT:
-                self._samples.sample_in_gripper.set(used)
-                self._samples.sample_in_gripper_name.set(packed_name)
+            # device methods write the whole record (slot 100 = gripper); a new or
+            # renamed sample starts with measurement status "none"
+            if used:
+                self._samples.set_sample_slot(slot, name, owner=owner)
             else:
-                getattr(self._samples.sample_placed, f"sample{slot}").set(used)
-                getattr(self._samples.sample_names, f"sample{slot}").set(packed_name)
+                self._samples.unset_sample_slot(slot)
             # drop the cached state for this slot so the refresh() right after
             # a write always repaints it, without waiting for the value to
             # differ from a possibly-stale cache entry

csaxs_bec/devices/omny/flomni_sample_storage.py

@@ -1,5 +1,27 @@
+"""flOMNI sample storage on the flOMNI soft IOC (``ioc_flomni.py`` in the flOMNI repo).
+
+One record per slot, all fields autosaved by the IOC::
+
+    FLOMNI-SAMPLE:<slot>:USED       0 = free, 1 = sample present
+    FLOMNI-SAMPLE:<slot>:NAME       sample name (long string, up to 256 chars)
+    FLOMNI-SAMPLE:<slot>:OWNER      e-account of the owner
+    FLOMNI-SAMPLE:<slot>:STATUS     none / started / completed (measurement status)
+    FLOMNI-SAMPLE:<slot>:MEAS_DATE  date of the measurement
+
+    <slot> = STAGE (slot 0, the flOMNI sample stage), T01 ... T20 (tray slots 1-20),
+             GRIPPER (slot 100)
+
+Replaces (2026-10-17) the former ``XOMNY-SAMPLE_DB_flomni<n>:GET`` / ``.DESC`` records,
+where name and owner were packed into one 40-character DESC (``sample_desc_codec``)
+and the measurement status lived in a separate BEC global variable keyed by that
+packed string. The status now travels with the sample: ``transfer_sample()`` moves
+all fields between slots, the gripper and the stage.
+
+An empty string cannot be written to the char-array NAME via Channel Access, so a
+free slot carries the name ``EMPTY_NAME`` ("-").
+"""
+
 import time
-from typing import Optional
 
 from ophyd import Component as Cpt
 from ophyd import Device
@@ -7,13 +29,36 @@ from ophyd import DynamicDeviceComponent as Dcpt
 from ophyd import EpicsSignal
 from prettytable import PrettyTable
 
-from csaxs_bec.devices.omny.sample_desc_codec import pack_desc, unpack_desc
+PV_PREFIX = "FLOMNI-SAMPLE:"
+GRIPPER_SLOT = 100
+STAGE_SLOT = 0
+TRAY_SLOTS = range(1, 21)
+EMPTY_NAME = "-"
+STATUS_NONE = "none"
+STATUS_STARTED = "started"
+STATUS_COMPLETED = "completed"
+STATUS_VALUES = (STATUS_NONE, STATUS_STARTED, STATUS_COMPLETED)
 
 
 class FlomniSampleStorageError(Exception):
     pass
 
 
+def _slot_pv(slot_nr: int) -> str:
+    if slot_nr == STAGE_SLOT:
+        return "STAGE"
+    if slot_nr == GRIPPER_SLOT:
+        return "GRIPPER"
+    return f"T{slot_nr:02d}"
+
+
+def _slot_signals(field: str, **kwargs) -> dict:
+    return {
+        f"sample{i}": (EpicsSignal, f"{PV_PREFIX}{_slot_pv(i)}:{field}", dict(auto_monitor=True, **kwargs))
+        for i in range(21)
+    }
+
+
 class FlomniSampleStorage(Device):
     USER_ACCESS = [
         "is_sample_slot_used",
@@ -25,30 +70,36 @@ class FlomniSampleStorage(Device):
         "get_sample_name",
         "get_sample_owner",
         "get_sample_name_and_owner",
+        "get_sample_status",
+        "get_sample_meas_date",
+        "get_sample_record",
+        "set_sample_measured",
+        "transfer_sample",
         "show_all",
     ]
     SUB_VALUE = "value"
     _default_sub = SUB_VALUE
-    sample_placed = {
-        f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_flomni{i}:GET", {"auto_monitor": True}) for i in range(21)
-    }
-    sample_placed = Dcpt(sample_placed)
 
-    sample_names = {
-        f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_flomni{i}:GET.DESC", {"string": True, "auto_monitor": True})
-        for i in range(21)
-    }
-    sample_names = Dcpt(sample_names)
+    # slots 0-20 (0 = flOMNI stage); attribute names kept from the former device
+    sample_placed = Dcpt(_slot_signals("USED"))
+    sample_names = Dcpt(_slot_signals("NAME", string=True))
+    sample_owners = Dcpt(_slot_signals("OWNER", string=True))
+    sample_status = Dcpt(_slot_signals("STATUS", string=True))
+    sample_meas_date = Dcpt(_slot_signals("MEAS_DATE", string=True))
 
-    sample_in_gripper = Cpt(
-        EpicsSignal, name="sample_in_gripper", read_pv="XOMNY-SAMPLE_DB_flomni100:GET", auto_monitor=True
-    )
+    # gripper (slot 100)
+    sample_in_gripper = Cpt(EpicsSignal, read_pv=f"{PV_PREFIX}GRIPPER:USED", auto_monitor=True)
     sample_in_gripper_name = Cpt(
-        EpicsSignal,
-        name="sample_in_gripper_name",
-        read_pv="XOMNY-SAMPLE_DB_flomni100:GET.DESC",
-        string=True,
-        auto_monitor=True
+        EpicsSignal, read_pv=f"{PV_PREFIX}GRIPPER:NAME", string=True, auto_monitor=True
+    )
+    sample_in_gripper_owner = Cpt(
+        EpicsSignal, read_pv=f"{PV_PREFIX}GRIPPER:OWNER", string=True, auto_monitor=True
+    )
+    sample_in_gripper_status = Cpt(
+        EpicsSignal, read_pv=f"{PV_PREFIX}GRIPPER:STATUS", string=True, auto_monitor=True
+    )
+    sample_in_gripper_meas_date = Cpt(
+        EpicsSignal, read_pv=f"{PV_PREFIX}GRIPPER:MEAS_DATE", string=True, auto_monitor=True
     )
 
     def __init__(self, prefix="", *, name, **kwargs):
@@ -60,72 +111,132 @@ class FlomniSampleStorage(Device):
         self.wait_for_connection()
         self._run_subs(sub_type=self.SUB_VALUE, timestamp=timestamp, obj=self)
 
-    def set_sample_slot(self, slot_nr: int, name: str, owner: str = "") -> bool:
-        if slot_nr > 20:
-            raise FlomniSampleStorageError(f"Invalid slot number {slot_nr}.")
+    # ── slot addressing ──────────────────────────────────────────────────────
+    @staticmethod
+    def _check_slot(slot_nr: int) -> int:
+        slot_nr = int(slot_nr)
+        if slot_nr == GRIPPER_SLOT or 0 <= slot_nr <= 20:
+            return slot_nr
+        raise FlomniSampleStorageError(
+            f"Invalid slot number {slot_nr} (0 = stage, 1-20 = tray, {GRIPPER_SLOT} = gripper)."
+        )
 
-        getattr(self.sample_placed, f"sample{slot_nr}").set(1).wait(timeout=5)
-        getattr(self.sample_names, f"sample{slot_nr}").set(pack_desc(name, owner)).wait(timeout=5)
+    def _sig(self, field: str, slot_nr: int) -> EpicsSignal:
+        slot_nr = self._check_slot(slot_nr)
+        if slot_nr == GRIPPER_SLOT:
+            return {
+                "used": self.sample_in_gripper,
+                "name": self.sample_in_gripper_name,
+                "owner": self.sample_in_gripper_owner,
+                "status": self.sample_in_gripper_status,
+                "meas_date": self.sample_in_gripper_meas_date,
+            }[field]
+        group = {
+            "used": self.sample_placed,
+            "name": self.sample_names,
+            "owner": self.sample_owners,
+            "status": self.sample_status,
+            "meas_date": self.sample_meas_date,
+        }[field]
+        return getattr(group, f"sample{slot_nr}")
 
-    def unset_sample_slot(self, slot_nr: int) -> bool:
-        if slot_nr > 20:
-            raise FlomniSampleStorageError(f"Invalid slot number {slot_nr}.")
+    @staticmethod
+    def _str(val) -> str:
+        return "" if val is None else str(val)
 
-        getattr(self.sample_placed, f"sample{slot_nr}").set(0).wait(timeout=5)
-        getattr(self.sample_names, f"sample{slot_nr}").set("-").wait(timeout=5)
+    # ── write ────────────────────────────────────────────────────────────────
+    def _write(self, slot_nr: int, used: int, name: str, owner: str, status: str, meas_date: str):
+        if status not in STATUS_VALUES:
+            raise FlomniSampleStorageError(f"Invalid status {status!r}, expected one of {STATUS_VALUES}.")
+        name = name if name else EMPTY_NAME
+        # name/owner/status/date first, USED last: a client watching USED sees a complete record
+        self._sig("name", slot_nr).set(name).wait(timeout=5)
+        self._sig("owner", slot_nr).set(owner or "").wait(timeout=5)
+        self._sig("status", slot_nr).set(status).wait(timeout=5)
+        self._sig("meas_date", slot_nr).set(meas_date or "").wait(timeout=5)
+        self._sig("used", slot_nr).set(int(used)).wait(timeout=5)
 
-    def set_sample_in_gripper(self, name: str, owner: str = "") -> bool:
-        self.sample_in_gripper.set(1).wait(timeout=5)
-        self.sample_in_gripper_name.set(pack_desc(name, owner)).wait(timeout=5)
+    def set_sample_slot(
+        self, slot_nr: int, name: str, owner: str = "", status: str = STATUS_NONE, meas_date: str = ""
+    ):
+        """Place a sample record in slot `slot_nr` (0-20, or 100 = gripper)."""
+        self._write(slot_nr, 1, name, owner, status, meas_date)
 
-    def unset_sample_in_gripper(self) -> bool:
-        self.sample_in_gripper.set(0).wait(timeout=5)
-        self.sample_in_gripper_name.set("-").wait(timeout=5)
+    def unset_sample_slot(self, slot_nr: int):
+        """Clear slot `slot_nr` (0-20, or 100 = gripper)."""
+        self._write(slot_nr, 0, EMPTY_NAME, "", STATUS_NONE, "")
 
+    def set_sample_in_gripper(self, name: str, owner: str = "", status: str = STATUS_NONE, meas_date: str = ""):
+        self.set_sample_slot(GRIPPER_SLOT, name, owner, status, meas_date)
+
+    def unset_sample_in_gripper(self):
+        self.unset_sample_slot(GRIPPER_SLOT)
+
+    def set_sample_measured(self, slot_nr: int, status: str, meas_date: str, owner_if_empty: str = ""):
+        """Record the measurement status/date of the sample in `slot_nr`.
+
+        If the sample has no owner yet, `owner_if_empty` (e.g. the active
+        e-account) is stored as owner -- the label printed when a measured
+        sample is cleared uses the owner as account.
+        """
+        if status not in STATUS_VALUES:
+            raise FlomniSampleStorageError(f"Invalid status {status!r}, expected one of {STATUS_VALUES}.")
+        if owner_if_empty and not self.get_sample_owner(slot_nr):
+            self._sig("owner", slot_nr).set(owner_if_empty).wait(timeout=5)
+        self._sig("status", slot_nr).set(status).wait(timeout=5)
+        self._sig("meas_date", slot_nr).set(meas_date or "").wait(timeout=5)
+
+    def transfer_sample(self, from_slot: int, to_slot: int):
+        """Move the complete record (name, owner, status, date) from one slot to
+        another (0-20, 100 = gripper) and clear the source slot."""
+        rec = self.get_sample_record(from_slot)
+        self.set_sample_slot(to_slot, rec["name"], rec["owner"], rec["status"], rec["meas_date"])
+        self.unset_sample_slot(from_slot)
+
+    # ── read ─────────────────────────────────────────────────────────────────
     def is_sample_slot_used(self, slot_nr: int) -> bool:
-        val = getattr(self.sample_placed, f"sample{slot_nr}").get(use_monitor=False)
-        return bool(val)
+        return bool(self._sig("used", slot_nr).get(use_monitor=False))
 
     def is_sample_in_gripper(self) -> bool:
-        val = self.sample_in_gripper.get(use_monitor=False)
-        return bool(val)
+        return self.is_sample_slot_used(GRIPPER_SLOT)
 
     def get_sample_name(self, slot_nr) -> str:
-        val = getattr(self.sample_names, f"sample{slot_nr}").get()
-        return unpack_desc(str(val))[0]
+        name = self._str(self._sig("name", slot_nr).get())
+        return name if name else EMPTY_NAME
 
     def get_sample_owner(self, slot_nr) -> str:
-        val = getattr(self.sample_names, f"sample{slot_nr}").get()
-        return unpack_desc(str(val))[1]
+        return self._str(self._sig("owner", slot_nr).get())
 
     def get_sample_name_and_owner(self, slot_nr) -> tuple:
-        val = getattr(self.sample_names, f"sample{slot_nr}").get()
-        return unpack_desc(str(val))
+        return (self.get_sample_name(slot_nr), self.get_sample_owner(slot_nr))
 
-    def show_all(self, measured: Optional[dict] = None):
-        """Print the tray grid plus loaded tray/gripper/stage samples.
+    def get_sample_status(self, slot_nr) -> str:
+        status = self._str(self._sig("status", slot_nr).get())
+        return status if status in STATUS_VALUES else STATUS_NONE
 
-        `measured`, if given, is a plain dict (as returned by
-        `_MeasuredSampleLog.snapshot()`) mapping packed
-        `pack_desc(name, owner)` strings to {"status": ..., ...} records --
-        this device class stays decoupled from the BEC-client-backed
-        measured-sample log that produces it; callers (e.g.
-        `Flomni.ftransfer_show_all()`) inject the snapshot instead.
+    def get_sample_meas_date(self, slot_nr) -> str:
+        return self._str(self._sig("meas_date", slot_nr).get())
 
-        Must be a plain, msgpack-serializable dict, not a callable: this
-        method runs via BEC's device-server RPC mechanism, which
-        serializes every argument before sending it, so a bound method
-        (an earlier version of this passed `self.measured_log.get`) fails
-        with "can not serialize 'method' object".
-        """
+    def get_sample_record(self, slot_nr) -> dict:
+        return {
+            "used": self.is_sample_slot_used(slot_nr),
+            "name": self.get_sample_name(slot_nr),
+            "owner": self.get_sample_owner(slot_nr),
+            "status": self.get_sample_status(slot_nr),
+            "meas_date": self.get_sample_meas_date(slot_nr),
+        }
 
-        def _measured_suffix(name: str, owner: str) -> str:
-            if not measured:
-                return ""
-            record = measured.get(pack_desc(name, owner))
-            if record is None:
-                return ""
-            return f"  [measured: {record['status']}]"
+    def show_all(self):
+        """Print the tray grid plus loaded tray/gripper/stage samples."""
+
+        def _suffix(slot_nr: int) -> str:
+            owner = self.get_sample_owner(slot_nr)
+            status = self.get_sample_status(slot_nr)
+            text = f"  (owner: {owner})" if owner else ""
+            if status != STATUS_NONE:
+                date = self.get_sample_meas_date(slot_nr)
+                text += f"  [measured: {status}{', ' + date if date else ''}]"
+            return text
 
         t = PrettyTable()
         t.title = "flOMNI sample storage"
@@ -141,21 +252,15 @@ class FlomniSampleStorage(Device):
             t.add_row(row)
         print(t)
         print("\n\nFollowing samples are currently loaded:\n")
-        for ct in range(1, 21):
+        for ct in TRAY_SLOTS:
             if self.is_sample_slot_used(ct):
-                name, owner = self.get_sample_name_and_owner(ct)
-                owner_suffix = f"  (owner: {owner})" if owner else ""
-                print(f"  Position {ct:2.0f}: {name}{owner_suffix}{_measured_suffix(name, owner)}")
-        if self.sample_in_gripper.get():
-            name, owner = unpack_desc(str(self.sample_in_gripper_name.get()))
-            owner_suffix = f"  (owner: {owner})" if owner else ""
-            print(f"\n  Gripper:        {name}{owner_suffix}{_measured_suffix(name, owner)}\n")
+                print(f"  Position {ct:2.0f}: {self.get_sample_name(ct)}{_suffix(ct)}")
+        if self.is_sample_in_gripper():
+            print(f"\n  Gripper:        {self.get_sample_name(GRIPPER_SLOT)}{_suffix(GRIPPER_SLOT)}\n")
         else:
             print(f"\n  Gripper:        no sample\n")
 
-        if self.is_sample_slot_used(0):
-            name, owner = self.get_sample_name_and_owner(0)
-            owner_suffix = f"  (owner: {owner})" if owner else ""
-            print(f"  flOMNI stage:   {name}{owner_suffix}{_measured_suffix(name, owner)}\n")
+        if self.is_sample_slot_used(STAGE_SLOT):
+            print(f"  flOMNI stage:   {self.get_sample_name(STAGE_SLOT)}{_suffix(STAGE_SLOT)}\n")
         else:
             print(f"  flOMNI stage:   no sample\n")

csaxs_bec/devices/sim/sim_sample_storage.py

@@ -1,7 +1,7 @@
 """
 Simulated flOMNI sample storage.
 
-`FlomniSampleStorage` is purely EPICS-backed (XOMNY-SAMPLE_DB_flomni* PVs). The simulated
+`FlomniSampleStorage` is purely EPICS-backed (FLOMNI-SAMPLE:<slot>:<field> PVs). The simulated
 version follows the patched-device pattern of `simulated_beamline_devices`: the real class
 is instantiated with all PVs replaced by in-memory mock PVs, so `set_sample_slot`,
 `show_all`, the gripper bookkeeping and the sample-transfer scripts run unchanged; the

omny_e2e_tests/sim_flomni_harness.py

@@ -417,36 +417,47 @@ def main():  # noqa: C901
         "slot 1 -> gripper bookkeeping",
     )
 
-    # --- 9b. sample owner (packed into the same DESC field as name) ------------------
+    # --- 9b. sample record: name, owner, measurement status (separate IOC fields) ----
     samples.set_sample_slot(2, "owner_test_sample", owner="mholler")
     check(
-        samples.get_sample_name(2) == "owner_test_sample", "get_sample_name returns plain name only"
+        samples.get_sample_name(2) == "owner_test_sample", "get_sample_name returns the name"
     )
-    check(samples.get_sample_owner(2) == "mholler", "get_sample_owner returns the packed owner")
+    check(samples.get_sample_owner(2) == "mholler", "get_sample_owner returns the owner")
     check(
         samples.get_sample_name_and_owner(2) == ("owner_test_sample", "mholler"),
         "get_sample_name_and_owner returns (name, owner)",
     )
+    check(samples.get_sample_status(2) == "none", "a new sample starts with status 'none'")
     samples.unset_sample_slot(2)
     check(
-        str(getattr(samples.sample_names, "sample2").get()) == "-",
-        "unset_sample_slot writes bare '-', not '- | <owner>'",
+        str(getattr(samples.sample_names, "sample2").get()) == "-"
+        and samples.get_sample_owner(2) == "",
+        "unset_sample_slot clears name ('-') and owner",
     )
-    # gripper move (slot -> gripper) must preserve owner, not just name. The
-    # gripper is a separate Cpt (sample_in_gripper_name), not part of the
-    # sample_names slot dict, so unpack it directly rather than via
-    # get_sample_name_and_owner (which only covers slots 0-20).
-    from csaxs_bec.devices.omny.sample_desc_codec import unpack_desc
-
+    # transfer slot -> gripper -> slot must carry the whole record (name, owner,
+    # measurement status and date), slot 100 = gripper
     samples.set_sample_slot(4, "transfer_test_sample", owner="e12345")
-    name, owner = samples.get_sample_name_and_owner(4)
-    samples.set_sample_in_gripper(name, owner=owner)
-    samples.unset_sample_slot(4)
+    samples.set_sample_measured(4, "completed", "2026-10")
+    samples.transfer_sample(4, 100)
     check(
-        unpack_desc(str(samples.sample_in_gripper_name.get())) == ("transfer_test_sample", "e12345"),
-        "owner survives a slot -> gripper move",
+        samples.is_sample_in_gripper()
+        and not samples.is_sample_slot_used(4)
+        and samples.get_sample_name_and_owner(100) == ("transfer_test_sample", "e12345")
+        and samples.get_sample_status(100) == "completed"
+        and samples.get_sample_meas_date(100) == "2026-10",
+        "slot -> gripper move carries name, owner, status and date",
     )
-    samples.unset_sample_in_gripper()
+    samples.transfer_sample(100, 5)
+    check(
+        not samples.is_sample_in_gripper() and samples.get_sample_status(5) == "completed",
+        "gripper -> slot move carries the status",
+    )
+    samples.unset_sample_slot(5)
+    # set_sample_measured fills an empty owner (label printing uses the owner as account)
+    samples.set_sample_slot(6, "no_owner_sample")
+    samples.set_sample_measured(6, "started", "2026-10", owner_if_empty="e99999")
+    check(samples.get_sample_owner(6) == "e99999", "owner_if_empty fills an empty owner")
+    samples.unset_sample_slot(6)
 
     frame_a = ids_cam.cam.get_image_data()
     frame_b = ids_cam.cam.get_image_data()

tests/tests_bec_ipython_client/test_flomni_tomo_angles.py

@@ -243,8 +243,6 @@ def make_flomni_for_golden_resume(monkeypatch, tomo_type: int) -> Flomni:
     obj.golden_max_number_of_projections = 21
     obj.tomo_id = -1
     obj.sample_get_name = lambda position: "test"  # backs the read-only sample_name property
-    obj.sample_get_measured_log_key = lambda position=0: "test|"
-    obj.measured_log = types.SimpleNamespace(mark_measured=lambda *a, **k: None)
     obj.special_angles = []
     obj._golden = _fake_golden
     obj._golden_equally_spaced = _fake_golden
@@ -270,6 +268,11 @@ def make_flomni_for_golden_resume(monkeypatch, tomo_type: int) -> Flomni:
     )
     monkeypatch.setitem(builtins.__dict__, "bec", fake_bec)
     monkeypatch.setitem(builtins.__dict__, "scans", _FakeScans())
+    # tomo_scan() records the measurement status on the sample record (stage slot 0)
+    fake_dev = types.SimpleNamespace(
+        flomni_samples=types.SimpleNamespace(set_sample_measured=lambda *a, **k: None)
+    )
+    monkeypatch.setitem(builtins.__dict__, "dev", fake_dev)
     return obj