OMNY sample storage on the OMNY soft IOC records (OMNY-SAMPLE:<slot>:...)
OMNYSampleStorage follows FlomniSampleStorage: one record per slot (USED/NAME/OWNER/STATUS/MEAS_DATE; slots A1..C6, P10..P101, STAGE, GRIPPER, PARK1..6), served by OMNY's caproto IOC (OMNY repo preemt_rt, OMNYtools/softioc/ioc_omny.py). Replaces the XOMNY-SAMPLE_DB_* records and the status x account VAL packing. - device: slot_name(), transfer_sample() (whole record moves), set_sample_measured(), get_sample_record()/get_sample_meas_date(); the (container, slot_nr) API is kept; the 0-20 slot check that rejected 32-34/ 100/101 is gone. - transfer mixin: picks/puts use transfer_sample(); otransfer_* wrappers for the new methods, the account wrappers removed. - omny.py tomo_scan(): started/completed + month on the STAGE record (as flOMNI). - sample_storage widget: bulk-read parser and writes on the new records; label prints owner as account and the stored date (as flOMNI). - tomo_params: sample name from sample_names.STAGE. - SimOMNYSampleStorage, tests, user docs. Checked against a local ioc_omny.py (real EPICS, localhost): long names, status enum, transfers, parking. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TNtKEk4jAyjebMDQFwRHJQ
This commit is contained in:
Mirko HollerandClaude Opus 5.5 committed
1 parent
f53a5ffb13
commit
89ba28550c
14 files changed
+508
-767
No files matched your search
@@ -816,6 +816,12 @@ work above). **Mirko's intent: port the same real-EPICS status+account channel t
|
||||
point** -- a separate, dedicated task (its own `.template` record + `FlomniSampleStorage` changes
|
||||
+ retiring/reconciling `_MeasuredSampleLog`), not bundled into this branch.
|
||||
|
||||
**Superseded 2026-10-18 (branch `feat/omny-sample-ioc`):** both instruments now keep one
|
||||
soft-IOC record per slot (`FLOMNI-SAMPLE:<slot>:...` since 2026-10-17, `OMNY-SAMPLE:<slot>:USED/
|
||||
NAME/OWNER/STATUS/MEAS_DATE` on OMNY's caproto IOC `ioc_omny.py`); the status is an enum plus a
|
||||
date, the status x account packing above and `_MeasuredSampleLog` are gone, `transfer_sample()`
|
||||
moves the whole record, and `tomo_scan()` sets started/completed on the STAGE record (as flOMNI).
|
||||
|
||||
## Reference images now saved at every manual confirm gate in otransfer_get_sample/put_sample -- flomni has no equivalent gate to save at
|
||||
|
||||
`_otransfer_save_reference_images()` used to be called only from inside `_otransfer_confirm()`'s
|
||||
|
||||
@@ -1365,6 +1365,14 @@ class OMNY(
|
||||
self.write_pdf_report()
|
||||
else:
|
||||
self.tomo_id = 0
|
||||
# measurement status on the sample record (sample stage) in the IOC
|
||||
dev.omny_samples.set_sample_measured(
|
||||
"STAGE",
|
||||
0,
|
||||
"started",
|
||||
datetime.date.today().strftime("%Y-%m"),
|
||||
owner_if_empty=bec.active_account or "",
|
||||
)
|
||||
self.progress["tomo_start_time"] = datetime.datetime.now().isoformat()
|
||||
# reset stale estimates from any previous scan, otherwise the GUI
|
||||
# would keep showing a leftover ETA from before this scan has
|
||||
@@ -1551,6 +1559,14 @@ class OMNY(
|
||||
self.progress["subtomo_projection"] = self.progress["subtomo_total_projections"]
|
||||
self._print_progress()
|
||||
self.OMNYTools.printgreenbold("Tomoscan finished")
|
||||
# measurement status on the sample record (sample stage) in the IOC
|
||||
dev.omny_samples.set_sample_measured(
|
||||
"STAGE",
|
||||
0,
|
||||
"completed",
|
||||
datetime.date.today().strftime("%Y-%m"),
|
||||
owner_if_empty=bec.active_account or "",
|
||||
)
|
||||
|
||||
idle_s = self.progress.get("accumulated_idle_time", 0.0)
|
||||
start_str = self.progress.get("tomo_start_time")
|
||||
|
||||
@@ -1233,14 +1233,22 @@ 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, account=None) -> bool:
|
||||
return dev.omny_samples.set_sample_status(container, slot_nr, status, account)
|
||||
def otransfer_set_sample_measured(
|
||||
self, container: str, slot_nr: int, status: str, meas_date: str, owner_if_empty: str = ""
|
||||
):
|
||||
return dev.omny_samples.set_sample_measured(container, slot_nr, status, meas_date, owner_if_empty)
|
||||
|
||||
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_meas_date(self, container: str, slot_nr: int) -> str:
|
||||
return dev.omny_samples.get_sample_meas_date(container, slot_nr)
|
||||
|
||||
def otransfer_get_sample_record(self, container: str, slot_nr: int) -> dict:
|
||||
return dev.omny_samples.get_sample_record(container, slot_nr)
|
||||
|
||||
def otransfer_transfer_sample(self, from_slot, to_slot):
|
||||
return dev.omny_samples.transfer_sample(from_slot, to_slot)
|
||||
|
||||
def otransfer_get_sample_owner_in_gripper(self) -> str:
|
||||
return dev.omny_samples.get_sample_owner_in_gripper()
|
||||
@@ -1248,24 +1256,12 @@ class OMNYSampleTransferMixin:
|
||||
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, 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_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
|
||||
@@ -1331,23 +1327,11 @@ class OMNYSampleTransferMixin:
|
||||
# is there a sample available?
|
||||
if dev.omny_samples.is_sample_slot_used(installedshuttle, _osampleposition):
|
||||
if _picked_it == 1:
|
||||
samplename = dev.omny_samples.get_sample_name(
|
||||
installedshuttle, _osampleposition
|
||||
)
|
||||
owner = dev.omny_samples.get_sample_owner(
|
||||
installedshuttle, _osampleposition
|
||||
)
|
||||
status = dev.omny_samples.get_sample_status(
|
||||
installedshuttle, _osampleposition
|
||||
)
|
||||
account = dev.omny_samples.get_sample_account(
|
||||
installedshuttle, _osampleposition
|
||||
)
|
||||
# remove sample from slot and put to gripper -- owner/status/account
|
||||
# remove sample from slot and put to gripper -- name/owner/status/date
|
||||
# carry over with it, the same physical sample
|
||||
dev.omny_samples.unset_sample_slot(installedshuttle, _osampleposition)
|
||||
dev.omny_samples.set_sample_in_gripper(samplename, owner)
|
||||
dev.omny_samples.set_sample_status_in_gripper(status, account)
|
||||
dev.omny_samples.transfer_sample(
|
||||
(installedshuttle, _osampleposition), "GRIPPER"
|
||||
)
|
||||
return True
|
||||
else:
|
||||
print("There is no sample availabe at that position.")
|
||||
@@ -1435,13 +1419,7 @@ class OMNYSampleTransferMixin:
|
||||
):
|
||||
if dev.omny_samples.is_sample_slot_used("O", _osampleposition):
|
||||
if _picked_it:
|
||||
samplename = dev.omny_samples.get_sample_name("O", _osampleposition)
|
||||
owner = dev.omny_samples.get_sample_owner("O", _osampleposition)
|
||||
status = dev.omny_samples.get_sample_status("O", _osampleposition)
|
||||
account = dev.omny_samples.get_sample_account("O", _osampleposition)
|
||||
dev.omny_samples.unset_sample_slot("O", _osampleposition)
|
||||
dev.omny_samples.set_sample_in_gripper(samplename, owner)
|
||||
dev.omny_samples.set_sample_status_in_gripper(status, account)
|
||||
dev.omny_samples.transfer_sample(("O", _osampleposition), "GRIPPER")
|
||||
# FZP was picked
|
||||
# if (_osampleposition == 10 || _osampleposition == 11 || _osampleposition == 100)
|
||||
# _otransfer_modify_sample("omny",110,2,samplename)
|
||||
@@ -1460,13 +1438,7 @@ class OMNYSampleTransferMixin:
|
||||
elif _osampleposition == 0:
|
||||
if dev.omny_samples.is_sample_in_samplestage():
|
||||
if _picked_it:
|
||||
samplename = dev.omny_samples.get_sample_name_in_samplestage()
|
||||
owner = dev.omny_samples.get_sample_owner_in_samplestage()
|
||||
status = dev.omny_samples.get_sample_status_in_samplestage()
|
||||
account = dev.omny_samples.get_sample_account_in_samplestage()
|
||||
dev.omny_samples.unset_sample_in_samplestage()
|
||||
dev.omny_samples.set_sample_in_gripper(samplename, owner)
|
||||
dev.omny_samples.set_sample_status_in_gripper(status, account)
|
||||
dev.omny_samples.transfer_sample("STAGE", "GRIPPER")
|
||||
# FZP was picked
|
||||
# if (_osampleposition == 10 || _osampleposition == 11 || _osampleposition == 100)
|
||||
# _otransfer_modify_sample("omny",110,2,samplename)
|
||||
@@ -1538,16 +1510,8 @@ class OMNYSampleTransferMixin:
|
||||
# is there free position available?
|
||||
if not dev.omny_samples.is_sample_slot_used(installedshuttle, _osampleposition):
|
||||
if _put_it:
|
||||
samplename = dev.omny_samples.get_sample_name_in_gripper()
|
||||
owner = dev.omny_samples.get_sample_owner_in_gripper()
|
||||
status = dev.omny_samples.get_sample_status_in_gripper()
|
||||
account = dev.omny_samples.get_sample_account_in_gripper()
|
||||
dev.omny_samples.unset_sample_in_gripper()
|
||||
dev.omny_samples.set_sample_slot(
|
||||
installedshuttle, _osampleposition, samplename, owner
|
||||
)
|
||||
dev.omny_samples.set_sample_status(
|
||||
installedshuttle, _osampleposition, status, account
|
||||
dev.omny_samples.transfer_sample(
|
||||
"GRIPPER", (installedshuttle, _osampleposition)
|
||||
)
|
||||
return True
|
||||
else:
|
||||
@@ -1636,13 +1600,7 @@ class OMNYSampleTransferMixin:
|
||||
# is there a free slot available?
|
||||
if not dev.omny_samples.is_sample_slot_used("O", _osampleposition):
|
||||
if _put_it:
|
||||
samplename = dev.omny_samples.get_sample_name_in_gripper()
|
||||
owner = dev.omny_samples.get_sample_owner_in_gripper()
|
||||
status = dev.omny_samples.get_sample_status_in_gripper()
|
||||
account = dev.omny_samples.get_sample_account_in_gripper()
|
||||
dev.omny_samples.set_sample_slot("O", _osampleposition, samplename, owner)
|
||||
dev.omny_samples.set_sample_status("O", _osampleposition, status, account)
|
||||
dev.omny_samples.unset_sample_in_gripper()
|
||||
dev.omny_samples.transfer_sample("GRIPPER", ("O", _osampleposition))
|
||||
return True
|
||||
else:
|
||||
print("There is no free space at this position.")
|
||||
@@ -1652,13 +1610,7 @@ class OMNYSampleTransferMixin:
|
||||
elif _osampleposition == 0:
|
||||
if not dev.omny_samples.is_sample_in_samplestage():
|
||||
if _put_it:
|
||||
samplename = dev.omny_samples.get_sample_name_in_gripper()
|
||||
owner = dev.omny_samples.get_sample_owner_in_gripper()
|
||||
status = dev.omny_samples.get_sample_status_in_gripper()
|
||||
account = dev.omny_samples.get_sample_account_in_gripper()
|
||||
dev.omny_samples.set_sample_in_samplestage(samplename, owner)
|
||||
dev.omny_samples.set_sample_status_in_samplestage(status, account)
|
||||
dev.omny_samples.unset_sample_in_gripper()
|
||||
dev.omny_samples.transfer_sample("GRIPPER", "STAGE")
|
||||
return True
|
||||
else:
|
||||
print("There is no sample in the sample stage.")
|
||||
|
||||
@@ -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, but
|
||||
via different backend mechanisms:
|
||||
Both setups require every sample slot to carry a name *and* an owner; both
|
||||
keep one soft-IOC record per slot (USED/NAME/OWNER/STATUS/MEAS_DATE):
|
||||
|
||||
FlOMNI (``dev.flomni_samples``)
|
||||
A flat magazine: slots 1-20, the sample stage (slot 0) and the gripper
|
||||
(slot 100). Name and owner are packed into one EPICS DESC field (see
|
||||
``sample_desc_codec``) since this backend has no separate owner PV.
|
||||
(slot 100), FLOMNI-SAMPLE:<slot>:... since 2026-10-17 (before: name and
|
||||
owner packed into one EPICS DESC field, ``sample_desc_codec``).
|
||||
|
||||
OMNY (``dev.omny_samples``)
|
||||
A richer structure: three shuttles (A/B/C) of 6 slots each, a sparse
|
||||
@@ -23,11 +23,10 @@ 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/status). Unlike FlOMNI, OMNY has a
|
||||
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
|
||||
so they carry no name/owner/status). Since 2026-10-18 every OMNY slot is
|
||||
one record on the OMNY soft IOC, like FlOMNI's
|
||||
(OMNY-SAMPLE:<slot>:USED/NAME/OWNER/STATUS/MEAS_DATE, see
|
||||
``OMNYSampleStorage``). 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()``).
|
||||
|
||||
@@ -66,9 +65,11 @@ FlOMNI
|
||||
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.
|
||||
OMNY, stage/gripper/fixed-O/shuttle slots -- see ``_write_slot_omny()``.
|
||||
OMNY, parking : parking_placed.parking{N}.set(used)
|
||||
parking_names.parking{N}.set(letter or "none")
|
||||
OMNY, stage/gripper/fixed-O/shuttle slots: set_sample_slot(container, N,
|
||||
name, owner) / unset_sample_slot(container, N) -- see
|
||||
``_write_slot_omny()``; the same one-record-per-slot layout
|
||||
(OMNY-SAMPLE:<slot>:..., 2026-10-18).
|
||||
OMNY, parking : set_shuttle_slot(letter, N) / unset_shuttle_slot(N)
|
||||
|
||||
Polling
|
||||
-------
|
||||
@@ -135,14 +136,6 @@ 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, duplicated locally rather than read
|
||||
# off the device -- self._samples is the BEC *client-side* device proxy,
|
||||
# which only exposes USER_ACCESS methods/signals, not plain class
|
||||
# attributes like STATUS_NONE (reaching for self._samples.STATUS_NONE
|
||||
# raised "'Device' object has no attribute 'STATUS_NONE'" here, even though
|
||||
# it works when testing directly against the real device class).
|
||||
OMNY_STATUS_NONE = 0
|
||||
|
||||
# colours
|
||||
# An occupied slot's name label intentionally has no fixed foreground colour
|
||||
# here (see _SlotCell.set_state()) -- it was previously hardcoded to
|
||||
@@ -337,10 +330,9 @@ class _SlotCell(QFrame):
|
||||
"""Update the cell's displayed state (called by the poll/refresh).
|
||||
|
||||
`measured_status` is "started"/"completed" (FLOMNI-SAMPLE:<slot>:STATUS)
|
||||
or None if this slot's current occupant has no measured record.
|
||||
OMNY cells always pass measured_status=None (no measured-sample log
|
||||
for OMNY -- see module docstring), which the measured label below
|
||||
already renders as blank.
|
||||
(OMNY: OMNY-SAMPLE:<slot>:STATUS) or None if this slot's current
|
||||
occupant has no measured record, which the measured label below
|
||||
renders as blank.
|
||||
"""
|
||||
self._occupied = occupied
|
||||
self._name = name if name else EMPTY_NAME
|
||||
@@ -599,11 +591,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. Owner and measured status each have their own
|
||||
dedicated OMNYSampleStorage signal (added alongside the existing
|
||||
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``.
|
||||
the 18 shuttle slots. One record per slot on the OMNY soft IOC
|
||||
(OMNY-SAMPLE:<slot>:USED/NAME/OWNER/STATUS/MEAS_DATE, see
|
||||
``OMNYSampleStorage``); the flat keys are
|
||||
``omny_samples_sample_<field>_<slot>`` with <slot> the IOC slot name
|
||||
(A1..C6, P10..P101, STAGE, GRIPPER).
|
||||
"""
|
||||
|
||||
def _val(key, default=None):
|
||||
@@ -612,50 +604,22 @@ class OMNY_SampleStorage(BECWidget, QWidget):
|
||||
return entry.get("value", default)
|
||||
return default
|
||||
|
||||
def _status_str(raw) -> Optional[str]:
|
||||
# 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
|
||||
if val == 0:
|
||||
return None
|
||||
return "started" if val < 0 else "completed"
|
||||
def _record(slot: str) -> tuple[bool, str, str, Optional[str]]:
|
||||
used = bool(_val(f"omny_samples_sample_used_{slot}", 0))
|
||||
name = _val(f"omny_samples_sample_names_{slot}", EMPTY_NAME) or EMPTY_NAME
|
||||
owner = _val(f"omny_samples_sample_owners_{slot}", "") or ""
|
||||
raw = _val(f"omny_samples_sample_status_{slot}")
|
||||
status = str(raw) if raw is not None else ""
|
||||
return (used, str(name), str(owner), status if used and status in ("started", "completed") else None)
|
||||
|
||||
state: dict[tuple, tuple[bool, str, str, Optional[str]]] = {}
|
||||
|
||||
used = bool(_val("omny_samples_sample_in_samplestage", 0))
|
||||
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 = _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)
|
||||
|
||||
state[("stage",)] = _record("STAGE")
|
||||
state[("gripper",)] = _record("GRIPPER")
|
||||
for n in OMNY_FIXED_POSITIONS:
|
||||
used = bool(_val(f"omny_samples_sample_placed_sample{n}", 0))
|
||||
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)
|
||||
|
||||
state[("O", n)] = _record(f"P{n}")
|
||||
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 = _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)
|
||||
|
||||
state[("shuttle", letter, n)] = _record(f"{letter}{n}")
|
||||
return state
|
||||
|
||||
def _parse_parking_assignment_omny(self, data: dict) -> dict[int, str]:
|
||||
@@ -769,39 +733,35 @@ class OMNY_SampleStorage(BECWidget, QWidget):
|
||||
(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.
|
||||
Status, owner and measurement date live on the slot record itself
|
||||
(OMNY-SAMPLE:<slot>:..., as FlOMNI), read fresh from the device right
|
||||
before the clear; the owner is printed as account, as on FlOMNI.
|
||||
"""
|
||||
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)
|
||||
record = self._samples.get_sample_record(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":
|
||||
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 ({status}), account {account}. Print a label?",
|
||||
f"Sample '{prev_name}' was measured ({record['status']}) on"
|
||||
f" {record['meas_date']}. 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_ACCOUNT: record["owner"],
|
||||
PTOUCH_FIELD_DATE: record["meas_date"],
|
||||
PTOUCH_FIELD_SAMPLENAME: prev_name,
|
||||
}
|
||||
# Retry loop, same reasoning as _offer_print_label_for_previous_occupant():
|
||||
@@ -852,14 +812,11 @@ 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()``. 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.
|
||||
gripper, a fixed O position, or a shuttle slot -- through the device
|
||||
methods, like ``_write_slot()``. Measured status is reset to "none"
|
||||
on every write here (``OMNYSampleStorage.set_sample_slot()``/
|
||||
``unset_sample_slot()`` write the whole record) -- a fresh name
|
||||
(even a 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
|
||||
@@ -873,31 +830,20 @@ class OMNY_SampleStorage(BECWidget, QWidget):
|
||||
if used == 0 and kind == "shuttle":
|
||||
self._offer_print_label_for_previous_occupant_omny(key)
|
||||
try:
|
||||
status_none = OMNY_STATUS_NONE
|
||||
if kind == "stage":
|
||||
self._samples.sample_in_samplestage.set(used)
|
||||
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)
|
||||
container, n = "STAGE", 0
|
||||
elif kind == "gripper":
|
||||
self._samples.sample_in_gripper.set(used)
|
||||
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)
|
||||
container, n = "GRIPPER", 0
|
||||
elif kind == "O":
|
||||
_, n = key
|
||||
getattr(self._samples.sample_placed, f"sample{n}").set(used)
|
||||
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)
|
||||
container, n = "O", key[1]
|
||||
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(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)
|
||||
container, n = key[1], key[2]
|
||||
else:
|
||||
raise ValueError(f"unknown OMNY slot key {key!r}")
|
||||
if used:
|
||||
self._samples.set_sample_slot(container, n, name, owner)
|
||||
else:
|
||||
self._samples.unset_sample_slot(container, n)
|
||||
self._last_state.pop(key, None)
|
||||
return True
|
||||
except Exception as exc:
|
||||
@@ -921,10 +867,10 @@ class OMNY_SampleStorage(BECWidget, QWidget):
|
||||
in-progress transfer.
|
||||
"""
|
||||
try:
|
||||
used = 1 if letter else 0
|
||||
value = letter if letter else PARKING_EMPTY_SENTINEL
|
||||
getattr(self._samples.parking_placed, f"parking{slot_nr}").set(used)
|
||||
getattr(self._samples.parking_names, f"parking{slot_nr}").set(value)
|
||||
if letter:
|
||||
self._samples.set_shuttle_slot(letter, slot_nr)
|
||||
else:
|
||||
self._samples.unset_shuttle_slot(slot_nr)
|
||||
return True
|
||||
except Exception as exc:
|
||||
logger.warning(f"OMNY_SampleStorage: write of parking slot {slot_nr} failed: {exc}")
|
||||
@@ -1146,9 +1092,10 @@ class OMNY_SampleStorage(BECWidget, QWidget):
|
||||
"""Return the slot key that already holds `name`, if any OTHER
|
||||
occupied slot of the active setup currently has this exact name, or
|
||||
None. Used by action_change_sample()/action_new_sample() to enforce
|
||||
that sample names are unique -- otherwise the measured-sample log
|
||||
(FlOMNI, keyed by name+owner) and operators reading the storage
|
||||
overview could no longer tell two same-named samples apart.
|
||||
that sample names are unique -- otherwise operators reading the
|
||||
storage overview (and the measurement records, scilog entries and
|
||||
labels keyed by sample name) could no longer tell two same-named
|
||||
samples apart.
|
||||
|
||||
Reads from ``self._last_state``, the same poll-refreshed snapshot
|
||||
the cells themselves are painted from, rather than issuing a fresh
|
||||
@@ -1168,7 +1115,7 @@ class OMNY_SampleStorage(BECWidget, QWidget):
|
||||
def action_change_sample(self, slot, current_name: str, current_owner: str = "") -> None:
|
||||
"""Replace an occupied slot's sample identity (name + owner), stays
|
||||
occupied. Named "change sample" rather than "rename" because
|
||||
_write_sample() resets the slot's measured status/account back to
|
||||
_write_sample() resets the slot's measured status/date back to
|
||||
"none" -- appropriate since a new name+owner means a different
|
||||
physical sample, not a correction to the existing one's record.
|
||||
Shared by both setups (see ``_SlotCell``) -- both require a name and
|
||||
@@ -1484,9 +1431,8 @@ class OMNY_SampleStorage(BECWidget, QWidget):
|
||||
def _draw_report_cell(self, pdf, x, y, w, h, slot, entry) -> None:
|
||||
"""Draw one slot's box, mirroring what ``_SlotCell.set_state()``
|
||||
shows on screen: slot caption, name (or "empty"), owner and measured
|
||||
marker. Shared by both setups' report builders -- OMNY entries
|
||||
always carry owner="" and measured_status=None, which render as
|
||||
blank without any OMNY-specific branch here.
|
||||
marker. Shared by both setups' report builders (same entry tuple
|
||||
for both, no setup-specific branch here).
|
||||
"""
|
||||
pdf.rect(x, y, w, h)
|
||||
occupied, name, owner, measured_status = entry if entry else (False, EMPTY_NAME, "", None)
|
||||
|
||||
@@ -3050,7 +3050,7 @@ SETUP_PROFILES: dict[str, dict[str, Any]] = {
|
||||
# get_sample_name_in_samplestage device_rpc entries stuck PENDING
|
||||
# after an abort left the queue paused, plus a GUI freeze while a
|
||||
# scan was running -- the wrapper method itself is just
|
||||
# `return str(self.sample_in_samplestage_name.get())`
|
||||
# the NAME of the STAGE record, OMNY-SAMPLE:STAGE:NAME
|
||||
# (omny_sample_storage.py), so reading the signal directly is
|
||||
# equivalent and free of both problems. cached=True goes one step
|
||||
# further: without it, DeviceBase.get() still issues a live
|
||||
@@ -3058,9 +3058,7 @@ SETUP_PROFILES: dict[str, dict[str, Any]] = {
|
||||
# cached=True reads BEC's already-locally-updated status cache
|
||||
# instead, no round-trip at all -- standard advice for GUI/widget
|
||||
# code (see e.g. sample_storage.py's oparkz.readback.get(cached=True)).
|
||||
"sample_name_getter": lambda w: w.dev.omny_samples.sample_in_samplestage_name.get(
|
||||
cached=True
|
||||
),
|
||||
"sample_name_getter": lambda w: w.dev.omny_samples.sample_names.STAGE.get(cached=True),
|
||||
"cli_hint_name": "omny",
|
||||
# OMNY has three "equally spaced sub-tomograms" flavors (2/4/8
|
||||
# sub-tomograms), unlike flomni's/lamni's single always-8 flavor --
|
||||
|
||||
@@ -1,5 +1,33 @@
|
||||
"""OMNY sample storage on the OMNY soft IOC (``OMNYtools/softioc/ioc_omny.py`` in the OMNY repo).
|
||||
|
||||
One record per slot, all fields autosaved by the IOC (same layout as flOMNI's
|
||||
``FLOMNI-SAMPLE:``, see ``flomni_sample_storage.py``)::
|
||||
|
||||
OMNY-SAMPLE:<slot>:USED 0 = free, 1 = sample present
|
||||
OMNY-SAMPLE:<slot>:NAME sample name (long string, up to 256 chars)
|
||||
OMNY-SAMPLE:<slot>:OWNER e-account of the owner
|
||||
OMNY-SAMPLE:<slot>:STATUS none / started / completed (measurement status)
|
||||
OMNY-SAMPLE:<slot>:MEAS_DATE date of the measurement
|
||||
|
||||
<slot> (container, slot_nr of this device in brackets):
|
||||
A1..A6, B1..B6, C1..C6 shuttles ("A".."C", 1-6)
|
||||
P10..P14, P32..P34, P100, P101 fixed positions ("O", 10-14, 32-34, 100, 101)
|
||||
STAGE sample stage (set/unset_sample_in_samplestage)
|
||||
GRIPPER gripper (set/unset_sample_in_gripper)
|
||||
PARK1..PARK6 parking slots: USED + NAME = shuttle letter (set_shuttle_slot)
|
||||
FZP0, OSA0, OSA1 exist in the IOC, not used by the transfer yet
|
||||
|
||||
Replaces (2026-10-18) the classic softIoc records ``XOMNY-SAMPLE_DB_<container>:<pos>``
|
||||
(VAL = placed, DESC = name) and ``...:<pos>:STATUS`` (VAL = status sign x account,
|
||||
DESC = owner). The status is now a plain enum plus a date, as on flOMNI, and 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,11 +35,68 @@ from ophyd import DynamicDeviceComponent as Dcpt
|
||||
from ophyd import EpicsSignal
|
||||
from prettytable import FRAME, PrettyTable
|
||||
|
||||
PV_PREFIX = "OMNY-SAMPLE:"
|
||||
SHUTTLES = ("A", "B", "C")
|
||||
SHUTTLE_SLOTS = range(1, 7)
|
||||
FIXED_SLOTS = (10, 11, 12, 13, 14, 32, 33, 34, 100, 101)
|
||||
PARKING_SLOTS = range(1, 7)
|
||||
STAGE = "STAGE"
|
||||
GRIPPER = "GRIPPER"
|
||||
EMPTY_NAME = "-"
|
||||
NO_SHUTTLE = "none"
|
||||
STATUS_NONE = "none"
|
||||
STATUS_STARTED = "started"
|
||||
STATUS_COMPLETED = "completed"
|
||||
STATUS_VALUES = (STATUS_NONE, STATUS_STARTED, STATUS_COMPLETED)
|
||||
FIELDS = ("used", "name", "owner", "status", "meas_date")
|
||||
|
||||
# every sample slot of the IOC that this device handles, by PV slot name
|
||||
SAMPLE_SLOTS = (
|
||||
[f"{s}{i}" for s in SHUTTLES for i in SHUTTLE_SLOTS] + [f"P{n}" for n in FIXED_SLOTS] + [STAGE, GRIPPER]
|
||||
)
|
||||
|
||||
|
||||
class OMNYSampleStorageError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def slot_name(container: str, slot_nr: int = 0) -> str:
|
||||
"""("A", 3) -> "A3", ("O", 10) -> "P10"; "STAGE"/"GRIPPER" (and a slot name
|
||||
such as "A3" or "P10") are passed through."""
|
||||
container = str(container)
|
||||
if container.upper() in (STAGE, GRIPPER):
|
||||
return container.upper()
|
||||
if container in SAMPLE_SLOTS:
|
||||
return container
|
||||
slot_nr = int(slot_nr)
|
||||
if container in SHUTTLES and slot_nr in SHUTTLE_SLOTS:
|
||||
return f"{container}{slot_nr}"
|
||||
if container == "O" and slot_nr in FIXED_SLOTS:
|
||||
return f"P{slot_nr}"
|
||||
raise OMNYSampleStorageError(
|
||||
f"Invalid sample slot ({container!r}, {slot_nr}): container A/B/C with slot 1-6, "
|
||||
f"O with {FIXED_SLOTS}, or STAGE/GRIPPER."
|
||||
)
|
||||
|
||||
|
||||
def _field_pv(slot: str, field: str) -> str:
|
||||
return f"{PV_PREFIX}{slot}:{field.upper()}"
|
||||
|
||||
|
||||
def _sample_signals(field: str) -> dict:
|
||||
kwargs = {"auto_monitor": True}
|
||||
if field != "used":
|
||||
kwargs["string"] = True
|
||||
return {slot: (EpicsSignal, _field_pv(slot, field), dict(kwargs)) for slot in SAMPLE_SLOTS}
|
||||
|
||||
|
||||
def _parking_signals(field: str) -> dict:
|
||||
kwargs = {"auto_monitor": True}
|
||||
if field != "used":
|
||||
kwargs["string"] = True
|
||||
return {f"parking{i}": (EpicsSignal, _field_pv(f"PARK{i}", field), dict(kwargs)) for i in PARKING_SLOTS}
|
||||
|
||||
|
||||
class OMNYSampleStorage(Device):
|
||||
USER_ACCESS = [
|
||||
"is_sample_slot_used",
|
||||
@@ -26,18 +111,16 @@ class OMNYSampleStorage(Device):
|
||||
"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",
|
||||
"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",
|
||||
"get_sample_meas_date",
|
||||
"get_sample_record",
|
||||
"set_sample_measured",
|
||||
"transfer_sample",
|
||||
"is_sample_in_samplestage",
|
||||
"set_shuttle_slot",
|
||||
"unset_shuttle_slot",
|
||||
@@ -50,482 +133,197 @@ class OMNYSampleStorage(Device):
|
||||
SUB_VALUE = "value"
|
||||
_default_sub = SUB_VALUE
|
||||
|
||||
# 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
|
||||
FALLBACK_ACCOUNT = 11111
|
||||
# sample slots, attribute = PV slot name (A1 ... C6, P10 ... P101, STAGE, GRIPPER)
|
||||
sample_used = Dcpt(_sample_signals("used"))
|
||||
sample_names = Dcpt(_sample_signals("name"))
|
||||
sample_owners = Dcpt(_sample_signals("owner"))
|
||||
sample_status = Dcpt(_sample_signals("status"))
|
||||
sample_meas_date = Dcpt(_sample_signals("meas_date"))
|
||||
|
||||
sample_shuttle_A_placed = {
|
||||
f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_shuttle_A:{i}", {"auto_monitor": True}) for i in range(1, 7)
|
||||
}
|
||||
sample_shuttle_A_placed = Dcpt(sample_shuttle_A_placed)
|
||||
|
||||
sample_shuttle_B_placed = {
|
||||
f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_shuttle_B:{i}", {"auto_monitor": True}) for i in range(1, 7)
|
||||
}
|
||||
sample_shuttle_B_placed = Dcpt(sample_shuttle_B_placed)
|
||||
|
||||
sample_shuttle_C_placed = {
|
||||
f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_shuttle_C:{i}", {"auto_monitor": True}) for i in range(1, 7)
|
||||
}
|
||||
sample_shuttle_C_placed = Dcpt(sample_shuttle_C_placed)
|
||||
|
||||
sample_shuttle_C_placed = {
|
||||
f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_shuttle_C:{i}", {"auto_monitor": True}) for i in range(1, 7)
|
||||
}
|
||||
sample_shuttle_C_placed = Dcpt(sample_shuttle_C_placed)
|
||||
|
||||
parking_placed = {
|
||||
f"parking{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_parking:{i}", {"auto_monitor": True}) for i in range(1, 7)
|
||||
}
|
||||
parking_placed = Dcpt(parking_placed)
|
||||
|
||||
sample_placed = {
|
||||
f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_omny:{i}", {"auto_monitor": True})
|
||||
for i in [10, 11, 12, 13, 14, 32, 33, 34, 100, 101]
|
||||
}
|
||||
sample_placed = Dcpt(sample_placed)
|
||||
|
||||
sample_shuttle_A_names = {
|
||||
f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_shuttle_A:{i}.DESC", {"string": True, "auto_monitor": True})
|
||||
for i in range(1, 7)
|
||||
}
|
||||
sample_shuttle_A_names = Dcpt(sample_shuttle_A_names)
|
||||
|
||||
sample_shuttle_B_names = {
|
||||
f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_shuttle_B:{i}.DESC", {"string": True, "auto_monitor": True})
|
||||
for i in range(1, 7)
|
||||
}
|
||||
sample_shuttle_B_names = Dcpt(sample_shuttle_B_names)
|
||||
|
||||
sample_shuttle_C_names = {
|
||||
f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_shuttle_C:{i}.DESC", {"string": True, "auto_monitor": True})
|
||||
for i in range(1, 7)
|
||||
}
|
||||
sample_shuttle_C_names = Dcpt(sample_shuttle_C_names)
|
||||
|
||||
parking_names = {
|
||||
f"parking{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_parking:{i}.DESC", {"string": True, "auto_monitor": True})
|
||||
for i in range(1, 7)
|
||||
}
|
||||
parking_names = Dcpt(parking_names)
|
||||
|
||||
sample_names = {
|
||||
f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_omny:{i}.DESC", {"string": True, "auto_monitor": True})
|
||||
for i in [10, 11, 12, 13, 14, 32, 33, 34, 100, 101]
|
||||
}
|
||||
sample_names = Dcpt(sample_names)
|
||||
|
||||
# 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)
|
||||
}
|
||||
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
|
||||
)
|
||||
sample_in_gripper_name = Cpt(
|
||||
EpicsSignal,
|
||||
name="sample_in_gripper_name",
|
||||
read_pv="XOMNY-SAMPLE_DB_omny:110.DESC",
|
||||
string=True,
|
||||
auto_monitor=True
|
||||
)
|
||||
|
||||
sample_in_samplestage = Cpt(
|
||||
EpicsSignal, name="sample_in_samplestage", read_pv="XOMNY-SAMPLE_DB_omny:0.VAL", auto_monitor=True
|
||||
)
|
||||
sample_in_samplestage_name = Cpt(
|
||||
EpicsSignal,
|
||||
name="sample_in_samplestage_name",
|
||||
read_pv="XOMNY-SAMPLE_DB_omny:0.DESC",
|
||||
string=True,
|
||||
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,
|
||||
)
|
||||
# parking slots 1-6: which shuttle (letter in NAME) is docked
|
||||
parking_placed = Dcpt(_parking_signals("used"))
|
||||
parking_names = Dcpt(_parking_signals("name"))
|
||||
|
||||
def __init__(self, prefix="", *, name, **kwargs):
|
||||
super().__init__(prefix, name=name, **kwargs)
|
||||
self.sample_shuttle_A_placed.sample1.subscribe(self._emit_value)
|
||||
self.sample_used.A1.subscribe(self._emit_value)
|
||||
|
||||
def _emit_value(self, **kwargs):
|
||||
timestamp = kwargs.pop("timestamp", time.time())
|
||||
self.wait_for_connection()
|
||||
self._run_subs(sub_type=self.SUB_VALUE, timestamp=timestamp, obj=self)
|
||||
|
||||
@classmethod
|
||||
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.
|
||||
# ── slot addressing ──────────────────────────────────────────────────────
|
||||
def _sig(self, field: str, slot: str) -> EpicsSignal:
|
||||
group = {
|
||||
"used": self.sample_used,
|
||||
"name": self.sample_names,
|
||||
"owner": self.sample_owners,
|
||||
"status": self.sample_status,
|
||||
"meas_date": self.sample_meas_date,
|
||||
}[field]
|
||||
return getattr(group, slot)
|
||||
|
||||
@staticmethod
|
||||
def _check_parking(slot_nr: int) -> int:
|
||||
slot_nr = int(slot_nr)
|
||||
if slot_nr not in PARKING_SLOTS:
|
||||
raise OMNYSampleStorageError(f"Invalid parking slot {slot_nr} (1-6).")
|
||||
return slot_nr
|
||||
|
||||
@staticmethod
|
||||
def _str(val) -> str:
|
||||
return "" if val is None else str(val)
|
||||
|
||||
# ── write ────────────────────────────────────────────────────────────────
|
||||
def _write(self, slot: str, used: int, name: str, owner: str, status: str, meas_date: str):
|
||||
if status not in STATUS_VALUES:
|
||||
raise OMNYSampleStorageError(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).set(name).wait(timeout=5)
|
||||
self._sig("owner", slot).set(owner or "").wait(timeout=5)
|
||||
self._sig("status", slot).set(status).wait(timeout=5)
|
||||
self._sig("meas_date", slot).set(meas_date or "").wait(timeout=5)
|
||||
self._sig("used", slot).set(int(used)).wait(timeout=5)
|
||||
|
||||
def set_sample_slot(
|
||||
self,
|
||||
container: str,
|
||||
slot_nr: int,
|
||||
name: str,
|
||||
owner: str = "",
|
||||
status: str = STATUS_NONE,
|
||||
meas_date: str = "",
|
||||
):
|
||||
"""Place a sample record in (container, slot_nr): A/B/C 1-6, O 10-14/32-34/100/101."""
|
||||
self._write(slot_name(container, slot_nr), 1, name, owner, status, meas_date)
|
||||
|
||||
def unset_sample_slot(self, container: str, slot_nr: int):
|
||||
"""Clear (container, slot_nr)."""
|
||||
self._write(slot_name(container, 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, 0, name, owner, status, meas_date)
|
||||
|
||||
def unset_sample_in_gripper(self):
|
||||
self.unset_sample_slot(GRIPPER, 0)
|
||||
|
||||
def set_sample_in_samplestage(
|
||||
self, name: str, owner: str = "", status: str = STATUS_NONE, meas_date: str = ""
|
||||
):
|
||||
self.set_sample_slot(STAGE, 0, name, owner, status, meas_date)
|
||||
|
||||
def unset_sample_in_samplestage(self):
|
||||
self.unset_sample_slot(STAGE, 0)
|
||||
|
||||
def set_sample_measured(self, container: str, slot_nr: int, status: str, meas_date: str, owner_if_empty: str = ""):
|
||||
"""Record the measurement status/date of the sample in (container, slot_nr)
|
||||
(container "STAGE" for the sample stage).
|
||||
|
||||
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.
|
||||
"""
|
||||
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
|
||||
return cls.FALLBACK_ACCOUNT
|
||||
if status not in STATUS_VALUES:
|
||||
raise OMNYSampleStorageError(f"Invalid status {status!r}, expected one of {STATUS_VALUES}.")
|
||||
slot = slot_name(container, slot_nr)
|
||||
if owner_if_empty and not self._str(self._sig("owner", slot).get()):
|
||||
self._sig("owner", slot).set(owner_if_empty).wait(timeout=5)
|
||||
self._sig("status", slot).set(status).wait(timeout=5)
|
||||
self._sig("meas_date", slot).set(meas_date or "").wait(timeout=5)
|
||||
|
||||
@classmethod
|
||||
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
|
||||
def transfer_sample(self, from_slot, to_slot):
|
||||
"""Move the complete record (name, owner, status, date) from one slot to
|
||||
another and clear the source. Slots as (container, slot_nr) tuples or slot
|
||||
names: e.g. transfer_sample(("A", 3), "GRIPPER"), transfer_sample("GRIPPER", "STAGE")."""
|
||||
src = slot_name(*from_slot) if isinstance(from_slot, tuple) else slot_name(from_slot)
|
||||
dst = slot_name(*to_slot) if isinstance(to_slot, tuple) else slot_name(to_slot)
|
||||
rec = self.get_sample_record(src)
|
||||
self._write(dst, 1, rec["name"], rec["owner"], rec["status"], rec["meas_date"])
|
||||
self._write(src, 0, EMPTY_NAME, "", STATUS_NONE, "")
|
||||
|
||||
@classmethod
|
||||
def _val_to_status(cls, val) -> str:
|
||||
try:
|
||||
v = int(round(float(val)))
|
||||
except (TypeError, ValueError):
|
||||
return "none"
|
||||
if v == 0:
|
||||
return "none"
|
||||
return "started" if v < 0 else "completed"
|
||||
def set_shuttle_slot(self, container: str, slot_nr: int):
|
||||
"""Shuttle `container` (A/B/C) is docked at parking slot `slot_nr` (1-6)."""
|
||||
slot_nr = self._check_parking(slot_nr)
|
||||
getattr(self.parking_names, f"parking{slot_nr}").set(container).wait(timeout=5)
|
||||
getattr(self.parking_placed, f"parking{slot_nr}").set(1).wait(timeout=5)
|
||||
|
||||
@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 unset_shuttle_slot(self, slot_nr: int):
|
||||
slot_nr = self._check_parking(slot_nr)
|
||||
getattr(self.parking_names, f"parking{slot_nr}").set(NO_SHUTTLE).wait(timeout=5)
|
||||
getattr(self.parking_placed, f"parking{slot_nr}").set(0).wait(timeout=5)
|
||||
|
||||
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_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_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, 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":
|
||||
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._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":
|
||||
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:
|
||||
raise OMNYSampleStorageError(f"Invalid slot number {slot_nr}.")
|
||||
getattr(self.parking_placed, f"parking{slot_nr}").set(1)
|
||||
getattr(self.parking_names, f"parking{slot_nr}").set(container)
|
||||
|
||||
def unset_shuttle_slot(self, slot_nr: int) -> bool:
|
||||
if slot_nr > 6:
|
||||
raise OMNYSampleStorageError(f"Invalid slot number {slot_nr}.")
|
||||
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, 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_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_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_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_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, 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._val_to_status(self.sample_in_gripper_status.get())
|
||||
|
||||
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._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":
|
||||
val = getattr(self.sample_shuttle_A_placed, f"sample{slot_nr}").get()
|
||||
if container == "B":
|
||||
val = getattr(self.sample_shuttle_B_placed, f"sample{slot_nr}").get()
|
||||
if container == "C":
|
||||
val = getattr(self.sample_shuttle_C_placed, f"sample{slot_nr}").get()
|
||||
elif container == "O":
|
||||
val = getattr(self.sample_placed, f"sample{slot_nr}").get()
|
||||
return bool(val)
|
||||
|
||||
def is_shuttle_slot_used(self, slot_nr: int) -> bool:
|
||||
val = getattr(self.parking_placed, f"parking{slot_nr}").get()
|
||||
return bool(val)
|
||||
# ── read ─────────────────────────────────────────────────────────────────
|
||||
def is_sample_slot_used(self, container: str, slot_nr: int = 0) -> bool:
|
||||
return bool(self._sig("used", slot_name(container, slot_nr)).get())
|
||||
|
||||
def is_sample_in_gripper(self) -> bool:
|
||||
val = self.sample_in_gripper.get()
|
||||
return bool(val)
|
||||
return self.is_sample_slot_used(GRIPPER)
|
||||
|
||||
def is_sample_in_samplestage(self) -> bool:
|
||||
val = self.sample_in_samplestage.get()
|
||||
return bool(val)
|
||||
return self.is_sample_slot_used(STAGE)
|
||||
|
||||
def get_sample_name(self, container, slot_nr) -> str:
|
||||
if container == "A":
|
||||
val = getattr(self.sample_shuttle_A_names, f"sample{slot_nr}").get()
|
||||
elif container == "B":
|
||||
val = getattr(self.sample_shuttle_B_names, f"sample{slot_nr}").get()
|
||||
elif container == "C":
|
||||
val = getattr(self.sample_shuttle_C_names, f"sample{slot_nr}").get()
|
||||
elif container == "O":
|
||||
val = getattr(self.sample_names, f"sample{slot_nr}").get()
|
||||
else:
|
||||
val = "unknown container"
|
||||
return str(val)
|
||||
def get_sample_name(self, container: str, slot_nr: int = 0) -> str:
|
||||
name = self._str(self._sig("name", slot_name(container, slot_nr)).get())
|
||||
return name if name else EMPTY_NAME
|
||||
|
||||
def get_shuttle_name_slot(self, slot_nr: int) -> str:
|
||||
val = getattr(self.parking_names, f"parking{slot_nr}").get()
|
||||
return str(val)
|
||||
def get_sample_owner(self, container: str, slot_nr: int = 0) -> str:
|
||||
return self._str(self._sig("owner", slot_name(container, slot_nr)).get())
|
||||
|
||||
def get_sample_name_and_owner(self, container: str, slot_nr: int = 0) -> tuple:
|
||||
return self.get_sample_name(container, slot_nr), self.get_sample_owner(container, slot_nr)
|
||||
|
||||
def get_sample_status(self, container: str, slot_nr: int = 0) -> str:
|
||||
status = self._str(self._sig("status", slot_name(container, slot_nr)).get())
|
||||
return status if status in STATUS_VALUES else STATUS_NONE
|
||||
|
||||
def get_sample_meas_date(self, container: str, slot_nr: int = 0) -> str:
|
||||
return self._str(self._sig("meas_date", slot_name(container, slot_nr)).get())
|
||||
|
||||
def get_sample_record(self, container: str, slot_nr: int = 0) -> dict:
|
||||
return {
|
||||
"used": self.is_sample_slot_used(container, slot_nr),
|
||||
"name": self.get_sample_name(container, slot_nr),
|
||||
"owner": self.get_sample_owner(container, slot_nr),
|
||||
"status": self.get_sample_status(container, slot_nr),
|
||||
"meas_date": self.get_sample_meas_date(container, slot_nr),
|
||||
}
|
||||
|
||||
def get_sample_name_in_gripper(self) -> str:
|
||||
val = self.sample_in_gripper_name.get()
|
||||
return str(val)
|
||||
return self.get_sample_name(GRIPPER)
|
||||
|
||||
def get_sample_name_in_samplestage(self) -> str:
|
||||
val = self.sample_in_samplestage_name.get()
|
||||
return str(val)
|
||||
return self.get_sample_name(STAGE)
|
||||
|
||||
def get_sample_owner_in_gripper(self) -> str:
|
||||
return self.get_sample_owner(GRIPPER)
|
||||
|
||||
def get_sample_owner_in_samplestage(self) -> str:
|
||||
return self.get_sample_owner(STAGE)
|
||||
|
||||
def get_sample_status_in_gripper(self) -> str:
|
||||
return self.get_sample_status(GRIPPER)
|
||||
|
||||
def get_sample_status_in_samplestage(self) -> str:
|
||||
return self.get_sample_status(STAGE)
|
||||
|
||||
def is_shuttle_slot_used(self, slot_nr: int) -> bool:
|
||||
return bool(getattr(self.parking_placed, f"parking{self._check_parking(slot_nr)}").get())
|
||||
|
||||
def get_shuttle_name_slot(self, slot_nr: int) -> str:
|
||||
return self._str(getattr(self.parking_names, f"parking{self._check_parking(slot_nr)}").get())
|
||||
|
||||
def search_shuttle_in_slot(self, shuttle: str) -> int:
|
||||
returnvalue = 0
|
||||
for i in range(1, 7):
|
||||
for i in PARKING_SLOTS:
|
||||
if self.get_shuttle_name_slot(i) == shuttle:
|
||||
returnvalue = i
|
||||
return returnvalue
|
||||
|
||||
def _describe_slot(self, container: str, slot_nr: int) -> str:
|
||||
def _describe_slot(self, container: str, slot_nr: int = 0) -> str:
|
||||
"""Sample name, plus owner/measured-status suffix when occupied --
|
||||
e.g. "sample1 (bob) ◐ started". Used by show_all() so the
|
||||
e.g. "sample1 (e12345) ◐ started 2026-10". 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)
|
||||
@@ -533,13 +331,12 @@ class OMNYSampleStorage(Device):
|
||||
return name
|
||||
owner = self.get_sample_owner(container, slot_nr)
|
||||
status = self.get_sample_status(container, slot_nr)
|
||||
date = self.get_sample_meas_date(container, slot_nr)
|
||||
suffix = f" ({owner})" if owner else ""
|
||||
if status == "started":
|
||||
account = self.get_sample_account(container, slot_nr)
|
||||
suffix += f" ◐ started (acc {account})"
|
||||
elif status == "completed":
|
||||
account = self.get_sample_account(container, slot_nr)
|
||||
suffix += f" ✓ completed (acc {account})"
|
||||
if status == STATUS_STARTED:
|
||||
suffix += " ◐ started" + (f" {date}" if date else "")
|
||||
elif status == STATUS_COMPLETED:
|
||||
suffix += " ✓ completed" + (f" {date}" if date else "")
|
||||
return name + suffix
|
||||
|
||||
def show_all(self):
|
||||
@@ -547,14 +344,13 @@ class OMNYSampleStorage(Device):
|
||||
red = "\x1b[91m"
|
||||
green = "\x1b[92m"
|
||||
white = "\x1b[0m"
|
||||
for ch in ["A", "B", "C"]:
|
||||
for ch in SHUTTLES:
|
||||
t.clear()
|
||||
shuttle_in_slot = self.search_shuttle_in_slot(ch)
|
||||
if shuttle_in_slot > 0:
|
||||
t.title = green + "Shuttle " + ch + " in OMNY slot " + str(shuttle_in_slot) + white
|
||||
else:
|
||||
t.title = red + "Shuttle " + ch + white
|
||||
field_names = [""]
|
||||
for ax in [1, 3, 5]:
|
||||
row = []
|
||||
row.extend([self._describe_slot(ch, ax)])
|
||||
@@ -567,28 +363,18 @@ class OMNYSampleStorage(Device):
|
||||
print(t)
|
||||
|
||||
if self.is_sample_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 += f" ◐ started (acc {self.get_sample_account_in_samplestage()})"
|
||||
elif status == "completed":
|
||||
suffix += f" ✓ completed (acc {self.get_sample_account_in_samplestage()})"
|
||||
print(f"\n\n Sample stage: {name}{suffix}")
|
||||
print(f"\n\n Sample stage: {self._describe_slot(STAGE)}")
|
||||
else:
|
||||
print(f"\n\n Sample stage: no sample")
|
||||
|
||||
if self.is_sample_in_gripper():
|
||||
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")
|
||||
print(f"\n Gripper: {self._describe_slot(GRIPPER)}\n")
|
||||
else:
|
||||
print(f"\n Gripper: no sample\n")
|
||||
|
||||
t.clear()
|
||||
t.title = "Fixed positions in OMNY"
|
||||
for i in [10, 11, 12, 13, 14, 32, 33, 34, 100, 101]:
|
||||
for i in FIXED_SLOTS:
|
||||
row = []
|
||||
row.extend([f"Position {i:3d}"])
|
||||
if self.is_sample_slot_used("O", i):
|
||||
@@ -608,12 +394,10 @@ 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',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.get_sample_record('system',position)")
|
||||
print(" dev.omny_samples.set_sample_measured('system',position,'status','YYYY-MM')")
|
||||
print(" status is one of 'none', 'started', 'completed'; system 'STAGE' for the sample stage")
|
||||
print(" dev.omny_samples.transfer_sample(('A',3), 'GRIPPER') (name, owner, status move along)")
|
||||
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)")
|
||||
|
||||
@@ -28,7 +28,11 @@ from ophyd_devices.tests.utils import patched_device
|
||||
|
||||
from csaxs_bec.devices.omny.galil.ogalil_ophyd import OMNYGalilMotor
|
||||
from csaxs_bec.devices.omny.omny_dewar import OMNYDewar
|
||||
from csaxs_bec.devices.omny.omny_sample_storage import OMNYSampleStorage
|
||||
from csaxs_bec.devices.omny.omny_sample_storage import (
|
||||
PARKING_SLOTS,
|
||||
SAMPLE_SLOTS,
|
||||
OMNYSampleStorage,
|
||||
)
|
||||
from csaxs_bec.devices.omny.omny_temperatures import OMNYTemperatures
|
||||
from csaxs_bec.devices.omny.rt.rt_omny_ophyd import RtOMNYFlyer, RtOMNYMotor
|
||||
from csaxs_bec.devices.sim.sim_galil import SimGalilSocket, SimGalilState, _sim_state_kwargs
|
||||
@@ -262,53 +266,11 @@ class SimOMNYSampleStorage(OMNYSampleStorage):
|
||||
# real backend's own empty sentinels ("-" for sample slots, "none"
|
||||
# for parking) -- explicitly unset every slot first, the same way
|
||||
# SimFlomniSampleStorage does, so untouched slots read back as
|
||||
# genuinely empty instead of a stray "0" name/shuttle-letter.
|
||||
#
|
||||
# Writes go directly to the Dcpt signals rather than through
|
||||
# unset_sample_slot()/set_sample_slot() -- those reject any
|
||||
# slot_nr > 20 (a pre-existing bug: it's copied from Flomni's
|
||||
# 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.
|
||||
# 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")
|
||||
storage.unset_sample_in_gripper()
|
||||
# genuinely empty (owner "", status "none") instead of a stray "0".
|
||||
for slot in SAMPLE_SLOTS:
|
||||
storage.unset_sample_slot(slot, 0)
|
||||
for slot_nr in PARKING_SLOTS:
|
||||
storage.unset_shuttle_slot(slot_nr)
|
||||
storage.set_sample_in_samplestage("sim_stage_sample")
|
||||
for slot_nr, name in samples.items():
|
||||
storage.set_sample_slot("A", int(slot_nr), str(name))
|
||||
|
||||
@@ -89,7 +89,7 @@ Day to day, sample storage is viewed and adjusted through the __OMNY Sample Stor
|
||||
The widget shows every position (shuttle pins A/B/C, fixed "O" positions, the gripper, and the sample stage) with its occupied state, sample name, owner, and measured status (unmeasured / ◐ started / ✓ completed). The parking-carousel view also shows which shuttle is in which parking slot and marks the currently __ACTIVE__ one (oparkz is physically sitting there, ready for transfer).
|
||||
|
||||
Right-click a slot for:
|
||||
- __Change sample…__ — set a new name and owner for an occupied slot (replaces the sample record, including resetting its measured status back to unmeasured — use this when a physically different sample is now there, not to correct a typo in the current one's account).
|
||||
- __Change sample…__ — set a new name and owner for an occupied slot (replaces the sample record, including resetting its measured status back to unmeasured — use this when a physically different sample is now there, not to correct a typo in the current one's record).
|
||||
- __Clear (set empty)__ — mark a slot empty after physically removing the sample. Only offered for shuttle slots, since only samples in a shuttle can leave vacuum: this also offers to print a P-touch label for the sample that was just removed.
|
||||
- __New sample…__ — occupy an empty slot with a name and owner.
|
||||
- __Set parking slot…__ / __Clear parking assignment__ (shuttle labels) — record which parking slot a shuttle occupies; a slot already held by a different shuttle is rejected, clear it there first.
|
||||
@@ -145,10 +145,11 @@ The shuttles are mounted in the parking station. Typically oparkz slot 3 is used
|
||||
`dev.omny_samples.set_shuttle_slot(container, slot_nr)`
|
||||
Here is an _example_: dev.omny_samples.set_shuttle_slot('A',2)
|
||||
|
||||
Measured status/account (normally set automatically during a tomographic scan) can be read and, if really needed, corrected manually via the `omny.otransfer_*` wrappers, e.g.:
|
||||
Measured status and date (set automatically when a tomographic scan starts and finishes: "started"/"completed" with the month) travel with the sample on every transfer. They can be read and, if really needed, corrected manually:
|
||||
`dev.omny_samples.get_sample_record('system', position)` returns used/name/owner/status/meas_date (`'STAGE'`/`'GRIPPER'` as system for the sample stage and the gripper)
|
||||
`dev.omny_samples.set_sample_measured('system', position, status, 'YYYY-MM')` (`status` is one of `"none"`/`"started"`/`"completed"`; also as `omny.otransfer_set_sample_measured`)
|
||||
`omny.otransfer_get_sample_status('system', position)` / `otransfer_get_sample_status_in_gripper()` / `otransfer_get_sample_status_in_samplestage()`
|
||||
`omny.otransfer_set_sample_status('system', position, status, account)` (`status` is one of `"none"`/`"started"`/`"completed"`) and the matching `_in_gripper`/`_in_samplestage` variants
|
||||
`omny.otransfer_get_sample_account(...)` and its `_in_gripper`/`_in_samplestage` variants read back the account tied to a "started"/"completed" status.
|
||||
Each position is one record on the OMNY soft IOC, `OMNY-SAMPLE:<slot>:USED/NAME/OWNER/STATUS/MEAS_DATE` (slots `A1`…`C6`, `P10`…`P101`, `STAGE`, `GRIPPER`, `PARK1`…`PARK6`).
|
||||
|
||||
### Status of OMNY
|
||||
|
||||
|
||||
@@ -301,7 +301,10 @@ def test_tomo_parameters_forces_type_1_for_non_superuser(monkeypatch, fake_bec):
|
||||
omny.tomo_angle_stepsize = 10.0
|
||||
omny.tomo_type = 1
|
||||
fake_dev = types.SimpleNamespace(
|
||||
omny_samples=types.SimpleNamespace(get_sample_name_in_samplestage=lambda: "test_sample")
|
||||
omny_samples=types.SimpleNamespace(
|
||||
get_sample_name_in_samplestage=lambda: "test_sample",
|
||||
set_sample_measured=lambda *a, **k: None,
|
||||
)
|
||||
)
|
||||
monkeypatch.setattr(omny_module, "dev", fake_dev, raising=False)
|
||||
omny.OMNYTools = types.SimpleNamespace(yesno=lambda *a, **k: False)
|
||||
@@ -335,7 +338,10 @@ def test_tomo_parameters_allows_superuser_to_unlock_8_subtomo_mode(monkeypatch,
|
||||
omny.tomo_angle_stepsize = 10.0
|
||||
omny.tomo_type = 1
|
||||
fake_dev = types.SimpleNamespace(
|
||||
omny_samples=types.SimpleNamespace(get_sample_name_in_samplestage=lambda: "test_sample")
|
||||
omny_samples=types.SimpleNamespace(
|
||||
get_sample_name_in_samplestage=lambda: "test_sample",
|
||||
set_sample_measured=lambda *a, **k: None,
|
||||
)
|
||||
)
|
||||
monkeypatch.setattr(omny_module, "dev", fake_dev, raising=False)
|
||||
omny.OMNYTools = types.SimpleNamespace(yesno=lambda *a, **k: False)
|
||||
@@ -371,7 +377,10 @@ def test_tomo_parameters_clamps_type_3_step_size_to_max_2_5_degrees(monkeypatch,
|
||||
omny.tomo_angle_stepsize = 10.0
|
||||
omny.tomo_type = 1
|
||||
fake_dev = types.SimpleNamespace(
|
||||
omny_samples=types.SimpleNamespace(get_sample_name_in_samplestage=lambda: "test_sample")
|
||||
omny_samples=types.SimpleNamespace(
|
||||
get_sample_name_in_samplestage=lambda: "test_sample",
|
||||
set_sample_measured=lambda *a, **k: None,
|
||||
)
|
||||
)
|
||||
monkeypatch.setattr(omny_module, "dev", fake_dev, raising=False)
|
||||
omny.OMNYTools = types.SimpleNamespace(yesno=lambda *a, **k: False)
|
||||
@@ -413,7 +422,10 @@ def test_tomo_parameters_warns_when_fermat_positions_below_minimum(monkeypatch,
|
||||
omny.tomo_angle_stepsize = 10.0
|
||||
omny.tomo_type = 1
|
||||
fake_dev = types.SimpleNamespace(
|
||||
omny_samples=types.SimpleNamespace(get_sample_name_in_samplestage=lambda: "test_sample")
|
||||
omny_samples=types.SimpleNamespace(
|
||||
get_sample_name_in_samplestage=lambda: "test_sample",
|
||||
set_sample_measured=lambda *a, **k: None,
|
||||
)
|
||||
)
|
||||
monkeypatch.setattr(omny_module, "dev", fake_dev, raising=False)
|
||||
omny.OMNYTools = types.SimpleNamespace(yesno=lambda *a, **k: True) # accept as-is, no edits
|
||||
@@ -440,7 +452,10 @@ def test_tomo_parameters_no_warning_when_fermat_positions_sufficient(monkeypatch
|
||||
omny.tomo_angle_stepsize = 10.0
|
||||
omny.tomo_type = 1
|
||||
fake_dev = types.SimpleNamespace(
|
||||
omny_samples=types.SimpleNamespace(get_sample_name_in_samplestage=lambda: "test_sample")
|
||||
omny_samples=types.SimpleNamespace(
|
||||
get_sample_name_in_samplestage=lambda: "test_sample",
|
||||
set_sample_measured=lambda *a, **k: None,
|
||||
)
|
||||
)
|
||||
monkeypatch.setattr(omny_module, "dev", fake_dev, raising=False)
|
||||
omny.OMNYTools = types.SimpleNamespace(yesno=lambda *a, **k: True)
|
||||
|
||||
@@ -9,6 +9,8 @@ grid: OMNY has three "equally spaced sub-tomograms" flavors (tomo_type
|
||||
N*n_subtomos total, where n_subtomos = {1: 2, 4: 4, 5: 8}[tomo_type].
|
||||
"""
|
||||
|
||||
import types
|
||||
|
||||
import pytest
|
||||
|
||||
from csaxs_bec.bec_ipython_client.plugins.omny.omny import OMNY
|
||||
@@ -245,7 +247,7 @@ def test_omny_sample_name_getter_reads_signal_not_rpc_method():
|
||||
|
||||
class _FakeOmnySamples:
|
||||
def __init__(self):
|
||||
self.sample_in_samplestage_name = _FakeSignal("my_sample")
|
||||
self.sample_names = types.SimpleNamespace(STAGE=_FakeSignal("my_sample"))
|
||||
|
||||
def get_sample_name_in_samplestage(self):
|
||||
raise AssertionError(
|
||||
@@ -259,7 +261,7 @@ def test_omny_sample_name_getter_reads_signal_not_rpc_method():
|
||||
class _FakeWidget:
|
||||
dev = _FakeDev()
|
||||
|
||||
signal = _FakeDev.omny_samples.sample_in_samplestage_name
|
||||
signal = _FakeDev.omny_samples.sample_names.STAGE
|
||||
result = SETUP_PROFILES["omny"]["sample_name_getter"](_FakeWidget())
|
||||
|
||||
assert result == "my_sample"
|
||||
|
||||
@@ -45,7 +45,10 @@ def make_omny(fake_client=None):
|
||||
# real BEC session only) -- stand in a fake so tomo_queue_execute()'s
|
||||
# confirmation prompt (which reads self.sample_name) doesn't NameError.
|
||||
omny_module.dev = types.SimpleNamespace(
|
||||
omny_samples=types.SimpleNamespace(get_sample_name_in_samplestage=lambda: "test_sample")
|
||||
omny_samples=types.SimpleNamespace(
|
||||
get_sample_name_in_samplestage=lambda: "test_sample",
|
||||
set_sample_measured=lambda *a, **k: None,
|
||||
)
|
||||
)
|
||||
|
||||
obj = object.__new__(OMNY)
|
||||
|
||||
@@ -57,7 +57,10 @@ def make_omny_for_tomo_scan(monkeypatch, tomo_angle_stepsize: float, active_acco
|
||||
# real BEC session only) -- stand in a fake so tomo_scan()'s sample-
|
||||
# database registration (which reads self.sample_name) doesn't NameError.
|
||||
omny_module.dev = types.SimpleNamespace(
|
||||
omny_samples=types.SimpleNamespace(get_sample_name_in_samplestage=lambda: "test_sample")
|
||||
omny_samples=types.SimpleNamespace(
|
||||
get_sample_name_in_samplestage=lambda: "test_sample",
|
||||
set_sample_measured=lambda *a, **k: None,
|
||||
)
|
||||
)
|
||||
|
||||
obj = object.__new__(OMNY)
|
||||
@@ -525,7 +528,10 @@ def test_write_tomo_scan_number_creates_missing_directory(monkeypatch, tmp_path)
|
||||
omny.tomo_id = 1
|
||||
fake_dev = types.SimpleNamespace(
|
||||
osamroy=types.SimpleNamespace(read=lambda: {"osamroy": {"value": 12.5}}),
|
||||
omny_samples=types.SimpleNamespace(get_sample_name_in_samplestage=lambda: "test_sample"),
|
||||
omny_samples=types.SimpleNamespace(
|
||||
get_sample_name_in_samplestage=lambda: "test_sample",
|
||||
set_sample_measured=lambda *a, **k: None,
|
||||
),
|
||||
)
|
||||
monkeypatch.setattr(omny_module, "dev", fake_dev, raising=False)
|
||||
|
||||
|
||||
@@ -1,47 +1,90 @@
|
||||
"""OMNYSampleStorage on the OMNY soft IOC records (OMNY-SAMPLE:<slot>:USED/NAME/OWNER/STATUS/MEAS_DATE)."""
|
||||
|
||||
import pytest
|
||||
|
||||
from csaxs_bec.devices.omny.omny_sample_storage import OMNYSampleStorage, OMNYSampleStorageError
|
||||
from csaxs_bec.devices.omny.omny_sample_storage import (
|
||||
SAMPLE_SLOTS,
|
||||
OMNYSampleStorage,
|
||||
OMNYSampleStorageError,
|
||||
slot_name,
|
||||
)
|
||||
from csaxs_bec.devices.sim.sim_omny import SimOMNYSampleStorage
|
||||
|
||||
|
||||
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.fixture
|
||||
def storage():
|
||||
dev = SimOMNYSampleStorage(name="omny_samples", sim_samples={})
|
||||
try:
|
||||
yield dev
|
||||
finally:
|
||||
dev.destroy()
|
||||
|
||||
|
||||
@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(
|
||||
"container, slot_nr, expected",
|
||||
[("A", 1, "A1"), ("C", 6, "C6"), ("O", 10, "P10"), ("O", 101, "P101"), ("STAGE", 0, "STAGE"),
|
||||
("gripper", 0, "GRIPPER"), ("B4", 0, "B4"), ("P32", 0, "P32")],
|
||||
)
|
||||
def test_slot_name(container, slot_nr, expected):
|
||||
assert slot_name(container, slot_nr) == expected
|
||||
|
||||
|
||||
@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():
|
||||
@pytest.mark.parametrize("container, slot_nr", [("A", 0), ("A", 7), ("O", 15), ("O", 1), ("D", 1)])
|
||||
def test_slot_name_rejects_unknown_slots(container, slot_nr):
|
||||
with pytest.raises(OMNYSampleStorageError):
|
||||
OMNYSampleStorage._status_to_val("bogus")
|
||||
slot_name(container, slot_nr)
|
||||
|
||||
|
||||
@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"
|
||||
def test_pv_names_match_the_ioc():
|
||||
assert OMNYSampleStorage.sample_used.defn["A1"][1] == "OMNY-SAMPLE:A1:USED"
|
||||
assert OMNYSampleStorage.sample_names.defn["P100"][1] == "OMNY-SAMPLE:P100:NAME"
|
||||
assert OMNYSampleStorage.sample_status.defn["STAGE"][1] == "OMNY-SAMPLE:STAGE:STATUS"
|
||||
assert OMNYSampleStorage.sample_meas_date.defn["GRIPPER"][1] == "OMNY-SAMPLE:GRIPPER:MEAS_DATE"
|
||||
assert OMNYSampleStorage.parking_names.defn["parking3"][1] == "OMNY-SAMPLE:PARK3:NAME"
|
||||
assert len(SAMPLE_SLOTS) == 18 + 10 + 2
|
||||
|
||||
|
||||
@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
|
||||
def test_set_and_unset_slot(storage):
|
||||
storage.set_sample_slot("O", 32, "pin1", "e12345")
|
||||
assert storage.get_sample_record("O", 32) == {
|
||||
"used": True, "name": "pin1", "owner": "e12345", "status": "none", "meas_date": ""}
|
||||
storage.unset_sample_slot("O", 32)
|
||||
rec = storage.get_sample_record("O", 32)
|
||||
assert rec["used"] is False and rec["name"] == "-" and rec["owner"] == ""
|
||||
|
||||
|
||||
def test_status_travels_with_the_sample(storage):
|
||||
storage.set_sample_slot("B", 2, "pin2", "e11111")
|
||||
storage.set_sample_measured("B", 2, "completed", "2026-10")
|
||||
storage.transfer_sample(("B", 2), "GRIPPER")
|
||||
storage.transfer_sample("GRIPPER", "STAGE")
|
||||
assert not storage.is_sample_slot_used("B", 2)
|
||||
assert not storage.is_sample_in_gripper()
|
||||
assert storage.get_sample_record("STAGE") == {
|
||||
"used": True, "name": "pin2", "owner": "e11111", "status": "completed", "meas_date": "2026-10"}
|
||||
|
||||
|
||||
def test_set_sample_measured_owner_if_empty(storage):
|
||||
storage.set_sample_in_samplestage("pin3")
|
||||
storage.set_sample_measured("STAGE", 0, "started", "2026-10", owner_if_empty="e22222")
|
||||
assert storage.get_sample_owner_in_samplestage() == "e22222"
|
||||
assert storage.get_sample_status_in_samplestage() == "started"
|
||||
# an existing owner is kept
|
||||
storage.set_sample_measured("STAGE", 0, "completed", "2026-10", owner_if_empty="e33333")
|
||||
assert storage.get_sample_owner_in_samplestage() == "e22222"
|
||||
|
||||
|
||||
def test_invalid_status_raises(storage):
|
||||
storage.set_sample_slot("A", 1, "pin4")
|
||||
with pytest.raises(OMNYSampleStorageError):
|
||||
storage.set_sample_measured("A", 1, "bogus", "")
|
||||
|
||||
|
||||
def test_parking(storage):
|
||||
storage.set_shuttle_slot("C", 2)
|
||||
assert storage.is_shuttle_slot_used(2) and storage.get_shuttle_name_slot(2) == "C"
|
||||
assert storage.search_shuttle_in_slot("C") == 2
|
||||
storage.unset_shuttle_slot(2)
|
||||
assert not storage.is_shuttle_slot_used(2) and storage.get_shuttle_name_slot(2) == "none"
|
||||
with pytest.raises(OMNYSampleStorageError):
|
||||
storage.set_shuttle_slot("A", 7)
|
||||
@@ -29,3 +29,10 @@ def test_default_samples_still_seeded_in_shuttle_a(storage):
|
||||
assert storage.get_sample_name("A", 1) == "sim_sample_1"
|
||||
assert storage.get_sample_name("A", 2) == "sim_sample_2"
|
||||
assert storage.get_sample_name("A", 3) == "sim_sample_3"
|
||||
|
||||
|
||||
def test_untouched_slots_read_empty(storage):
|
||||
rec = storage.get_sample_record("O", 100)
|
||||
assert rec == {"used": False, "name": "-", "owner": "", "status": "none", "meas_date": ""}
|
||||
assert storage.get_sample_name_in_samplestage() == "sim_stage_sample"
|
||||
assert not storage.is_sample_in_gripper()
|
||||
Reference in new issue
Block a user