fix(omny): parking overwrite guard, active-slot leica targets, shuttle-scoped print; add measured-status+account encoding and P-touch label printing

- Widget: reject assigning a shuttle to a parking slot already held by a
  different shuttle ("Slot in use", clear it first) instead of silently
  overwriting.
- Widget: PDF report now shows only the 3 shuttles + their samples (title
  "OMNY Shuttle Samples") -- stage/gripper/fixed-O positions can't
  physically leave the vacuum chamber, so a printed report has nothing
  useful to say about them. The on-screen widget still shows/manages all
  of it.
- Widget: fixed the active-parking-slot badge never showing for a
  loadlock-driven park. Root cause: otransfer_park_loadlock_slot() drives
  oparkz to a different target-position table (base -6, "leica") than the
  regular otransfer_park_slot() (base -0.9); the widget only recognized
  the regular one. Now checks both.
- OMNYSampleStorage: measured status now packs status *and* the beamtime
  account into one VAL -- 0 = not measured, negative 5-digit = started,
  positive 5-digit = completed, falling back to FALLBACK_ACCOUNT (11111)
  for anything that doesn't parse as exactly 5 digits. No IOC schema
  change needed (same ai record). New get_sample_account()(+gripper/
  samplestage variants); set_sample_status() now takes an account param;
  show_all()/help() updated. otransfer_* CLI wrappers updated to match.
- Widget: new P-touch label-print offer, but only when clearing a
  *shuttle* slot (not fixed-O/stage/gripper -- those are bookkeeping
  corrections, not physical sample removal, since only shuttle samples
  leave the vacuum chamber). Uses a separate PTouchLabelPrinter instance
  at template 3 (not yet bench-tested against a real printer, unlike
  FlOMNI's confirmed-working template 1), same field names as FlOMNI.
- AI_docs/OPEN_ISSUES.md: noted that the measured-status+account encoding
  should be ported to FlOMNI at some point (separate task, not this
  branch's scope -- FlOMNI's own status tracking is untouched, still the
  redis-based _MeasuredSampleLog).
- New tests: encode/decode round-trips + fallback behavior for the
  status+account VAL scheme, plus oparkz leica-target coverage.

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

csaxs_bec/bec_ipython_client/plugins/omny/AI_docs/OPEN_ISSUES.md

@@ -794,6 +794,26 @@ where OMNY's version was ported from, so it inherited the bug faithfully. **Not
 (`update-omny`); flagging for a separate, dedicated look rather than bundling an unreviewed
 flomni-affecting change into this branch.
 
+## Measured-status + account tracking now real EPICS for OMNY -- port to flomni later
+
+OMNY sample storage gained a real EPICS channel for measured status and the beamtime account
+(`OMNYSampleStorage`'s new `sample_status`/`sample_shuttle_{A,B,C}_status` signals, one new `ai`
+record per sample position -- see `omny_samples_status.template` in `OMNY/OMNY.git` and
+`OMNY/ioc-samples-and-xrayeye`). Encoding: VAL packs *both* status and account into one number --
+`0` means not measured; otherwise VAL's magnitude is the 5-digit account and its sign is the
+status (negative = "started", positive = "completed"), falling back to `FALLBACK_ACCOUNT` (11111)
+for anything that doesn't parse as exactly 5 digits. See `OMNYSampleStorage.STATUS_NONE`'s class
+comment and `_status_to_val()`/`_val_to_status()`/`_val_to_account()`/`_normalize_account()` in
+`csaxs_bec/devices/omny/omny_sample_storage.py`.
+
+This deliberately mirrors, but does not replace, FlOMNI's own measured-status tracking, which
+still lives entirely in a BEC redis global var (`_MeasuredSampleLog`/`flomni_measured_samples` in
+`sample_storage.py`/`flomni.py`), keyed by `pack_desc(name, owner)` rather than a fixed physical
+position -- FlOMNI wasn't touched this branch (explicitly out of scope, see the sample-storage
+work above). **Mirko's intent: port the same real-EPICS status+account channel to FlOMNI at some
+point** -- a separate, dedicated task (its own `.template` record + `FlomniSampleStorage` changes
++ retiring/reconciling `_MeasuredSampleLog`), not bundled into this branch.
+
 ## Environment note (not code, but will bite again if forgotten)
 
 Both `csaxs_bec` and `bec_widgets` were pip-installed editable pointing at pre-repo-

csaxs_bec/bec_ipython_client/plugins/omny/omny_sample_transfer_mixin.py

@@ -1170,30 +1170,39 @@ class OMNYSampleTransferMixin:
     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_set_sample_status(self, container: str, slot_nr: int, status: str, account=None) -> bool:
+        return dev.omny_samples.set_sample_status(container, slot_nr, status, account)
 
     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_account(self, container: str, slot_nr: int):
+        return dev.omny_samples.get_sample_account(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_set_sample_status_in_gripper(self, status: str, account=None) -> bool:
+        return dev.omny_samples.set_sample_status_in_gripper(status, account)
 
     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_account_in_gripper(self):
+        return dev.omny_samples.get_sample_account_in_gripper()
+
+    def otransfer_set_sample_status_in_samplestage(self, status: str, account=None) -> bool:
+        return dev.omny_samples.set_sample_status_in_samplestage(status, account)
 
     def otransfer_get_sample_status_in_samplestage(self) -> str:
         return dev.omny_samples.get_sample_status_in_samplestage()
 
+    def otransfer_get_sample_account_in_samplestage(self):
+        return dev.omny_samples.get_sample_account_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

@@ -24,11 +24,11 @@ OMNY (``dev.omny_samples``)
     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/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.
+    dedicated owner channel and a measured-status channel (packing status
+    *and* account into one VAL -- see ``OMNY_SampleStorage.STATUS_NONE``)
+    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
@@ -157,10 +157,13 @@ COLOR_SLOT_BORDER = "#c0c0c0"
 # already-transferred template (confirmed working) -- NOT the generic
 # Text1/Text2/Text6 names that OMNY_shared/flomni.lbx's own objectName
 # attributes happen to use for its (different) template source file.
-# FlOMNI-only: OMNY has no measured-sample-log integration (see module
-# docstring), so it never drives this printer.
+# Same printer, same field names, but a *different* template number is
+# pre-loaded on the printer for the OMNY label -- not yet bench-tested
+# against a real printer the way FlOMNI's template 1 has been, so the
+# field names are an assumption for now, not confirmed working.
 PTOUCH_PRINTER_HOST = "BRN94DDF8AAB8EC.psi.ch"
 PTOUCH_TEMPLATE_NUMBER = 1
+PTOUCH_TEMPLATE_NUMBER_OMNY = 3
 PTOUCH_FIELD_ACCOUNT = "ACCOUNT"
 PTOUCH_FIELD_DATE = "DATE"
 PTOUCH_FIELD_SAMPLENAME = "SAMPLENAME"
@@ -197,20 +200,38 @@ _OPARKZ_TARGET_POSITIONS: dict[int, float] = {
     5: -0.9 - 29,
     6: -0.9,
 }
+# otransfer_park_loadlock_slot()'s targets (_oparkz_leica_slot() in the same
+# mixin) -- the *same* slot numbers, reached via the loadlock/leica transfer
+# path instead of the regular gripper path, at a different oparkz offset
+# (base -6, not -0.9). Both genuinely mean "the parking stage is at that
+# slot" -- "active" must recognize either, or a loadlock-driven park (like
+# otransfer_park_loadlock_slot()) never shows as active in the GUI even
+# though it physically is.
+_OPARKZ_LEICA_TARGET_POSITIONS: dict[int, float] = {
+    2: -6 - 29 * 3 - 42 - 0.05,
+    3: -6 - 29 * 3,
+    4: -6 - 29 * 2,
+    5: -6 - 29,
+    6: -6,
+}
 _OPARKZ_TOLERANCE = 0.05
 
 
 def oparkz_active_parking_slot(readback: float, tol: float = _OPARKZ_TOLERANCE) -> Optional[int]:
     """Return which parking slot (2-6) the oparkz motor is currently sitting
     at, i.e. ready for a sample transfer, or None if it's between positions
-    (or at slot 1, which has no drivable transfer target). Pure function --
-    see the module-level docstring above ``_OPARKZ_TARGET_POSITIONS`` for
-    why this duplicates rather than imports omny_sample_transfer_mixin.py's
-    ``_oparkz_slot_check()``.
+    (or at slot 1, which has no drivable transfer target). Checks both the
+    regular-gripper and loadlock/leica target tables (see
+    ``_OPARKZ_LEICA_TARGET_POSITIONS``) since either represents the parking
+    stage genuinely being at that slot. Pure function -- see the
+    module-level docstring above ``_OPARKZ_TARGET_POSITIONS`` for why this
+    duplicates rather than imports omny_sample_transfer_mixin.py's
+    ``_oparkz_slot_check()``/``_oparkz_leica_slot()``.
     """
-    for slot, target in _OPARKZ_TARGET_POSITIONS.items():
-        if abs(readback - target) < tol:
-            return slot
+    for targets in (_OPARKZ_TARGET_POSITIONS, _OPARKZ_LEICA_TARGET_POSITIONS):
+        for slot, target in targets.items():
+            if abs(readback - target) < tol:
+                return slot
     return None
 
 
@@ -504,6 +525,9 @@ class OMNY_SampleStorage(BECWidget, QWidget):
         self.label_printer = PTouchLabelPrinter(
             host=PTOUCH_PRINTER_HOST, template_number=PTOUCH_TEMPLATE_NUMBER
         )
+        self.label_printer_omny = PTouchLabelPrinter(
+            host=PTOUCH_PRINTER_HOST, template_number=PTOUCH_TEMPLATE_NUMBER_OMNY
+        )
         self._setup = self._detect_setup()
         self._build_ui()
         self._poll_timer = QTimer(self)
@@ -634,9 +658,9 @@ class OMNY_SampleStorage(BECWidget, QWidget):
         sample-kind slot -- stage, gripper, the 10 fixed O positions, and
         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``.
+        placed/name signals -- see that class's status+account VAL
+        encoding, ``STATUS_NONE``), unlike FlOMNI which packs name+owner
+        into one field via ``sample_desc_codec``.
         """
 
         def _val(key, default=None):
@@ -646,13 +670,19 @@ class OMNY_SampleStorage(BECWidget, QWidget):
             return default
 
         def _status_str(raw) -> Optional[str]:
-            # mirrors OMNYSampleStorage._status_val_to_str(), without a live
+            # mirrors OMNYSampleStorage._val_to_status(), without a live
             # device round-trip: this widget only has the bulk-read dict.
+            # VAL packs status+account (see that class's status+account VAL
+            # encoding comment): 0 -> no badge, negative -> "started",
+            # positive -> "completed" (the magnitude is the account, not
+            # needed here -- see _status_account() for that).
             try:
                 val = int(round(float(raw)))
             except (TypeError, ValueError):
                 return None
-            return {1: "started", 2: "completed"}.get(val)  # 0 ("none") -> None: no badge
+            if val == 0:
+                return None
+            return "started" if val < 0 else "completed"
 
         state: dict[tuple, tuple[bool, str, str, Optional[str]]] = {}
 
@@ -788,6 +818,74 @@ class OMNY_SampleStorage(BECWidget, QWidget):
             if reply != QMessageBox.StandardButton.Retry:
                 break
 
+    def _offer_print_label_for_previous_occupant_omny(self, key: tuple) -> None:
+        """If the shuttle slot about to be cleared has a measured status
+        (started or completed), offer to print an account/date/name label
+        for it -- OMNY's counterpart to
+        ``_offer_print_label_for_previous_occupant()``, but only ever
+        called for shuttle-slot clears (see ``_write_slot_omny``): unlike
+        the fixed-O/stage/gripper positions, a shuttle's samples are the
+        only ones that can physically leave the vacuum chamber, so
+        printing a label only makes sense there.
+
+        Uses ``self.label_printer_omny`` (``PTOUCH_TEMPLATE_NUMBER_OMNY``),
+        a different template number than FlOMNI's -- same field names
+        (ACCOUNT/DATE/SAMPLENAME) by assumption, not yet bench-tested
+        against a real printer the way FlOMNI's template 1 is.
+
+        OMNY has no FlOMNI-style measured-sample log to pop a record from
+        -- status and account already live directly on the slot itself
+        (see ``OMNYSampleStorage``'s status+account VAL encoding), so this
+        reads them fresh from the device right before the clear, instead.
+        There's also no stored measurement date (OMNY doesn't track one),
+        so the label's DATE field is simply today's date, at print time.
+        """
+        assert key[0] == "shuttle"
+        _, letter, n = key
+        try:
+            was_used = self._samples.is_sample_slot_used(letter, n)
+            prev_name = self._samples.get_sample_name(letter, n)
+            status = self._samples.get_sample_status(letter, n)
+            account = self._samples.get_sample_account(letter, n)
+        except Exception as exc:
+            logger.warning(
+                f"OMNY_SampleStorage: could not read previous state of slot {key} for the"
+                f" measured-label check: {exc}"
+            )
+            return
+        if not was_used or status == "none":
+            return
+        reply = QMessageBox.question(
+            self,
+            "Print label?",
+            f"Sample '{prev_name}' was measured ({status}), account {account}. Print a label?",
+            QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
+        )
+        if reply != QMessageBox.StandardButton.Yes:
+            return
+        fields = {
+            PTOUCH_FIELD_ACCOUNT: str(account),
+            PTOUCH_FIELD_DATE: datetime.date.today().strftime("%Y-%m-%d"),
+            PTOUCH_FIELD_SAMPLENAME: prev_name,
+        }
+        # Retry loop, same reasoning as _offer_print_label_for_previous_occupant():
+        # the slot is about to be cleared regardless of whether printing
+        # succeeds, so give the operator a real chance to fix the printer
+        # and retry rather than losing the label silently.
+        while not self.label_printer_omny.print_label(fields):
+            reply = QMessageBox.question(
+                self,
+                "Print failed",
+                f"Could not print label for '{prev_name}' — printer unreachable.\n\n"
+                "Power it on / check it has tape loaded, then Retry, or Ignore to"
+                " give up on printing this label.\n\nThe slot will be updated"
+                " regardless, once you close this dialog.",
+                QMessageBox.StandardButton.Retry | QMessageBox.StandardButton.Ignore,
+                QMessageBox.StandardButton.Retry,
+            )
+            if reply != QMessageBox.StandardButton.Retry:
+                break
+
     def _write_slot(self, slot: int, used: int, name: str, owner: str = "") -> bool:
         """
         Write (used, name, owner) to a FlOMNI storage slot 0–20 or the
@@ -826,13 +924,21 @@ class OMNY_SampleStorage(BECWidget, QWidget):
         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).
+        rename) or a cleared slot isn't mid-measurement.
+
+        Offers a P-touch label print for the previous occupant only when
+        *clearing* a *shuttle* slot (see
+        ``_offer_print_label_for_previous_occupant_omny``) -- shuttles are
+        the only OMNY samples that can physically leave the vacuum
+        chamber, so that's the only "sample remove" this widget can
+        meaningfully print a label for; every other clear (fixed-O/stage/
+        gripper) is a bookkeeping correction, not a physical removal.
         """
+        kind = key[0]
+        if used == 0 and kind == "shuttle":
+            self._offer_print_label_for_previous_occupant_omny(key)
         try:
             status_none = OMNY_STATUS_NONE
-            kind = key[0]
             if kind == "stage":
                 self._samples.sample_in_samplestage.set(used)
                 self._samples.sample_in_samplestage_name.set(name)
@@ -1224,11 +1330,18 @@ class OMNY_SampleStorage(BECWidget, QWidget):
 
     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
-        previous occupant, if any, without clearing wherever `letter` was
-        previously parked -- mirrors ``set_shuttle_slot()``'s own
-        semantics exactly; keeping the two in sync is operator discipline,
-        same as everywhere else in this widget (see module docstring).
+        ``_ShuttleStatusLabel``'s context menu. Rejects the assignment if
+        that slot is already assigned to a *different* shuttle -- unlike
+        ``set_shuttle_slot()`` itself, which would silently overwrite it
+        (see module docstring on why writes elsewhere in this widget don't
+        add interlocks: parking is the one exception, since two different
+        shuttles both being "in" the same slot is a bookkeeping
+        impossibility, not a legitimate concurrent-edit race). Re-assigning
+        the *same* shuttle to a slot it's already in is a no-op-safe
+        confirm, not an error.
+
+        Doesn't clear wherever `letter` was previously parked, if anywhere
+        else -- still the operator's job, same as everywhere else here.
         """
         slot_nr, ok = QInputDialog.getInt(
             self,
@@ -1241,6 +1354,19 @@ class OMNY_SampleStorage(BECWidget, QWidget):
         )
         if not ok:
             return
+        data = self._bulk_read_omny()
+        parking_assignment = self._parse_parking_assignment_omny(data) if data is not None else {}
+        occupant = parking_assignment.get(slot_nr)
+        if occupant is not None and occupant != letter:
+            QMessageBox.warning(
+                self,
+                "Slot in use",
+                f"Parking slot {slot_nr} is already assigned to shuttle {occupant}.\n\n"
+                f"Clear shuttle {occupant}'s parking assignment first (right-click its "
+                f"status line → Clear parking assignment), then assign shuttle "
+                f"{letter} to slot {slot_nr}.",
+            )
+            return
         if self._write_parking_omny(slot_nr, letter):
             self.refresh()
 
@@ -1358,10 +1484,16 @@ class OMNY_SampleStorage(BECWidget, QWidget):
         return path
 
     def _build_report_pdf_omny(self) -> str:
-        """Render a page mirroring the on-screen OMNY layout (stage +
-        gripper header, fixed-O grid, 3 shuttle sections with their
-        installed/active status text) from a fresh device read, and return
-        the path of the temp PDF file it was written to.
+        """Render a page with just the 3 shuttles and their samples (status
+        line + slot grid) from a fresh device read, and return the path of
+        the temp PDF file it was written to.
+
+        Deliberately excludes the stage/gripper/fixed-O positions that the
+        on-screen widget also shows -- unlike the shuttles, none of those
+        samples can physically leave the vacuum chamber, so a printed
+        report (meant to travel with a shuttle's physical samples) has
+        nothing useful to say about them. The widget itself still displays
+        and manages all of it; only this printed report is scoped down.
         """
         from fpdf import XPos, YPos
 
@@ -1375,7 +1507,7 @@ class OMNY_SampleStorage(BECWidget, QWidget):
         with PDFWriter(path) as file:
             pdf = file._pdf
             pdf.set_font("Helvetica", "B", 16)
-            pdf.cell(0, 10, "OMNY Sample Storage", new_x=XPos.LMARGIN, new_y=YPos.NEXT, align="C")
+            pdf.cell(0, 10, "OMNY Shuttle Samples", new_x=XPos.LMARGIN, new_y=YPos.NEXT, align="C")
             pdf.set_font("Helvetica", "", 10)
             timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
             pdf.cell(0, 6, timestamp, new_x=XPos.LMARGIN, new_y=YPos.NEXT, align="C")
@@ -1383,27 +1515,10 @@ class OMNY_SampleStorage(BECWidget, QWidget):
 
             margin = pdf.l_margin
             usable_w = pdf.w - 2 * margin
-            cell_w = usable_w / 5
             cell_h = 28
 
-            # header: stage + gripper, same widths as two fixed-O cells
-            y0 = pdf.get_y()
-            self._draw_report_cell(pdf, margin, y0, cell_w, cell_h, ("stage",), state.get(("stage",)))
-            self._draw_report_cell(
-                pdf, margin + cell_w, y0, cell_w, cell_h, ("gripper",), state.get(("gripper",))
-            )
-
-            # fixed O positions, same 5-wide grid as the on-screen layout
-            fixed_y0 = y0 + cell_h + 4
-            for idx, n in enumerate(OMNY_FIXED_POSITIONS):
-                r, c = divmod(idx, 5)
-                x = margin + c * cell_w
-                y = fixed_y0 + r * cell_h
-                key = ("O", n)
-                self._draw_report_cell(pdf, x, y, cell_w, cell_h, key, state.get(key))
-
             # shuttles A/B/C: status line + 3-wide slot grid, side by side
-            shuttle_y0 = fixed_y0 + 2 * cell_h + 8
+            shuttle_y0 = pdf.get_y()
             shuttle_w = usable_w / 3
             for si, letter in enumerate(OMNY_SHUTTLE_LETTERS):
                 x0 = margin + si * shuttle_w

csaxs_bec/devices/omny/omny_sample_storage.py

@@ -1,4 +1,5 @@
 import time
+from typing import Optional
 
 from ophyd import Component as Cpt
 from ophyd import Device
@@ -27,13 +28,16 @@ class OMNYSampleStorage(Device):
         "get_sample_owner_in_gripper",
         "set_sample_status_in_samplestage",
         "get_sample_status_in_samplestage",
+        "get_sample_account_in_samplestage",
         "set_sample_status_in_gripper",
         "get_sample_status_in_gripper",
+        "get_sample_account_in_gripper",
         "get_sample_name",
         "get_sample_owner",
         "get_sample_name_and_owner",
         "set_sample_status",
         "get_sample_status",
+        "get_sample_account",
         "is_sample_in_samplestage",
         "set_shuttle_slot",
         "unset_shuttle_slot",
@@ -46,17 +50,26 @@ 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.
+    # Measured status *and* account, packed into one number -- 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 STATUS record follows the same
+    # convention rather than introducing a new record type used nowhere
+    # else here.
+    #
+    # Encoding: VAL == 0 means "not measured"; otherwise VAL's magnitude is
+    # the 5-digit beamtime account and its sign is the status -- negative
+    # while a measurement is in progress ("started"), positive once it's
+    # done ("completed"). This packs status+account into the one VAL this
+    # IOC's "ai"-record convention already gives us, with no new EPICS
+    # record needed. An account that isn't exactly 5 digits (e.g. a
+    # non-numeric test account) falls back to FALLBACK_ACCOUNT (11111) --
+    # safe because real account numbers are already above 20000, so this
+    # sentinel can never collide with a real one, and only ever shows up
+    # for test accounts.
     STATUS_NONE = 0
-    STATUS_STARTED = 1
-    STATUS_COMPLETED = 2
-    _STATUS_LABELS = {STATUS_NONE: "none", STATUS_STARTED: "started", STATUS_COMPLETED: "completed"}
+    FALLBACK_ACCOUNT = 11111
 
     sample_shuttle_A_placed = {
         f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_shuttle_A:{i}", {"auto_monitor": True}) for i in range(1, 7)
@@ -119,11 +132,11 @@ 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.
+    # Measured status+account (VAL) + owner (DESC) -- one new record per
+    # sample position, added alongside the existing placed/name records
+    # (see the status+account VAL encoding described 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)
@@ -242,27 +255,53 @@ class OMNYSampleStorage(Device):
         self._run_subs(sub_type=self.SUB_VALUE, timestamp=timestamp, obj=self)
 
     @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:
+    def _normalize_account(cls, account) -> int:
+        """A positive 5-digit account number for the status+account VAL
+        encoding (see the class docstring comment above), or
+        FALLBACK_ACCOUNT if `account` doesn't parse as exactly 5 digits.
+        """
+        digits = "".join(ch for ch in str(account) if ch.isdigit()) if account is not None else ""
+        if len(digits) == 5:
+            val = int(digits)
+            if 10000 <= val <= 99999:
                 return val
-        raise OMNYSampleStorageError(
-            f"Invalid status {status!r}; expected one of {sorted(cls._STATUS_LABELS.values())}."
-        )
+        return cls.FALLBACK_ACCOUNT
 
     @classmethod
-    def _status_val_to_str(cls, val) -> str:
+    def _status_to_val(cls, status: str, account=None) -> int:
+        status = (status or "none").strip().lower()
+        if status == "none":
+            return 0
+        if status not in ("started", "completed"):
+            raise OMNYSampleStorageError(
+                f"Invalid status {status!r}; expected one of ['none', 'started', 'completed']."
+            )
+        acc = cls._normalize_account(account)
+        return -acc if status == "started" else acc
+
+    @classmethod
+    def _val_to_status(cls, val) -> str:
         try:
-            return cls._STATUS_LABELS.get(int(round(float(val))), "none")
+            v = int(round(float(val)))
         except (TypeError, ValueError):
             return "none"
+        if v == 0:
+            return "none"
+        return "started" if v < 0 else "completed"
+
+    @classmethod
+    def _val_to_account(cls, val) -> Optional[int]:
+        try:
+            v = int(round(float(val)))
+        except (TypeError, ValueError):
+            return None
+        return abs(v) if v != 0 else 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")
+        status = self._status_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)
@@ -288,7 +327,7 @@ class OMNYSampleStorage(Device):
         if slot_nr > 20:
             raise OMNYSampleStorageError(f"Invalid slot number {slot_nr}.")
 
-        status = self._status_str_to_val("none")
+        status = self._status_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("-")
@@ -310,8 +349,8 @@ class OMNYSampleStorage(Device):
             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)
+    def set_sample_status(self, container: str, slot_nr: int, status: str, account=None) -> bool:
+        val = self._status_to_val(status, account)
         if container == "A":
             getattr(self.sample_shuttle_A_status, f"sample{slot_nr}").set(val)
         elif container == "B":
@@ -332,7 +371,24 @@ class OMNYSampleStorage(Device):
             val = getattr(self.sample_status, f"sample{slot_nr}").get()
         else:
             return "none"
-        return self._status_val_to_str(val)
+        return self._val_to_status(val)
+
+    def get_sample_account(self, container: str, slot_nr: int) -> Optional[int]:
+        """The account packed into this slot's status VAL (see the
+        status+account encoding described above ``STATUS_NONE``), or None
+        if the slot isn't measured ("none" status).
+        """
+        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._val_to_account(val)
 
     def get_sample_owner(self, container: str, slot_nr: int) -> str:
         if container == "A":
@@ -366,25 +422,25 @@ class OMNYSampleStorage(Device):
         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"))
+        self.sample_in_gripper_status.set(self._status_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"))
+        self.sample_in_gripper_status.set(self._status_to_val("none"))
 
     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"))
+        self.sample_in_samplestage_status.set(self._status_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"))
+        self.sample_in_samplestage_status.set(self._status_to_val("none"))
 
     def get_sample_owner_in_gripper(self) -> str:
         val = self.sample_in_gripper_owner.get()
@@ -394,17 +450,23 @@ class OMNYSampleStorage(Device):
         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 set_sample_status_in_gripper(self, status: str, account=None) -> bool:
+        self.sample_in_gripper_status.set(self._status_to_val(status, account))
 
     def get_sample_status_in_gripper(self) -> str:
-        return self._status_val_to_str(self.sample_in_gripper_status.get())
+        return self._val_to_status(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_account_in_gripper(self) -> Optional[int]:
+        return self._val_to_account(self.sample_in_gripper_status.get())
+
+    def set_sample_status_in_samplestage(self, status: str, account=None) -> bool:
+        self.sample_in_samplestage_status.set(self._status_to_val(status, account))
 
     def get_sample_status_in_samplestage(self) -> str:
-        return self._status_val_to_str(self.sample_in_samplestage_status.get())
+        return self._val_to_status(self.sample_in_samplestage_status.get())
+
+    def get_sample_account_in_samplestage(self) -> Optional[int]:
+        return self._val_to_account(self.sample_in_samplestage_status.get())
 
     def is_sample_slot_used(self, container, slot_nr: int) -> bool:
         if container == "A":
@@ -473,9 +535,11 @@ class OMNYSampleStorage(Device):
         status = self.get_sample_status(container, slot_nr)
         suffix = f" ({owner})" if owner else ""
         if status == "started":
-            suffix += "  ◐ started"
+            account = self.get_sample_account(container, slot_nr)
+            suffix += f"  ◐ started (acc {account})"
         elif status == "completed":
-            suffix += "  ✓ completed"
+            account = self.get_sample_account(container, slot_nr)
+            suffix += f"  ✓ completed (acc {account})"
         return name + suffix
 
     def show_all(self):
@@ -508,9 +572,9 @@ class OMNYSampleStorage(Device):
             status = self.get_sample_status_in_samplestage()
             suffix = f" ({owner})" if owner else ""
             if status == "started":
-                suffix += "  ◐ started"
+                suffix += f"  ◐ started (acc {self.get_sample_account_in_samplestage()})"
             elif status == "completed":
-                suffix += "  ✓ completed"
+                suffix += f"  ✓ completed (acc {self.get_sample_account_in_samplestage()})"
             print(f"\n\n  Sample stage:   {name}{suffix}")
         else:
             print(f"\n\n  Sample stage:   no sample")
@@ -544,9 +608,12 @@ class OMNYSampleStorage(Device):
         print(" dev.omny_samples.set_sample_slot('system',position,'name','owner')")
         print("   system can be A, B, C, O")
         print(" dev.omny_samples.get_sample_owner('system',position)")
-        print(" dev.omny_samples.set_sample_status('system',position,'status')")
+        print(" dev.omny_samples.set_sample_status('system',position,'status',account)")
         print("   status is one of 'none', 'started', 'completed'")
+        print("   account is required for 'started'/'completed' (a 5-digit account number,")
+        print("   falls back to 11111 if not exactly 5 digits)")
         print(" dev.omny_samples.get_sample_status('system',position)")
+        print(" dev.omny_samples.get_sample_account('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)")

tests/tests_bec_ipython_client/test_omny_sample_storage_widget_math.py

@@ -31,6 +31,26 @@ def test_oparkz_active_parking_slot_matches_each_target(readback, expected_slot)
     assert oparkz_active_parking_slot(readback) == expected_slot
 
 
+@pytest.mark.parametrize(
+    "readback,expected_slot",
+    [
+        (-6, 6),
+        (-6 - 29, 5),
+        (-6 - 29 * 2, 4),
+        (-6 - 29 * 3, 3),
+        (-6 - 29 * 3 - 42 - 0.05, 2),
+    ],
+)
+def test_oparkz_active_parking_slot_matches_each_leica_target(readback, expected_slot):
+    """otransfer_park_loadlock_slot()'s targets (_oparkz_leica_slot(), a
+    different oparkz offset than the regular otransfer_park_slot() table)
+    must be recognized as "active" too -- a real bug: the GUI never showed
+    the active badge after a loadlock-driven park, since only the regular
+    target table was checked.
+    """
+    assert oparkz_active_parking_slot(readback) == expected_slot
+
+
 def test_oparkz_active_parking_slot_none_when_far_from_any_target():
     assert oparkz_active_parking_slot(0.0) is None
 

tests/tests_devices/test_omny_sample_storage.py

@@ -0,0 +1,47 @@
+import pytest
+
+from csaxs_bec.devices.omny.omny_sample_storage import OMNYSampleStorage, OMNYSampleStorageError
+
+
+def test_status_none_encodes_to_zero():
+    assert OMNYSampleStorage._status_to_val("none") == 0
+    assert OMNYSampleStorage._val_to_status(0) == "none"
+    assert OMNYSampleStorage._val_to_account(0) is None
+
+
+@pytest.mark.parametrize("account", [20123, "20123", "e20123"])
+def test_started_encodes_negative_account(account):
+    val = OMNYSampleStorage._status_to_val("started", account)
+    assert val == -20123
+    assert OMNYSampleStorage._val_to_status(val) == "started"
+    assert OMNYSampleStorage._val_to_account(val) == 20123
+
+
+@pytest.mark.parametrize("account", [20123, "20123", "e20123"])
+def test_completed_encodes_positive_account(account):
+    val = OMNYSampleStorage._status_to_val("completed", account)
+    assert val == 20123
+    assert OMNYSampleStorage._val_to_status(val) == "completed"
+    assert OMNYSampleStorage._val_to_account(val) == 20123
+
+
+@pytest.mark.parametrize("bad_account", [None, "", "not-a-number", "123", "123456", "e-test"])
+def test_unparseable_account_falls_back_to_sentinel(bad_account):
+    val = OMNYSampleStorage._status_to_val("started", bad_account)
+    assert OMNYSampleStorage._val_to_account(val) == OMNYSampleStorage.FALLBACK_ACCOUNT
+    assert OMNYSampleStorage.FALLBACK_ACCOUNT == 11111
+
+
+def test_invalid_status_raises():
+    with pytest.raises(OMNYSampleStorageError):
+        OMNYSampleStorage._status_to_val("bogus")
+
+
+@pytest.mark.parametrize("raw", ["not-a-number", None, object()])
+def test_val_to_status_defaults_to_none_on_bad_input(raw):
+    assert OMNYSampleStorage._val_to_status(raw) == "none"
+
+
+@pytest.mark.parametrize("raw", ["not-a-number", None, object()])
+def test_val_to_account_defaults_to_none_on_bad_input(raw):
+    assert OMNYSampleStorage._val_to_account(raw) is None