bec/csaxs_bec
0
0 Fork 3
Code Issues 17 Pull Requests 3 Actions Packages Projects Releases Wiki Activity

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

Merged
holler merged 1 commits from feat/flomni-sample-ioc into main 2026-10-17 23:34:17 +02:00
Conversation 0 Commits 1 Files Changed 7
+293 -339
7 changed files with 293 additions and 339 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py
+28 -124
View File
@@ -42,7 +42,6 @@ from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.omny_general_tools import
)
from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.ptouch_printer import PTouchLabelPrinter
from csaxs_bec.devices.omny.galil.galil_ophyd import GalilError
from csaxs_bec.devices.omny.sample_desc_codec import pack_desc, unpack_desc
logger = bec_logger.logger
@@ -818,13 +817,11 @@ class FlomniSampleTransferMixin:
self.ftransfer_controller_disable_mount_mode()
self.ensure_gripper_up()
signal_name = getattr(dev.flomni_samples.sample_names, f"sample{position}")
name, owner = unpack_desc(signal_name.get())
self.flomni_modify_storage_non_interactive(100, 1, name, owner=owner)
self.flomni_modify_storage_non_interactive(position, 0, "-")
# the whole record (name, owner, measurement status/date) moves to the gripper
dev.flomni_samples.transfer_sample(position, 100)
def ftransfer_show_all(self):
dev.flomni_samples.show_all(measured=self.measured_log.snapshot())
dev.flomni_samples.show_all()
def ftransfer_put_sample(self, position: int):
self.check_position_is_valid(position)
@@ -877,9 +874,8 @@ class FlomniSampleTransferMixin:
self.ftransfer_controller_disable_mount_mode()
self.ensure_gripper_up()
sample_name, sample_owner = unpack_desc(dev.flomni_samples.sample_in_gripper_name.get())
self.flomni_modify_storage_non_interactive(100, 0, "-")
self.flomni_modify_storage_non_interactive(position, 1, sample_name, owner=sample_owner)
# the whole record (name, owner, measurement status/date) moves from the gripper
dev.flomni_samples.transfer_sample(100, position)
if position == 0:
self.ftransfer_flomni_stage_in()
@@ -889,18 +885,7 @@ class FlomniSampleTransferMixin:
"""
Get the name of the sample currently in the given position.
"""
signal_name = getattr(dev.flomni_samples.sample_names, f"sample{position}")
return unpack_desc(signal_name.get())[0]
def sample_get_measured_log_key(self, position: int = 0) -> str:
"""Packed `name | owner` identity key for the sample at `position`,
for use with `self.measured_log` -- NOT the same as `sample_name`
(a `sample_get_name()` alias), which is name-only and would silently
under-key the measured log against a different sample of the same
name but a different owner.
"""
name, owner = dev.flomni_samples.get_sample_name_and_owner(position)
return pack_desc(name, owner)
return dev.flomni_samples.get_sample_name(position)
def ftransfer_sample_change(self, new_sample_position: int):
self.check_tray_in()
@@ -1039,23 +1024,22 @@ class FlomniSampleTransferMixin:
print offer would incorrectly fire on every ordinary transfer step
instead of only on a genuine manual clear/replace.
"""
if not dev.flomni_samples.is_sample_slot_used(position):
return
prev_name, prev_owner = dev.flomni_samples.get_sample_name_and_owner(position)
prev_packed = pack_desc(prev_name, prev_owner)
record = self.measured_log.pop(prev_packed)
if record is None:
# measurement status/date/owner live on the sample record in the IOC
# (FLOMNI-SAMPLE:<slot>:STATUS/MEAS_DATE/OWNER) since 2026-10-17
record = dev.flomni_samples.get_sample_record(position)
if not record["used"] or record["status"] == "none":
return
prev_name = record["name"]
if self.OMNYTools.yesno(
f"Sample '{prev_name}' was measured ({record['status']}) on"
f" {record['measured_date']}. Print a label?",
f" {record['meas_date']}. Print a label?",
"y",
):
self.label_printer.ensure_ready()
printed = self.label_printer.print_label(
{
self._PTOUCH_FIELD_ACCOUNT: record["account"],
self._PTOUCH_FIELD_DATE: record["measured_date"],
self._PTOUCH_FIELD_ACCOUNT: record["owner"],
self._PTOUCH_FIELD_DATE: record["meas_date"],
self._PTOUCH_FIELD_SAMPLENAME: prev_name,
}
)
@@ -1068,15 +1052,11 @@ class FlomniSampleTransferMixin:
def flomni_modify_storage_non_interactive(
self, position: int, used: int, name: str, owner: str = ""
):
packed_name = pack_desc(name, owner)
if position == 100:
dev.flomni_samples.sample_in_gripper.set(used).wait(timeout=5)
dev.flomni_samples.sample_in_gripper_name.set(packed_name).wait(timeout=5)
# position 100 = gripper; a new/renamed sample starts with measurement status "none"
if used:
dev.flomni_samples.set_sample_slot(position, name, owner=owner)
else:
signal = getattr(dev.flomni_samples.sample_placed, f"sample{position}")
signal.set(used).wait(timeout=5)
signal_name = getattr(dev.flomni_samples.sample_names, f"sample{position}")
signal_name.set(packed_name).wait(timeout=5)
dev.flomni_samples.unset_sample_slot(position)
def check_position_is_valid(self, position: int):
if 0 <= position < 21:
@@ -1710,83 +1690,6 @@ class _ProgressProxy:
return self._load()
class _MeasuredSampleLog:
"""Tracks which flomni samples have been measured, for the P-touch
label-printing feature (print an account/date/name label when a
measured sample is cleared or replaced in tray storage).
Keyed by the packed ``name | owner`` string (see pack_desc/unpack_desc
in sample_desc_codec.py) rather than by slot number, since that packed
string is the identity that already travels with a sample across
ftransfer_get_sample/ftransfer_put_sample transfers between tray
slots, the gripper, and the stage -- tracking by slot number would
require threading a flag through every one of those transfer calls.
Persisted as a BEC global var (same pattern as _ProgressProxy above)
so it's visible to any client session -- CLI or GUI widget -- and
survives a kernel restart.
Two statuses: "started" (a fresh tomogram began for this sample) and
"completed" (that tomogram's scan loop ran to completion without
error). Both count as "measured" for is_measured()/the print offer --
"started" without "completed" just means the tomogram may be
incomplete, which callers surface in the offer wording rather than
suppressing the offer entirely.
"""
_GLOBAL_VAR_KEY = "flomni_measured_samples"
def __init__(self, client):
self._client = client
def _load(self) -> dict:
val = self._client.get_global_var(self._GLOBAL_VAR_KEY)
return val if val is not None else {}
def _save(self, data: dict) -> None:
self._client.set_global_var(self._GLOBAL_VAR_KEY, data)
def mark_measured(self, sample_key: str, account: str, date: str, status: str) -> None:
"""Record `sample_key` as measured, overwriting any existing record.
Overwriting (rather than merging) means re-measuring a sample from
scratch correctly resets a stale "completed" back to "started".
"""
data = self._load()
data[sample_key] = {"account": account, "measured_date": date, "status": status}
self._save(data)
def is_measured(self, sample_key: str) -> bool:
return sample_key in self._load()
def get(self, sample_key: str) -> dict | None:
return self._load().get(sample_key)
def snapshot(self) -> dict:
"""One bulk fetch of the whole log, as a plain dict.
Used instead of `get` when the caller needs to hand the data to
`dev.flomni_samples.show_all()`, which runs via BEC's
device-server RPC mechanism -- every argument gets msgpack-
serialized, so a bound method (`self.measured_log.get`) can't be
passed directly; a plain dict can.
"""
return self._load()
def pop(self, sample_key: str) -> dict | None:
"""Remove and return the record for `sample_key`, if any.
Called once the print offer for that sample has been resolved
(printed or declined), so the log stays bounded to
currently-measured-but-not-yet-cleared samples instead of growing
unbounded over a beamtime.
"""
data = self._load()
record = data.pop(sample_key, None)
if record is not None:
self._save(data)
return record
class Flomni(
TomoQueueMixin,
TomoAngleFileMixin,
@@ -1843,7 +1746,6 @@ class Flomni(
self.reconstructor = PtychoReconstructor(self.ptycho_reconstruct_foldername)
self.tomo_id_manager = TomoIDManager()
self.align = XrayEyeAlign(self.client, self)
self.measured_log = _MeasuredSampleLog(self.client)
self.label_printer = PTouchLabelPrinter(host="BRN94DDF8AAB8EC.psi.ch", template_number=1)
self.set_client(client)
@@ -2842,11 +2744,12 @@ class Flomni(
"test additional info",
"BEC",
)
self.measured_log.mark_measured(
self.sample_get_measured_log_key(0),
bec.active_account or "",
# measurement status on the sample record (stage slot 0) in the IOC
dev.flomni_samples.set_sample_measured(
0,
"started",
datetime.date.today().strftime("%Y-%m"),
status="started",
owner_if_empty=bec.active_account or "",
)
self.write_pdf_report()
# estimated_remaining_time/estimated_finish_time/accumulated_idle_time/
@@ -3035,11 +2938,12 @@ class Flomni(
self._print_progress()
self._log_tomogram_timing()
self.OMNYTools.printgreenbold("Tomoscan finished")
self.measured_log.mark_measured(
self.sample_get_measured_log_key(0),
bec.active_account or "",
# measurement status on the sample record (stage slot 0) in the IOC
dev.flomni_samples.set_sample_measured(
0,
"completed",
datetime.date.today().strftime("%Y-%m"),
status="completed",
owner_if_empty=bec.active_account or "",
)
timing_lines = self._build_timing_lines()
for line in timing_lines[3:]:
csaxs_bec/bec_ipython_client/plugins/flomni/flomni_webpage_generator.py
+8 -12
View File
@@ -43,7 +43,6 @@ from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.webpage_generator_base imp
_safe_get,
_safe_float,
)
from csaxs_bec.devices.omny.sample_desc_codec import unpack_desc
class FlomniWebpageGenerator(WebpageGeneratorBase):
@@ -106,19 +105,16 @@ class FlomniWebpageGenerator(WebpageGeneratorBase):
# ── Sample name ───────────────────────────────────────────────
# slot 0 of flomni_samples is the flOMNI measurement stage itself
# (see FlomniSampleStorage.show_all()'s own "flOMNI stage:" label
# for slot 0 -- not an arbitrary tray position). The raw DESC value
# is packed as "name | owner" (sample_desc_codec.pack_desc, added
# 7e8e807) -- unpack it here too, the same way
# FlomniSampleStorage.get_sample_name_and_owner() and show_all()
# already do, so the page doesn't show the raw delimiter-joined
# string as the "sample name". Owner is appended in parentheses,
# mirroring show_all()'s own "(owner: ...)" formatting.
raw_sample = _safe_get(dm, "flomni_samples.sample_names.sample0")
if raw_sample is None:
# for slot 0 -- not an arbitrary tray position). Name and owner are
# separate IOC fields since 2026-10-17 (FLOMNI-SAMPLE:STAGE:NAME /
# :OWNER); owner is appended in parentheses, mirroring show_all()'s
# own "(owner: ...)" formatting.
name = _safe_get(dm, "flomni_samples.sample_names.sample0")
if name is None:
sample_name = "N/A"
else:
name, owner = unpack_desc(raw_sample)
sample_name = f"{name} (owner: {owner})" if owner else name
owner = _safe_get(dm, "flomni_samples.sample_owners.sample0")
sample_name = f"{name} (owner: {owner})" if owner else str(name)
# ── Temperatures ──────────────────────────────────────────────
temperatures = {
csaxs_bec/bec_widgets/widgets/sample_storage/sample_storage.py
+40 -105
View File
@@ -29,10 +29,14 @@ single source of truth also consulted by the transfer state machine. The
widget reads/writes it directly, mirroring
``Flomni.flomni_modify_storage_non_interactive()``:
slots 0–20 : sample_placed.sample{N}.set(used)
sample_names.sample{N}.set(name)
gripper : sample_in_gripper.set(used)
sample_in_gripper_name.set(name)
writes : set_sample_slot(N, name, owner) / unset_sample_slot(N)
(N = 0–20, 100 = gripper; device methods, all record fields)
reads : one bulk read() of sample_placed / sample_names /
sample_owners / sample_status (gripper: sample_in_gripper*)
Since 2026-10-17 every slot is one record on the flOMNI soft IOC
(FLOMNI-SAMPLE:<slot>:USED/NAME/OWNER/STATUS/MEAS_DATE); the measurement
status travels with the sample and is no longer kept in a BEC global var.
Polling
-------
@@ -69,7 +73,6 @@ from qtpy.QtWidgets import (
)
from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.ptouch_printer import PTouchLabelPrinter
from csaxs_bec.devices.omny.sample_desc_codec import pack_desc, unpack_desc
logger = bec_logger.logger
@@ -115,54 +118,6 @@ PTOUCH_FIELD_SAMPLENAME = "SAMPLENAME"
PRINTER_NAME = "WSLA_X12SA"
# ── measured-sample log ──────────────────────────────────────────────────────
class _MeasuredSampleLog:
"""Tracks which flomni samples have been measured, for the P-touch
label-printing feature.
Deliberately duplicated here rather than imported from
``flomni.py``'s ``_MeasuredSampleLog`` -- pulling in that whole plugin
module (CLI-only dependencies, module-level ``bec``/``dev`` builtins
access) into this widget's gui-server process is a heavier and more
fragile coupling than re-declaring this ~15-line global-var wrapper,
and this widget already duplicates ``flomni_modify_storage_non_interactive()``
the same way (see module docstring). Both copies read/write the same
``flomni_measured_samples`` global-var key, so they stay consistent in
effect despite being independent code, exactly like the existing
``_write_slot``/``flomni_modify_storage_non_interactive`` duplication.
"""
_GLOBAL_VAR_KEY = "flomni_measured_samples"
def __init__(self, client):
self._client = client
def _load(self) -> dict:
val = self._client.get_global_var(self._GLOBAL_VAR_KEY)
return val if val is not None else {}
def _save(self, data: dict) -> None:
self._client.set_global_var(self._GLOBAL_VAR_KEY, data)
def get(self, sample_key: str) -> Optional[dict]:
return self._load().get(sample_key)
def snapshot(self) -> dict:
"""One bulk fetch of the whole log, for callers (like the poll
loop below) that need to look up many sample keys at once without
one redis round-trip per key."""
return self._load()
def pop(self, sample_key: str) -> Optional[dict]:
data = self._load()
record = data.pop(sample_key, None)
if record is not None:
self._save(data)
return record
# ── slot cell widget ─────────────────────────────────────────────────────────
@@ -232,7 +187,7 @@ class _SlotCell(QFrame):
) -> None:
"""Update the cell's displayed state (called by the poll/refresh).
`measured_status` is "started"/"completed" (from `_MeasuredSampleLog`)
`measured_status` is "started"/"completed" (FLOMNI-SAMPLE:<slot>:STATUS)
or None if this slot's current occupant has no measured record.
"""
self._occupied = occupied
@@ -309,7 +264,6 @@ class OMNY_SampleStorage(BECWidget, QWidget):
# last-seen (occupied, name, owner, measured_status) per slot, for
# change-detection in refresh()
self._last_state: dict[int, tuple[bool, str, str, Optional[str]]] = {}
self.measured_log = _MeasuredSampleLog(self.client)
self.label_printer = PTouchLabelPrinter(
host=PTOUCH_PRINTER_HOST, template_number=PTOUCH_TEMPLATE_NUMBER
)
@@ -345,9 +299,8 @@ class OMNY_SampleStorage(BECWidget, QWidget):
def _read_all_slots(self) -> dict[int, tuple[bool, str, str, Optional[str]]]:
"""Return {slot: (occupied, name, owner, measured_status)} for every
slot 0–20 and the gripper from a SINGLE bulk
``dev.flomni_samples.read()`` round-trip plus a single bulk
``_MeasuredSampleLog.snapshot()`` (one redis round-trip, not one
per occupied slot).
``dev.flomni_samples.read()`` round-trip (name, owner and
measurement status are separate IOC fields since 2026-10-17).
The per-slot accessors (``is_sample_slot_used`` +
``sample_names.sample{N}.get()``) would be ~44 blocking device
@@ -358,8 +311,7 @@ class OMNY_SampleStorage(BECWidget, QWidget):
``flomni_samples_sample_names_sample{N}`` /
``flomni_samples_sample_in_gripper`` /
``flomni_samples_sample_in_gripper_name``), the same keys the CLI
iterated over in ftransfer_sample_change. The raw name signal packs
name+owner (see ``sample_desc_codec``), unpacked here.
iterated over in ftransfer_sample_change.
Every underlying signal is an ``auto_monitor=True`` ``EpicsSignal``
(see ``FlomniSampleStorage``), so BEC already keeps a fresh cached
@@ -375,40 +327,31 @@ class OMNY_SampleStorage(BECWidget, QWidget):
# keep whatever is currently displayed rather than blanking out
return {}
try:
measured = self.measured_log.snapshot()
except Exception as exc:
logger.warning(f"OMNY_SampleStorage: measured-log read failed: {exc}")
measured = {}
def _val(key, default=None):
entry = data.get(key)
if isinstance(entry, dict):
return entry.get("value", default)
return default
def _measured_status(used: bool, name: str, owner: str) -> Optional[str]:
if not used:
return None
record = measured.get(pack_desc(name, owner))
return record["status"] if record else None
def _status(used: bool, raw) -> Optional[str]:
status = str(raw) if raw is not None else ""
return status if used and status in ("started", "completed") else None
for slot in (STAGE_SLOT, *STORAGE_SLOTS):
used = _val(f"flomni_samples_sample_placed_sample{slot}", 0)
raw_name = _val(f"flomni_samples_sample_names_sample{slot}", EMPTY_NAME)
name, owner = unpack_desc(raw_name if raw_name else EMPTY_NAME)
name = name if name else EMPTY_NAME
result[slot] = (bool(used), name, owner, _measured_status(bool(used), name, owner))
used = bool(_val(f"flomni_samples_sample_placed_sample{slot}", 0))
name = _val(f"flomni_samples_sample_names_sample{slot}", EMPTY_NAME) or EMPTY_NAME
owner = _val(f"flomni_samples_sample_owners_sample{slot}", "") or ""
status = _status(used, _val(f"flomni_samples_sample_status_sample{slot}"))
result[slot] = (used, str(name), str(owner), status)
g_used = _val("flomni_samples_sample_in_gripper", 0)
g_raw_name = _val("flomni_samples_sample_in_gripper_name", EMPTY_NAME)
g_name, g_owner = unpack_desc(g_raw_name if g_raw_name else EMPTY_NAME)
g_name = g_name if g_name else EMPTY_NAME
g_used = bool(_val("flomni_samples_sample_in_gripper", 0))
g_name = _val("flomni_samples_sample_in_gripper_name", EMPTY_NAME) or EMPTY_NAME
g_owner = _val("flomni_samples_sample_in_gripper_owner", "") or ""
result[GRIPPER_SLOT] = (
bool(g_used),
g_name,
g_owner,
_measured_status(bool(g_used), g_name, g_owner),
g_used,
str(g_name),
str(g_owner),
_status(g_used, _val("flomni_samples_sample_in_gripper_status")),
)
return result
@@ -425,39 +368,32 @@ class OMNY_SampleStorage(BECWidget, QWidget):
``Flomni.flomni_modify_storage_non_interactive()``.
"""
try:
if slot == GRIPPER_SLOT:
was_used = bool(self._samples.sample_in_gripper.get())
prev_name, prev_owner = unpack_desc(str(self._samples.sample_in_gripper_name.get()))
else:
was_used = self._samples.is_sample_slot_used(slot)
prev_name, prev_owner = self._samples.get_sample_name_and_owner(slot)
record = self._samples.get_sample_record(slot)
except Exception as exc:
logger.warning(
f"OMNY_SampleStorage: could not read previous state of slot {slot} for the"
f" measured-label check: {exc}"
)
return
if not was_used:
return
record = self.measured_log.pop(pack_desc(prev_name, prev_owner))
if record is None:
if not record["used"] or record["status"] not in ("started", "completed"):
return
prev_name = record["name"]
reply = QMessageBox.question(
self,
"Print label?",
f"Sample '{prev_name}' was measured ({record['status']}) on"
f" {record['measured_date']}. Print a label?",
f" {record['meas_date']}. Print a label?",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
)
if reply != QMessageBox.StandardButton.Yes:
return
fields = {
PTOUCH_FIELD_ACCOUNT: record["account"],
PTOUCH_FIELD_DATE: record["measured_date"],
PTOUCH_FIELD_ACCOUNT: record["owner"],
PTOUCH_FIELD_DATE: record["meas_date"],
PTOUCH_FIELD_SAMPLENAME: prev_name,
}
# Retry loop instead of a single attempt: the record was already
# popped from measured_log above, so if we gave up after one
# Retry loop instead of a single attempt: the slot record is
# overwritten right after this offer, so if we gave up after one
# failure the label's account/date/name would be gone for good
# even though the operator could just switch the printer on and
# try again a moment later. Mirrors the CLI's ensure_ready() retry
@@ -487,13 +423,12 @@ class OMNY_SampleStorage(BECWidget, QWidget):
"""
self._offer_print_label_for_previous_occupant(slot)
try:
packed_name = pack_desc(name, owner)
if slot == GRIPPER_SLOT:
self._samples.sample_in_gripper.set(used)
self._samples.sample_in_gripper_name.set(packed_name)
# device methods write the whole record (slot 100 = gripper); a new or
# renamed sample starts with measurement status "none"
if used:
self._samples.set_sample_slot(slot, name, owner=owner)
else:
getattr(self._samples.sample_placed, f"sample{slot}").set(used)
getattr(self._samples.sample_names, f"sample{slot}").set(packed_name)
self._samples.unset_sample_slot(slot)
# drop the cached state for this slot so the refresh() right after
# a write always repaints it, without waiting for the value to
# differ from a possibly-stale cache entry
csaxs_bec/devices/omny/flomni_sample_storage.py
+183 -78
View File
@@ -1,5 +1,27 @@
"""flOMNI sample storage on the flOMNI soft IOC (``ioc_flomni.py`` in the flOMNI repo).
One record per slot, all fields autosaved by the IOC::
FLOMNI-SAMPLE:<slot>:USED 0 = free, 1 = sample present
FLOMNI-SAMPLE:<slot>:NAME sample name (long string, up to 256 chars)
FLOMNI-SAMPLE:<slot>:OWNER e-account of the owner
FLOMNI-SAMPLE:<slot>:STATUS none / started / completed (measurement status)
FLOMNI-SAMPLE:<slot>:MEAS_DATE date of the measurement
<slot> = STAGE (slot 0, the flOMNI sample stage), T01 ... T20 (tray slots 1-20),
GRIPPER (slot 100)
Replaces (2026-10-17) the former ``XOMNY-SAMPLE_DB_flomni<n>:GET`` / ``.DESC`` records,
where name and owner were packed into one 40-character DESC (``sample_desc_codec``)
and the measurement status lived in a separate BEC global variable keyed by that
packed string. The status now travels with the sample: ``transfer_sample()`` moves
all fields between slots, the gripper and the stage.
An empty string cannot be written to the char-array NAME via Channel Access, so a
free slot carries the name ``EMPTY_NAME`` ("-").
"""
import time
from typing import Optional
from ophyd import Component as Cpt
from ophyd import Device
@@ -7,13 +29,36 @@ from ophyd import DynamicDeviceComponent as Dcpt
from ophyd import EpicsSignal
from prettytable import PrettyTable
from csaxs_bec.devices.omny.sample_desc_codec import pack_desc, unpack_desc
PV_PREFIX = "FLOMNI-SAMPLE:"
GRIPPER_SLOT = 100
STAGE_SLOT = 0
TRAY_SLOTS = range(1, 21)
EMPTY_NAME = "-"
STATUS_NONE = "none"
STATUS_STARTED = "started"
STATUS_COMPLETED = "completed"
STATUS_VALUES = (STATUS_NONE, STATUS_STARTED, STATUS_COMPLETED)
class FlomniSampleStorageError(Exception):
pass
def _slot_pv(slot_nr: int) -> str:
if slot_nr == STAGE_SLOT:
return "STAGE"
if slot_nr == GRIPPER_SLOT:
return "GRIPPER"
return f"T{slot_nr:02d}"
def _slot_signals(field: str, **kwargs) -> dict:
return {
f"sample{i}": (EpicsSignal, f"{PV_PREFIX}{_slot_pv(i)}:{field}", dict(auto_monitor=True, **kwargs))
for i in range(21)
}
class FlomniSampleStorage(Device):
USER_ACCESS = [
"is_sample_slot_used",
@@ -25,30 +70,36 @@ class FlomniSampleStorage(Device):
"get_sample_name",
"get_sample_owner",
"get_sample_name_and_owner",
"get_sample_status",
"get_sample_meas_date",
"get_sample_record",
"set_sample_measured",
"transfer_sample",
"show_all",
]
SUB_VALUE = "value"
_default_sub = SUB_VALUE
sample_placed = {
f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_flomni{i}:GET", {"auto_monitor": True}) for i in range(21)
}
sample_placed = Dcpt(sample_placed)
sample_names = {
f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_flomni{i}:GET.DESC", {"string": True, "auto_monitor": True})
for i in range(21)
}
sample_names = Dcpt(sample_names)
# slots 0-20 (0 = flOMNI stage); attribute names kept from the former device
sample_placed = Dcpt(_slot_signals("USED"))
sample_names = Dcpt(_slot_signals("NAME", string=True))
sample_owners = Dcpt(_slot_signals("OWNER", string=True))
sample_status = Dcpt(_slot_signals("STATUS", string=True))
sample_meas_date = Dcpt(_slot_signals("MEAS_DATE", string=True))
sample_in_gripper = Cpt(
EpicsSignal, name="sample_in_gripper", read_pv="XOMNY-SAMPLE_DB_flomni100:GET", auto_monitor=True
)
# gripper (slot 100)
sample_in_gripper = Cpt(EpicsSignal, read_pv=f"{PV_PREFIX}GRIPPER:USED", auto_monitor=True)
sample_in_gripper_name = Cpt(
EpicsSignal,
name="sample_in_gripper_name",
read_pv="XOMNY-SAMPLE_DB_flomni100:GET.DESC",
string=True,
auto_monitor=True
EpicsSignal, read_pv=f"{PV_PREFIX}GRIPPER:NAME", string=True, auto_monitor=True
)
sample_in_gripper_owner = Cpt(
EpicsSignal, read_pv=f"{PV_PREFIX}GRIPPER:OWNER", string=True, auto_monitor=True
)
sample_in_gripper_status = Cpt(
EpicsSignal, read_pv=f"{PV_PREFIX}GRIPPER:STATUS", string=True, auto_monitor=True
)
sample_in_gripper_meas_date = Cpt(
EpicsSignal, read_pv=f"{PV_PREFIX}GRIPPER:MEAS_DATE", string=True, auto_monitor=True
)
def __init__(self, prefix="", *, name, **kwargs):
@@ -60,72 +111,132 @@ class FlomniSampleStorage(Device):
self.wait_for_connection()
self._run_subs(sub_type=self.SUB_VALUE, timestamp=timestamp, obj=self)
def set_sample_slot(self, slot_nr: int, name: str, owner: str = "") -> bool:
if slot_nr > 20:
raise FlomniSampleStorageError(f"Invalid slot number {slot_nr}.")
# ── slot addressing ──────────────────────────────────────────────────────
@staticmethod
def _check_slot(slot_nr: int) -> int:
slot_nr = int(slot_nr)
if slot_nr == GRIPPER_SLOT or 0 <= slot_nr <= 20:
return slot_nr
raise FlomniSampleStorageError(
f"Invalid slot number {slot_nr} (0 = stage, 1-20 = tray, {GRIPPER_SLOT} = gripper)."
)
getattr(self.sample_placed, f"sample{slot_nr}").set(1).wait(timeout=5)
getattr(self.sample_names, f"sample{slot_nr}").set(pack_desc(name, owner)).wait(timeout=5)
def _sig(self, field: str, slot_nr: int) -> EpicsSignal:
slot_nr = self._check_slot(slot_nr)
if slot_nr == GRIPPER_SLOT:
return {
"used": self.sample_in_gripper,
"name": self.sample_in_gripper_name,
"owner": self.sample_in_gripper_owner,
"status": self.sample_in_gripper_status,
"meas_date": self.sample_in_gripper_meas_date,
}[field]
group = {
"used": self.sample_placed,
"name": self.sample_names,
"owner": self.sample_owners,
"status": self.sample_status,
"meas_date": self.sample_meas_date,
}[field]
return getattr(group, f"sample{slot_nr}")
def unset_sample_slot(self, slot_nr: int) -> bool:
if slot_nr > 20:
raise FlomniSampleStorageError(f"Invalid slot number {slot_nr}.")
@staticmethod
def _str(val) -> str:
return "" if val is None else str(val)
getattr(self.sample_placed, f"sample{slot_nr}").set(0).wait(timeout=5)
getattr(self.sample_names, f"sample{slot_nr}").set("-").wait(timeout=5)
# ── write ────────────────────────────────────────────────────────────────
def _write(self, slot_nr: int, used: int, name: str, owner: str, status: str, meas_date: str):
if status not in STATUS_VALUES:
raise FlomniSampleStorageError(f"Invalid status {status!r}, expected one of {STATUS_VALUES}.")
name = name if name else EMPTY_NAME
# name/owner/status/date first, USED last: a client watching USED sees a complete record
self._sig("name", slot_nr).set(name).wait(timeout=5)
self._sig("owner", slot_nr).set(owner or "").wait(timeout=5)
self._sig("status", slot_nr).set(status).wait(timeout=5)
self._sig("meas_date", slot_nr).set(meas_date or "").wait(timeout=5)
self._sig("used", slot_nr).set(int(used)).wait(timeout=5)
def set_sample_in_gripper(self, name: str, owner: str = "") -> bool:
self.sample_in_gripper.set(1).wait(timeout=5)
self.sample_in_gripper_name.set(pack_desc(name, owner)).wait(timeout=5)
def set_sample_slot(
self, slot_nr: int, name: str, owner: str = "", status: str = STATUS_NONE, meas_date: str = ""
):
"""Place a sample record in slot `slot_nr` (0-20, or 100 = gripper)."""
self._write(slot_nr, 1, name, owner, status, meas_date)
def unset_sample_in_gripper(self) -> bool:
self.sample_in_gripper.set(0).wait(timeout=5)
self.sample_in_gripper_name.set("-").wait(timeout=5)
def unset_sample_slot(self, slot_nr: int):
"""Clear slot `slot_nr` (0-20, or 100 = gripper)."""
self._write(slot_nr, 0, EMPTY_NAME, "", STATUS_NONE, "")
def set_sample_in_gripper(self, name: str, owner: str = "", status: str = STATUS_NONE, meas_date: str = ""):
self.set_sample_slot(GRIPPER_SLOT, name, owner, status, meas_date)
def unset_sample_in_gripper(self):
self.unset_sample_slot(GRIPPER_SLOT)
def set_sample_measured(self, slot_nr: int, status: str, meas_date: str, owner_if_empty: str = ""):
"""Record the measurement status/date of the sample in `slot_nr`.
If the sample has no owner yet, `owner_if_empty` (e.g. the active
e-account) is stored as owner -- the label printed when a measured
sample is cleared uses the owner as account.
"""
if status not in STATUS_VALUES:
raise FlomniSampleStorageError(f"Invalid status {status!r}, expected one of {STATUS_VALUES}.")
if owner_if_empty and not self.get_sample_owner(slot_nr):
self._sig("owner", slot_nr).set(owner_if_empty).wait(timeout=5)
self._sig("status", slot_nr).set(status).wait(timeout=5)
self._sig("meas_date", slot_nr).set(meas_date or "").wait(timeout=5)
def transfer_sample(self, from_slot: int, to_slot: int):
"""Move the complete record (name, owner, status, date) from one slot to
another (0-20, 100 = gripper) and clear the source slot."""
rec = self.get_sample_record(from_slot)
self.set_sample_slot(to_slot, rec["name"], rec["owner"], rec["status"], rec["meas_date"])
self.unset_sample_slot(from_slot)
# ── read ─────────────────────────────────────────────────────────────────
def is_sample_slot_used(self, slot_nr: int) -> bool:
val = getattr(self.sample_placed, f"sample{slot_nr}").get(use_monitor=False)
return bool(val)
return bool(self._sig("used", slot_nr).get(use_monitor=False))
def is_sample_in_gripper(self) -> bool:
val = self.sample_in_gripper.get(use_monitor=False)
return bool(val)
return self.is_sample_slot_used(GRIPPER_SLOT)
def get_sample_name(self, slot_nr) -> str:
val = getattr(self.sample_names, f"sample{slot_nr}").get()
return unpack_desc(str(val))[0]
name = self._str(self._sig("name", slot_nr).get())
return name if name else EMPTY_NAME
def get_sample_owner(self, slot_nr) -> str:
val = getattr(self.sample_names, f"sample{slot_nr}").get()
return unpack_desc(str(val))[1]
return self._str(self._sig("owner", slot_nr).get())
def get_sample_name_and_owner(self, slot_nr) -> tuple:
val = getattr(self.sample_names, f"sample{slot_nr}").get()
return unpack_desc(str(val))
return (self.get_sample_name(slot_nr), self.get_sample_owner(slot_nr))
def show_all(self, measured: Optional[dict] = None):
"""Print the tray grid plus loaded tray/gripper/stage samples.
def get_sample_status(self, slot_nr) -> str:
status = self._str(self._sig("status", slot_nr).get())
return status if status in STATUS_VALUES else STATUS_NONE
`measured`, if given, is a plain dict (as returned by
`_MeasuredSampleLog.snapshot()`) mapping packed
`pack_desc(name, owner)` strings to {"status": ..., ...} records --
this device class stays decoupled from the BEC-client-backed
measured-sample log that produces it; callers (e.g.
`Flomni.ftransfer_show_all()`) inject the snapshot instead.
def get_sample_meas_date(self, slot_nr) -> str:
return self._str(self._sig("meas_date", slot_nr).get())
Must be a plain, msgpack-serializable dict, not a callable: this
method runs via BEC's device-server RPC mechanism, which
serializes every argument before sending it, so a bound method
(an earlier version of this passed `self.measured_log.get`) fails
with "can not serialize 'method' object".
"""
def get_sample_record(self, slot_nr) -> dict:
return {
"used": self.is_sample_slot_used(slot_nr),
"name": self.get_sample_name(slot_nr),
"owner": self.get_sample_owner(slot_nr),
"status": self.get_sample_status(slot_nr),
"meas_date": self.get_sample_meas_date(slot_nr),
}
def _measured_suffix(name: str, owner: str) -> str:
if not measured:
return ""
record = measured.get(pack_desc(name, owner))
if record is None:
return ""
return f" [measured: {record['status']}]"
def show_all(self):
"""Print the tray grid plus loaded tray/gripper/stage samples."""
def _suffix(slot_nr: int) -> str:
owner = self.get_sample_owner(slot_nr)
status = self.get_sample_status(slot_nr)
text = f" (owner: {owner})" if owner else ""
if status != STATUS_NONE:
date = self.get_sample_meas_date(slot_nr)
text += f" [measured: {status}{', ' + date if date else ''}]"
return text
t = PrettyTable()
t.title = "flOMNI sample storage"
@@ -141,21 +252,15 @@ class FlomniSampleStorage(Device):
t.add_row(row)
print(t)
print("\n\nFollowing samples are currently loaded:\n")
for ct in range(1, 21):
for ct in TRAY_SLOTS:
if self.is_sample_slot_used(ct):
name, owner = self.get_sample_name_and_owner(ct)
owner_suffix = f" (owner: {owner})" if owner else ""
print(f" Position {ct:2.0f}: {name}{owner_suffix}{_measured_suffix(name, owner)}")
if self.sample_in_gripper.get():
name, owner = unpack_desc(str(self.sample_in_gripper_name.get()))
owner_suffix = f" (owner: {owner})" if owner else ""
print(f"\n Gripper: {name}{owner_suffix}{_measured_suffix(name, owner)}\n")
print(f" Position {ct:2.0f}: {self.get_sample_name(ct)}{_suffix(ct)}")
if self.is_sample_in_gripper():
print(f"\n Gripper: {self.get_sample_name(GRIPPER_SLOT)}{_suffix(GRIPPER_SLOT)}\n")
else:
print(f"\n Gripper: no sample\n")
if self.is_sample_slot_used(0):
name, owner = self.get_sample_name_and_owner(0)
owner_suffix = f" (owner: {owner})" if owner else ""
print(f" flOMNI stage: {name}{owner_suffix}{_measured_suffix(name, owner)}\n")
if self.is_sample_slot_used(STAGE_SLOT):
print(f" flOMNI stage: {self.get_sample_name(STAGE_SLOT)}{_suffix(STAGE_SLOT)}\n")
else:
print(f" flOMNI stage: no sample\n")
csaxs_bec/devices/sim/sim_sample_storage.py
+1 -1
View File
@@ -1,7 +1,7 @@
"""
Simulated flOMNI sample storage.
`FlomniSampleStorage` is purely EPICS-backed (XOMNY-SAMPLE_DB_flomni* PVs). The simulated
`FlomniSampleStorage` is purely EPICS-backed (FLOMNI-SAMPLE:<slot>:<field> PVs). The simulated
version follows the patched-device pattern of `simulated_beamline_devices`: the real class
is instantiated with all PVs replaced by in-memory mock PVs, so `set_sample_slot`,
`show_all`, the gripper bookkeeping and the sample-transfer scripts run unchanged; the
omny_e2e_tests/sim_flomni_harness.py
+28 -17
View File
@@ -417,36 +417,47 @@ def main(): # noqa: C901
"slot 1 -> gripper bookkeeping",
)
# --- 9b. sample owner (packed into the same DESC field as name) ------------------
# --- 9b. sample record: name, owner, measurement status (separate IOC fields) ----
samples.set_sample_slot(2, "owner_test_sample", owner="mholler")
check(
samples.get_sample_name(2) == "owner_test_sample", "get_sample_name returns plain name only"
samples.get_sample_name(2) == "owner_test_sample", "get_sample_name returns the name"
)
check(samples.get_sample_owner(2) == "mholler", "get_sample_owner returns the packed owner")
check(samples.get_sample_owner(2) == "mholler", "get_sample_owner returns the owner")
check(
samples.get_sample_name_and_owner(2) == ("owner_test_sample", "mholler"),
"get_sample_name_and_owner returns (name, owner)",
)
check(samples.get_sample_status(2) == "none", "a new sample starts with status 'none'")
samples.unset_sample_slot(2)
check(
str(getattr(samples.sample_names, "sample2").get()) == "-",
"unset_sample_slot writes bare '-', not '- | <owner>'",
str(getattr(samples.sample_names, "sample2").get()) == "-"
and samples.get_sample_owner(2) == "",
"unset_sample_slot clears name ('-') and owner",
)
# gripper move (slot -> gripper) must preserve owner, not just name. The
# gripper is a separate Cpt (sample_in_gripper_name), not part of the
# sample_names slot dict, so unpack it directly rather than via
# get_sample_name_and_owner (which only covers slots 0-20).
from csaxs_bec.devices.omny.sample_desc_codec import unpack_desc
# transfer slot -> gripper -> slot must carry the whole record (name, owner,
# measurement status and date), slot 100 = gripper
samples.set_sample_slot(4, "transfer_test_sample", owner="e12345")
name, owner = samples.get_sample_name_and_owner(4)
samples.set_sample_in_gripper(name, owner=owner)
samples.unset_sample_slot(4)
samples.set_sample_measured(4, "completed", "2026-10")
samples.transfer_sample(4, 100)
check(
unpack_desc(str(samples.sample_in_gripper_name.get())) == ("transfer_test_sample", "e12345"),
"owner survives a slot -> gripper move",
samples.is_sample_in_gripper()
and not samples.is_sample_slot_used(4)
and samples.get_sample_name_and_owner(100) == ("transfer_test_sample", "e12345")
and samples.get_sample_status(100) == "completed"
and samples.get_sample_meas_date(100) == "2026-10",
"slot -> gripper move carries name, owner, status and date",
)
samples.unset_sample_in_gripper()
samples.transfer_sample(100, 5)
check(
not samples.is_sample_in_gripper() and samples.get_sample_status(5) == "completed",
"gripper -> slot move carries the status",
)
samples.unset_sample_slot(5)
# set_sample_measured fills an empty owner (label printing uses the owner as account)
samples.set_sample_slot(6, "no_owner_sample")
samples.set_sample_measured(6, "started", "2026-10", owner_if_empty="e99999")
check(samples.get_sample_owner(6) == "e99999", "owner_if_empty fills an empty owner")
samples.unset_sample_slot(6)
frame_a = ids_cam.cam.get_image_data()
frame_b = ids_cam.cam.get_image_data()
tests/tests_bec_ipython_client/test_flomni_tomo_angles.py
+5 -2
View File
@@ -243,8 +243,6 @@ def make_flomni_for_golden_resume(monkeypatch, tomo_type: int) -> Flomni:
obj.golden_max_number_of_projections = 21
obj.tomo_id = -1
obj.sample_get_name = lambda position: "test" # backs the read-only sample_name property
obj.sample_get_measured_log_key = lambda position=0: "test|"
obj.measured_log = types.SimpleNamespace(mark_measured=lambda *a, **k: None)
obj.special_angles = []
obj._golden = _fake_golden
obj._golden_equally_spaced = _fake_golden
@@ -270,6 +268,11 @@ def make_flomni_for_golden_resume(monkeypatch, tomo_type: int) -> Flomni:
)
monkeypatch.setitem(builtins.__dict__, "bec", fake_bec)
monkeypatch.setitem(builtins.__dict__, "scans", _FakeScans())
# tomo_scan() records the measurement status on the sample record (stage slot 0)
fake_dev = types.SimpleNamespace(
flomni_samples=types.SimpleNamespace(set_sample_measured=lambda *a, **k: None)
)
monkeypatch.setitem(builtins.__dict__, "dev", fake_dev)
return obj