diff --git a/csaxs_bec/bec_ipython_client/plugins/omny/omny_sample_transfer_mixin.py b/csaxs_bec/bec_ipython_client/plugins/omny/omny_sample_transfer_mixin.py index 2c00948f..05ac44fc 100644 --- a/csaxs_bec/bec_ipython_client/plugins/omny/omny_sample_transfer_mixin.py +++ b/csaxs_bec/bec_ipython_client/plugins/omny/omny_sample_transfer_mixin.py @@ -1077,20 +1077,20 @@ class OMNYSampleTransferMixin: def otransfer_is_sample_in_gripper(self) -> bool: return dev.omny_samples.is_sample_in_gripper() - def otransfer_set_sample_slot(self, container: str, slot_nr: int, name: str) -> bool: - return dev.omny_samples.set_sample_slot(container, slot_nr, name) + def otransfer_set_sample_slot(self, container: str, slot_nr: int, name: str, owner: str = "") -> bool: + return dev.omny_samples.set_sample_slot(container, slot_nr, name, owner) def otransfer_unset_sample_slot(self, container: str, slot_nr: int) -> bool: return dev.omny_samples.unset_sample_slot(container, slot_nr) - def otransfer_set_sample_in_gripper(self, name: str) -> bool: - return dev.omny_samples.set_sample_in_gripper(name) + def otransfer_set_sample_in_gripper(self, name: str, owner: str = "") -> bool: + return dev.omny_samples.set_sample_in_gripper(name, owner) def otransfer_unset_sample_in_gripper(self) -> bool: return dev.omny_samples.unset_sample_in_gripper() - def otransfer_set_sample_in_samplestage(self, name: str) -> bool: - return dev.omny_samples.set_sample_in_samplestage(name) + def otransfer_set_sample_in_samplestage(self, name: str, owner: str = "") -> bool: + return dev.omny_samples.set_sample_in_samplestage(name, owner) def otransfer_unset_sample_in_samplestage(self) -> bool: return dev.omny_samples.unset_sample_in_samplestage() @@ -1122,6 +1122,36 @@ class OMNYSampleTransferMixin: def otransfer_search_shuttle_in_slot(self, shuttle: str) -> int: return dev.omny_samples.search_shuttle_in_slot(shuttle) + def otransfer_get_sample_owner(self, container: str, slot_nr: int) -> str: + return dev.omny_samples.get_sample_owner(container, slot_nr) + + def otransfer_get_sample_name_and_owner(self, container: str, slot_nr: int) -> tuple: + return dev.omny_samples.get_sample_name_and_owner(container, slot_nr) + + def otransfer_set_sample_status(self, container: str, slot_nr: int, status: str) -> bool: + return dev.omny_samples.set_sample_status(container, slot_nr, status) + + def otransfer_get_sample_status(self, container: str, slot_nr: int) -> str: + return dev.omny_samples.get_sample_status(container, slot_nr) + + def otransfer_get_sample_owner_in_gripper(self) -> str: + return dev.omny_samples.get_sample_owner_in_gripper() + + def otransfer_get_sample_owner_in_samplestage(self) -> str: + return dev.omny_samples.get_sample_owner_in_samplestage() + + def otransfer_set_sample_status_in_gripper(self, status: str) -> bool: + return dev.omny_samples.set_sample_status_in_gripper(status) + + def otransfer_get_sample_status_in_gripper(self) -> str: + return dev.omny_samples.get_sample_status_in_gripper() + + def otransfer_set_sample_status_in_samplestage(self, status: str) -> bool: + return dev.omny_samples.set_sample_status_in_samplestage(status) + + def otransfer_get_sample_status_in_samplestage(self) -> str: + return dev.omny_samples.get_sample_status_in_samplestage() + # def _otransfer_modify_sample(_ocontainer, _oposition, _ostatus, _oname): # #_ocontainer is A, B, C for the pin containers # #it can also be FZP, OSA and omny diff --git a/csaxs_bec/bec_widgets/widgets/sample_storage/sample_storage.py b/csaxs_bec/bec_widgets/widgets/sample_storage/sample_storage.py index 8b4bc0b1..e6120e13 100644 --- a/csaxs_bec/bec_widgets/widgets/sample_storage/sample_storage.py +++ b/csaxs_bec/bec_widgets/widgets/sample_storage/sample_storage.py @@ -9,13 +9,13 @@ different setups, detected at construction time (see ``SETUP_PROFILES``, ``_detect_setup()``) — a BEC session is always exactly one setup, never both at once: -Both setups require every sample slot to carry a name *and* an owner, -packed into one EPICS DESC field (see ``sample_desc_codec``) since neither -backend has a separate owner PV. +Both setups require every sample slot to carry a name *and* an owner, but +via different backend mechanisms: FlOMNI (``dev.flomni_samples``) A flat magazine: slots 1-20, the sample stage (slot 0) and the gripper - (slot 100). + (slot 100). Name and owner are packed into one EPICS DESC field (see + ``sample_desc_codec``) since this backend has no separate owner PV. OMNY (``dev.omny_samples``) A richer structure: three shuttles (A/B/C) of 6 slots each, a sparse @@ -23,9 +23,13 @@ OMNY (``dev.omny_samples``) separate sample stage and gripper, and a 6-slot "parking" carousel recording which shuttle letter is docked at each physical parking position (parking slots are a shuttle-letter assignment, not a sample, - so they carry no name/owner). Whichever shuttle's parking slot matches - the ``oparkz`` motor's current readback is "active", i.e. at the - transfer-ready position (see ``oparkz_active_parking_slot()``). + so they carry no name/owner/status). Unlike FlOMNI, OMNY has a + dedicated owner channel (and a measured-status channel -- see + ``OMNY_SampleStorage.STATUS_NONE``/``STATUS_STARTED``/ + ``STATUS_COMPLETED``) next to each sample slot's existing placed/name + channels, so no packing is needed there. Whichever shuttle's parking + slot matches the ``oparkz`` motor's current readback is "active", i.e. + at the transfer-ready position (see ``oparkz_active_parking_slot()``). A LamNI session has neither device; the widget shows a "no known setup detected" message in that case. @@ -127,6 +131,14 @@ OMNY_PARKING_SLOTS = tuple(range(1, 7)) # slots (which use EMPTY_NAME = "-"), parking slots are cleared to "none". PARKING_EMPTY_SENTINEL = "none" +# Mirrors OMNYSampleStorage.STATUS_NONE/STATUS_STARTED/STATUS_COMPLETED's +# int<->str mapping, duplicated locally rather than reaching into that +# class's own (private) _status_str_to_val()/_status_val_to_str() +# classmethods -- this widget already duplicates other device-side +# conventions the same way (see EMPTY_NAME/PARKING_EMPTY_SENTINEL above). +OMNY_STATUS_OPTIONS = ("none", "started", "completed") +OMNY_STATUS_STR_TO_VAL = {"none": 0, "started": 1, "completed": 2} + # colours # An occupied slot's name label intentionally has no fixed foreground colour # here (see _SlotCell.set_state()) -- it was previously hardcoded to @@ -388,10 +400,17 @@ class _SlotCell(QFrame): menu = QMenu(self) if self._occupied: act_rename = menu.addAction("Change name…") + # OMNY-only (tuple key): measured status has a real EPICS + # channel there (see OMNY_SampleStorage), unlike FlOMNI's + # scan-lifecycle-driven redis log, which isn't manually + # editable from here. + act_status = menu.addAction("Set measured status…") if isinstance(self._slot, tuple) else None act_clear = menu.addAction("Clear (set empty)") chosen = menu.exec_(self.mapToGlobal(pos)) if chosen == act_rename: self._owner.action_change_name(self._slot, self._name, self._sample_owner) + elif act_status is not None and chosen == act_status: + self._owner.action_set_status(self._slot) elif chosen == act_clear: self._owner.action_clear(self._slot) else: @@ -620,10 +639,11 @@ class OMNY_SampleStorage(BECWidget, QWidget): """Parse a dev.omny_samples.read(cached=True) dict into {key: (occupied, name, owner, measured_status)} for every sample-kind slot -- stage, gripper, the 10 fixed O positions, and - the 18 shuttle slots. Like FlOMNI, OMNY has no separate owner PV, so - the name+owner pair is packed into the one raw name field (see - ``sample_desc_codec``); measured_status is always None -- OMNY has - no measured-sample-log equivalent (see module docstring). + the 18 shuttle slots. Owner and measured status each have their own + dedicated OMNYSampleStorage signal (added alongside the existing + placed/name signals -- see that class's ``STATUS_NONE``/ + ``STATUS_STARTED``/``STATUS_COMPLETED``), unlike FlOMNI which packs + name+owner into one field via ``sample_desc_codec``. """ def _val(key, default=None): @@ -632,32 +652,43 @@ class OMNY_SampleStorage(BECWidget, QWidget): return entry.get("value", default) return default - def _name_owner(raw): - name, owner = unpack_desc(raw if raw else EMPTY_NAME) - return (name if name else EMPTY_NAME), owner + def _status_str(raw) -> Optional[str]: + # mirrors OMNYSampleStorage._status_val_to_str(), without a live + # device round-trip: this widget only has the bulk-read dict. + try: + val = int(round(float(raw))) + except (TypeError, ValueError): + return None + return {1: "started", 2: "completed"}.get(val) # 0 ("none") -> None: no badge state: dict[tuple, tuple[bool, str, str, Optional[str]]] = {} used = bool(_val("omny_samples_sample_in_samplestage", 0)) - name, owner = _name_owner(_val("omny_samples_sample_in_samplestage_name", EMPTY_NAME)) - state[("stage",)] = (used, name, owner, None) + name = _val("omny_samples_sample_in_samplestage_name", EMPTY_NAME) or EMPTY_NAME + owner = _val("omny_samples_sample_in_samplestage_owner", "") or "" + status = _status_str(_val("omny_samples_sample_in_samplestage_status", 0)) + state[("stage",)] = (used, name, owner, status) used = bool(_val("omny_samples_sample_in_gripper", 0)) - name, owner = _name_owner(_val("omny_samples_sample_in_gripper_name", EMPTY_NAME)) - state[("gripper",)] = (used, name, owner, None) + name = _val("omny_samples_sample_in_gripper_name", EMPTY_NAME) or EMPTY_NAME + owner = _val("omny_samples_sample_in_gripper_owner", "") or "" + status = _status_str(_val("omny_samples_sample_in_gripper_status", 0)) + state[("gripper",)] = (used, name, owner, status) for n in OMNY_FIXED_POSITIONS: used = bool(_val(f"omny_samples_sample_placed_sample{n}", 0)) - name, owner = _name_owner(_val(f"omny_samples_sample_names_sample{n}", EMPTY_NAME)) - state[("O", n)] = (used, name, owner, None) + name = _val(f"omny_samples_sample_names_sample{n}", EMPTY_NAME) or EMPTY_NAME + owner = _val(f"omny_samples_sample_owners_sample{n}", "") or "" + status = _status_str(_val(f"omny_samples_sample_status_sample{n}", 0)) + state[("O", n)] = (used, name, owner, status) for letter in OMNY_SHUTTLE_LETTERS: for n in OMNY_SHUTTLE_SLOTS: used = bool(_val(f"omny_samples_sample_shuttle_{letter}_placed_sample{n}", 0)) - name, owner = _name_owner( - _val(f"omny_samples_sample_shuttle_{letter}_names_sample{n}", EMPTY_NAME) - ) - state[("shuttle", letter, n)] = (used, name, owner, None) + name = _val(f"omny_samples_sample_shuttle_{letter}_names_sample{n}", EMPTY_NAME) or EMPTY_NAME + owner = _val(f"omny_samples_sample_shuttle_{letter}_owners_sample{n}", "") or "" + status = _status_str(_val(f"omny_samples_sample_shuttle_{letter}_status_sample{n}", 0)) + state[("shuttle", letter, n)] = (used, name, owner, status) return state @@ -796,29 +827,41 @@ class OMNY_SampleStorage(BECWidget, QWidget): def _write_slot_omny(self, key: tuple, used: int, name: str, owner: str = "") -> bool: """Write (used, name, owner) to an OMNY sample-kind slot -- stage, gripper, a fixed O position, or a shuttle slot -- mirroring the - direct-signal style of ``_write_slot()``, including packing - name+owner into the one raw name field the same way (see - ``sample_desc_codec``). No measured-sample-log integration exists - for OMNY (see module docstring), so unlike ``_write_slot`` this - never offers to print a P-touch label for the previous occupant. + direct-signal style of ``_write_slot()``. Unlike FlOMNI, OMNY has a + dedicated owner signal (added alongside placed/name -- see + ``OMNYSampleStorage``), so name and owner are written separately, + no packing needed. Measured status is reset to "none" on every + write here, mirroring ``OMNYSampleStorage.set_sample_slot()``'s/ + ``unset_sample_slot()``'s own behavior -- a fresh name (even a + rename) or a cleared slot isn't mid-measurement. No P-touch + label-printing offer for the previous occupant (FlOMNI-only, see + module docstring). """ try: - packed_name = pack_desc(name, owner) + status_none = self._samples.STATUS_NONE kind = key[0] if kind == "stage": self._samples.sample_in_samplestage.set(used) - self._samples.sample_in_samplestage_name.set(packed_name) + self._samples.sample_in_samplestage_name.set(name) + self._samples.sample_in_samplestage_owner.set(owner) + self._samples.sample_in_samplestage_status.set(status_none) elif kind == "gripper": self._samples.sample_in_gripper.set(used) - self._samples.sample_in_gripper_name.set(packed_name) + self._samples.sample_in_gripper_name.set(name) + self._samples.sample_in_gripper_owner.set(owner) + self._samples.sample_in_gripper_status.set(status_none) elif kind == "O": _, n = key getattr(self._samples.sample_placed, f"sample{n}").set(used) - getattr(self._samples.sample_names, f"sample{n}").set(packed_name) + getattr(self._samples.sample_names, f"sample{n}").set(name) + getattr(self._samples.sample_owners, f"sample{n}").set(owner) + getattr(self._samples.sample_status, f"sample{n}").set(status_none) elif kind == "shuttle": _, letter, n = key getattr(getattr(self._samples, f"sample_shuttle_{letter}_placed"), f"sample{n}").set(used) - getattr(getattr(self._samples, f"sample_shuttle_{letter}_names"), f"sample{n}").set(packed_name) + getattr(getattr(self._samples, f"sample_shuttle_{letter}_names"), f"sample{n}").set(name) + getattr(getattr(self._samples, f"sample_shuttle_{letter}_owners"), f"sample{n}").set(owner) + getattr(getattr(self._samples, f"sample_shuttle_{letter}_status"), f"sample{n}").set(status_none) else: raise ValueError(f"unknown OMNY slot key {key!r}") self._last_state.pop(key, None) @@ -832,6 +875,39 @@ class OMNY_SampleStorage(BECWidget, QWidget): ) return False + def _write_status_omny(self, key: tuple, status: str) -> bool: + """Write just the measured-status signal for an OMNY sample-kind + slot, leaving used/name/owner untouched -- the counterpart to + ``_write_slot_omny()`` resetting status on every name/owner write, + for the manual "Set measured status…" action (see + ``action_set_status``). + """ + try: + val = OMNY_STATUS_STR_TO_VAL[status] + kind = key[0] + if kind == "stage": + self._samples.sample_in_samplestage_status.set(val) + elif kind == "gripper": + self._samples.sample_in_gripper_status.set(val) + elif kind == "O": + _, n = key + getattr(self._samples.sample_status, f"sample{n}").set(val) + elif kind == "shuttle": + _, letter, n = key + getattr(getattr(self._samples, f"sample_shuttle_{letter}_status"), f"sample{n}").set(val) + else: + raise ValueError(f"unknown OMNY slot key {key!r}") + self._last_state.pop(key, None) + return True + except Exception as exc: + logger.warning(f"OMNY_SampleStorage: write of status for slot {key} failed: {exc}") + QMessageBox.critical( + self, + "Write failed", + f"Could not write status for slot {self._slot_title(key)} to the device:\n{exc}", + ) + return False + def _write_parking_omny(self, slot_nr: int, letter: Optional[str]) -> bool: """Assign (or clear) which shuttle letter is recorded as parked at OMNY parking slot `slot_nr` (1-6), mirroring @@ -1186,6 +1262,32 @@ class OMNY_SampleStorage(BECWidget, QWidget): if self._write_sample(slot, 0, EMPTY_NAME): self.refresh() + def action_set_status(self, slot: tuple) -> None: + """Manually set an OMNY sample-kind slot's measured status. OMNY + only (tuple key) -- see ``_SlotCell._show_menu``. FlOMNI's status + stays a separate, scan-lifecycle-driven redis log (see module + docstring), not exposed for manual editing here. + """ + if not isinstance(slot, tuple): + return + current = self._last_state.get(slot, (False, "", "", None))[3] or "none" + try: + current_index = OMNY_STATUS_OPTIONS.index(current) + except ValueError: + current_index = 0 + status, ok = QInputDialog.getItem( + self, + "Set measured status", + f"Status for slot {self._slot_title(slot)}:", + OMNY_STATUS_OPTIONS, + current_index, + False, + ) + if not ok: + return + if self._write_status_omny(slot, status): + self.refresh() + def action_set_parking(self, letter: str, current_slot: Optional[int]) -> None: """Assign OMNY shuttle `letter` to a parking slot (1-6), via ``_ShuttleStatusLabel``'s context menu. Overwrites that slot's diff --git a/csaxs_bec/devices/omny/omny_sample_storage.py b/csaxs_bec/devices/omny/omny_sample_storage.py index 4675176f..000cb19a 100644 --- a/csaxs_bec/devices/omny/omny_sample_storage.py +++ b/csaxs_bec/devices/omny/omny_sample_storage.py @@ -23,7 +23,17 @@ class OMNYSampleStorage(Device): "unset_sample_in_samplestage", "get_sample_name_in_samplestage", "get_sample_name_in_gripper", + "get_sample_owner_in_samplestage", + "get_sample_owner_in_gripper", + "set_sample_status_in_samplestage", + "get_sample_status_in_samplestage", + "set_sample_status_in_gripper", + "get_sample_status_in_gripper", "get_sample_name", + "get_sample_owner", + "get_sample_name_and_owner", + "set_sample_status", + "get_sample_status", "is_sample_in_samplestage", "set_shuttle_slot", "unset_shuttle_slot", @@ -36,6 +46,18 @@ class OMNYSampleStorage(Device): SUB_VALUE = "value" _default_sub = SUB_VALUE + # Measured status, one of these three -- an "ai" record's VAL (see + # omny_samples_status.template in the OMNY/OMNY.git IOC repo), not a + # proper EPICS enum: this IOC's sample DB uses plain "ai" records + # uniformly even for boolean-ish fields (see the placed-flag records), + # so the new STATUS record follows the same convention rather than + # introducing an mbbo/mbbi record type used nowhere else here. The + # int<->str mapping lives entirely in this class. + STATUS_NONE = 0 + STATUS_STARTED = 1 + STATUS_COMPLETED = 2 + _STATUS_LABELS = {STATUS_NONE: "none", STATUS_STARTED: "started", STATUS_COMPLETED: "completed"} + sample_shuttle_A_placed = { f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_shuttle_A:{i}", {"auto_monitor": True}) for i in range(1, 7) } @@ -97,6 +119,75 @@ class OMNYSampleStorage(Device): } sample_names = Dcpt(sample_names) + # Measured status (VAL) + owner (DESC) -- one new record per sample + # position, added alongside the existing placed/name records (see + # STATUS_NONE/STARTED/COMPLETED above and omny_samples_status.template + # in OMNY/OMNY.git). Parking slots have no equivalent -- they hold a + # shuttle letter, not a sample. + sample_shuttle_A_status = { + f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_shuttle_A:{i}:STATUS", {"auto_monitor": True}) + for i in range(1, 7) + } + sample_shuttle_A_status = Dcpt(sample_shuttle_A_status) + + sample_shuttle_B_status = { + f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_shuttle_B:{i}:STATUS", {"auto_monitor": True}) + for i in range(1, 7) + } + sample_shuttle_B_status = Dcpt(sample_shuttle_B_status) + + sample_shuttle_C_status = { + f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_shuttle_C:{i}:STATUS", {"auto_monitor": True}) + for i in range(1, 7) + } + sample_shuttle_C_status = Dcpt(sample_shuttle_C_status) + + sample_status = { + f"sample{i}": (EpicsSignal, f"XOMNY-SAMPLE_DB_omny:{i}:STATUS", {"auto_monitor": True}) + for i in [10, 11, 12, 13, 14, 32, 33, 34, 100, 101] + } + sample_status = Dcpt(sample_status) + + sample_shuttle_A_owners = { + f"sample{i}": ( + EpicsSignal, + f"XOMNY-SAMPLE_DB_shuttle_A:{i}:STATUS.DESC", + {"string": True, "auto_monitor": True}, + ) + for i in range(1, 7) + } + sample_shuttle_A_owners = Dcpt(sample_shuttle_A_owners) + + sample_shuttle_B_owners = { + f"sample{i}": ( + EpicsSignal, + f"XOMNY-SAMPLE_DB_shuttle_B:{i}:STATUS.DESC", + {"string": True, "auto_monitor": True}, + ) + for i in range(1, 7) + } + sample_shuttle_B_owners = Dcpt(sample_shuttle_B_owners) + + sample_shuttle_C_owners = { + f"sample{i}": ( + EpicsSignal, + f"XOMNY-SAMPLE_DB_shuttle_C:{i}:STATUS.DESC", + {"string": True, "auto_monitor": True}, + ) + for i in range(1, 7) + } + sample_shuttle_C_owners = Dcpt(sample_shuttle_C_owners) + + sample_owners = { + f"sample{i}": ( + EpicsSignal, + f"XOMNY-SAMPLE_DB_omny:{i}:STATUS.DESC", + {"string": True, "auto_monitor": True}, + ) + for i in [10, 11, 12, 13, 14, 32, 33, 34, 100, 101] + } + sample_owners = Dcpt(sample_owners) + sample_in_gripper = Cpt( EpicsSignal, name="sample_in_gripper", read_pv="XOMNY-SAMPLE_DB_omny:110.VAL", auto_monitor=True ) @@ -119,6 +210,28 @@ class OMNYSampleStorage(Device): auto_monitor=True ) + sample_in_gripper_status = Cpt( + EpicsSignal, name="sample_in_gripper_status", read_pv="XOMNY-SAMPLE_DB_omny:110:STATUS", auto_monitor=True + ) + sample_in_gripper_owner = Cpt( + EpicsSignal, + name="sample_in_gripper_owner", + read_pv="XOMNY-SAMPLE_DB_omny:110:STATUS.DESC", + string=True, + auto_monitor=True, + ) + + sample_in_samplestage_status = Cpt( + EpicsSignal, name="sample_in_samplestage_status", read_pv="XOMNY-SAMPLE_DB_omny:0:STATUS", auto_monitor=True + ) + sample_in_samplestage_owner = Cpt( + EpicsSignal, + name="sample_in_samplestage_owner", + read_pv="XOMNY-SAMPLE_DB_omny:0:STATUS.DESC", + string=True, + auto_monitor=True, + ) + def __init__(self, prefix="", *, name, **kwargs): super().__init__(prefix, name=name, **kwargs) self.sample_shuttle_A_placed.sample1.subscribe(self._emit_value) @@ -128,41 +241,114 @@ class OMNYSampleStorage(Device): self.wait_for_connection() self._run_subs(sub_type=self.SUB_VALUE, timestamp=timestamp, obj=self) - def set_sample_slot(self, container: str, slot_nr: int, name: str) -> bool: + @classmethod + def _status_str_to_val(cls, status: str) -> int: + status = (status or "none").strip().lower() + for val, label in cls._STATUS_LABELS.items(): + if label == status: + return val + raise OMNYSampleStorageError( + f"Invalid status {status!r}; expected one of {sorted(cls._STATUS_LABELS.values())}." + ) + + @classmethod + def _status_val_to_str(cls, val) -> str: + try: + return cls._STATUS_LABELS.get(int(round(float(val))), "none") + except (TypeError, ValueError): + return "none" + + def set_sample_slot(self, container: str, slot_nr: int, name: str, owner: str = "") -> bool: if slot_nr > 20: raise OMNYSampleStorageError(f"Invalid slot number {slot_nr}.") + status = self._status_str_to_val("none") if container == "A": getattr(self.sample_shuttle_A_placed, f"sample{slot_nr}").set(1) getattr(self.sample_shuttle_A_names, f"sample{slot_nr}").set(name) + getattr(self.sample_shuttle_A_owners, f"sample{slot_nr}").set(owner) + getattr(self.sample_shuttle_A_status, f"sample{slot_nr}").set(status) elif container == "B": getattr(self.sample_shuttle_B_placed, f"sample{slot_nr}").set(1) getattr(self.sample_shuttle_B_names, f"sample{slot_nr}").set(name) + getattr(self.sample_shuttle_B_owners, f"sample{slot_nr}").set(owner) + getattr(self.sample_shuttle_B_status, f"sample{slot_nr}").set(status) elif container == "C": getattr(self.sample_shuttle_C_placed, f"sample{slot_nr}").set(1) getattr(self.sample_shuttle_C_names, f"sample{slot_nr}").set(name) + getattr(self.sample_shuttle_C_owners, f"sample{slot_nr}").set(owner) + getattr(self.sample_shuttle_C_status, f"sample{slot_nr}").set(status) elif container == "O": getattr(self.sample_placed, f"sample{slot_nr}").set(1) getattr(self.sample_names, f"sample{slot_nr}").set(name) - + getattr(self.sample_owners, f"sample{slot_nr}").set(owner) + getattr(self.sample_status, f"sample{slot_nr}").set(status) def unset_sample_slot(self, container: str, slot_nr: int) -> bool: if slot_nr > 20: raise OMNYSampleStorageError(f"Invalid slot number {slot_nr}.") + status = self._status_str_to_val("none") if container == "A": getattr(self.sample_shuttle_A_placed, f"sample{slot_nr}").set(0) getattr(self.sample_shuttle_A_names, f"sample{slot_nr}").set("-") + getattr(self.sample_shuttle_A_owners, f"sample{slot_nr}").set("") + getattr(self.sample_shuttle_A_status, f"sample{slot_nr}").set(status) elif container == "B": getattr(self.sample_shuttle_B_placed, f"sample{slot_nr}").set(0) getattr(self.sample_shuttle_B_names, f"sample{slot_nr}").set("-") + getattr(self.sample_shuttle_B_owners, f"sample{slot_nr}").set("") + getattr(self.sample_shuttle_B_status, f"sample{slot_nr}").set(status) elif container == "C": getattr(self.sample_shuttle_C_placed, f"sample{slot_nr}").set(0) getattr(self.sample_shuttle_C_names, f"sample{slot_nr}").set("-") + getattr(self.sample_shuttle_C_owners, f"sample{slot_nr}").set("") + getattr(self.sample_shuttle_C_status, f"sample{slot_nr}").set(status) elif container == "O": getattr(self.sample_placed, f"sample{slot_nr}").set(0) getattr(self.sample_names, f"sample{slot_nr}").set("-") + getattr(self.sample_owners, f"sample{slot_nr}").set("") + getattr(self.sample_status, f"sample{slot_nr}").set(status) + def set_sample_status(self, container: str, slot_nr: int, status: str) -> bool: + val = self._status_str_to_val(status) + if container == "A": + getattr(self.sample_shuttle_A_status, f"sample{slot_nr}").set(val) + elif container == "B": + getattr(self.sample_shuttle_B_status, f"sample{slot_nr}").set(val) + elif container == "C": + getattr(self.sample_shuttle_C_status, f"sample{slot_nr}").set(val) + elif container == "O": + getattr(self.sample_status, f"sample{slot_nr}").set(val) + + def get_sample_status(self, container: str, slot_nr: int) -> str: + if container == "A": + val = getattr(self.sample_shuttle_A_status, f"sample{slot_nr}").get() + elif container == "B": + val = getattr(self.sample_shuttle_B_status, f"sample{slot_nr}").get() + elif container == "C": + val = getattr(self.sample_shuttle_C_status, f"sample{slot_nr}").get() + elif container == "O": + val = getattr(self.sample_status, f"sample{slot_nr}").get() + else: + return "none" + return self._status_val_to_str(val) + + def get_sample_owner(self, container: str, slot_nr: int) -> str: + if container == "A": + val = getattr(self.sample_shuttle_A_owners, f"sample{slot_nr}").get() + elif container == "B": + val = getattr(self.sample_shuttle_B_owners, f"sample{slot_nr}").get() + elif container == "C": + val = getattr(self.sample_shuttle_C_owners, f"sample{slot_nr}").get() + elif container == "O": + val = getattr(self.sample_owners, f"sample{slot_nr}").get() + else: + return "" + return str(val) if val else "" + + def get_sample_name_and_owner(self, container: str, slot_nr: int) -> tuple: + return self.get_sample_name(container, slot_nr), self.get_sample_owner(container, slot_nr) def set_shuttle_slot(self, container: str, slot_nr: int) -> bool: if slot_nr > 6: @@ -176,21 +362,49 @@ class OMNYSampleStorage(Device): getattr(self.parking_placed, f"parking{slot_nr}").set(0) getattr(self.parking_names, f"parking{slot_nr}").set("none") - def set_sample_in_gripper(self, name: str) -> bool: + def set_sample_in_gripper(self, name: str, owner: str = "") -> bool: self.sample_in_gripper.set(1) self.sample_in_gripper_name.set(name) + self.sample_in_gripper_owner.set(owner) + self.sample_in_gripper_status.set(self._status_str_to_val("none")) def unset_sample_in_gripper(self) -> bool: self.sample_in_gripper.set(0) self.sample_in_gripper_name.set("-") + self.sample_in_gripper_owner.set("") + self.sample_in_gripper_status.set(self._status_str_to_val("none")) - def set_sample_in_samplestage(self, name: str) -> bool: + def set_sample_in_samplestage(self, name: str, owner: str = "") -> bool: self.sample_in_samplestage.set(1) self.sample_in_samplestage_name.set(name) + self.sample_in_samplestage_owner.set(owner) + self.sample_in_samplestage_status.set(self._status_str_to_val("none")) def unset_sample_in_samplestage(self) -> bool: self.sample_in_samplestage.set(0) self.sample_in_samplestage_name.set("-") + self.sample_in_samplestage_owner.set("") + self.sample_in_samplestage_status.set(self._status_str_to_val("none")) + + def get_sample_owner_in_gripper(self) -> str: + val = self.sample_in_gripper_owner.get() + return str(val) if val else "" + + def get_sample_owner_in_samplestage(self) -> str: + val = self.sample_in_samplestage_owner.get() + return str(val) if val else "" + + def set_sample_status_in_gripper(self, status: str) -> bool: + self.sample_in_gripper_status.set(self._status_str_to_val(status)) + + def get_sample_status_in_gripper(self) -> str: + return self._status_val_to_str(self.sample_in_gripper_status.get()) + + def set_sample_status_in_samplestage(self, status: str) -> bool: + self.sample_in_samplestage_status.set(self._status_str_to_val(status)) + + def get_sample_status_in_samplestage(self) -> str: + return self._status_val_to_str(self.sample_in_samplestage_status.get()) def is_sample_slot_used(self, container, slot_nr: int) -> bool: if container == "A": @@ -247,6 +461,23 @@ class OMNYSampleStorage(Device): returnvalue = i return returnvalue + def _describe_slot(self, container: str, slot_nr: int) -> str: + """Sample name, plus owner/measured-status suffix when occupied -- + e.g. "sample1 (bob) ◐ started". Used by show_all() so the + overview surfaces owner+status without changing get_sample_name()'s + own plain-name contract (still used elsewhere, e.g. transfers).""" + name = self.get_sample_name(container, slot_nr) + if not self.is_sample_slot_used(container, slot_nr): + return name + owner = self.get_sample_owner(container, slot_nr) + status = self.get_sample_status(container, slot_nr) + suffix = f" ({owner})" if owner else "" + if status == "started": + suffix += " ◐ started" + elif status == "completed": + suffix += " ✓ completed" + return name + suffix + def show_all(self): t = PrettyTable() red = "\x1b[91m" @@ -262,22 +493,32 @@ class OMNYSampleStorage(Device): field_names = [""] for ax in [1, 3, 5]: row = [] - row.extend([self.get_sample_name(ch, ax)]) + row.extend([self._describe_slot(ch, ax)]) row.extend(str(ax)) row.extend(str(ax + 1)) - row.extend([self.get_sample_name(ch, ax + 1)]) + row.extend([self._describe_slot(ch, ax + 1)]) t.add_row(row) t.header = False t.vrules = FRAME print(t) if self.is_sample_in_samplestage(): - print(f"\n\n Sample stage: {self.get_sample_name_in_samplestage()}") + name = self.get_sample_name_in_samplestage() + owner = self.get_sample_owner_in_samplestage() + status = self.get_sample_status_in_samplestage() + suffix = f" ({owner})" if owner else "" + if status == "started": + suffix += " ◐ started" + elif status == "completed": + suffix += " ✓ completed" + print(f"\n\n Sample stage: {name}{suffix}") else: print(f"\n\n Sample stage: no sample") if self.is_sample_in_gripper(): - print(f"\n Gripper: {self.get_sample_name_in_gripper()}\n") + name = self.get_sample_name_in_gripper() + owner = self.get_sample_owner_in_gripper() + print(f"\n Gripper: {name}" + (f" ({owner})" if owner else "") + "\n") else: print(f"\n Gripper: no sample\n") @@ -287,8 +528,7 @@ class OMNYSampleStorage(Device): row = [] row.extend([f"Position {i:3d}"]) if self.is_sample_slot_used("O", i): - name = self.get_sample_name("O", i) - row.extend([name]) + row.extend([self._describe_slot("O", i)]) else: row.extend(["free"]) t.add_row(row) @@ -301,10 +541,14 @@ class OMNYSampleStorage(Device): print(" To get an overview use dev.omny_samples.show_all()") print(" Modify a slot:") print(" dev.omny_samples.unset_sample_slot('system',position)") - print(" dev.omny_samples.set_sample_slot('system',position,'name')") + print(" dev.omny_samples.set_sample_slot('system',position,'name','owner')") print(" system can be A, B, C, O") - print(" dev.omny_samples.set_sample_in_gripper('name') / unset_sample_in_gripper()") - print(" dev.omny_samples.set_sample_in_samplestage('name'), unset_sample_in_samplestage()") + print(" dev.omny_samples.get_sample_owner('system',position)") + print(" dev.omny_samples.set_sample_status('system',position,'status')") + print(" status is one of 'none', 'started', 'completed'") + print(" dev.omny_samples.get_sample_status('system',position)") + print(" dev.omny_samples.set_sample_in_gripper('name','owner') / unset_sample_in_gripper()") + print(" dev.omny_samples.set_sample_in_samplestage('name','owner'), unset_sample_in_samplestage()") print(" dev.omny_samples.set_shuttle_slot(container, slot_nr) / unset_shuttle_slot(slot_nr)") print(" dev.omny_samples.set_shuttle_slot('A',2)") print(" omny.otransfer_help()") diff --git a/csaxs_bec/devices/sim/sim_omny.py b/csaxs_bec/devices/sim/sim_omny.py index d5e60ead..19a6015e 100644 --- a/csaxs_bec/devices/sim/sim_omny.py +++ b/csaxs_bec/devices/sim/sim_omny.py @@ -266,17 +266,41 @@ class SimOMNYSampleStorage(OMNYSampleStorage): # 0-20-only range check, but OMNY's own "O" fixed positions # legitimately include 32-34/100/101), which would raise here # for exactly the positions this needs to clear. - for shuttle_placed, shuttle_names in ( - (storage.sample_shuttle_A_placed, storage.sample_shuttle_A_names), - (storage.sample_shuttle_B_placed, storage.sample_shuttle_B_names), - (storage.sample_shuttle_C_placed, storage.sample_shuttle_C_names), + # Same clearing for the owner/status signals added alongside + # placed/names -- untouched positions should read owner="" and + # status="none", not the mock-PV "0" default (an invalid status + # string, would fail _status_val_to_str's lookup silently back + # to "none" anyway, but owner would show a stray "0"). + for shuttle_placed, shuttle_names, shuttle_owners, shuttle_status in ( + ( + storage.sample_shuttle_A_placed, + storage.sample_shuttle_A_names, + storage.sample_shuttle_A_owners, + storage.sample_shuttle_A_status, + ), + ( + storage.sample_shuttle_B_placed, + storage.sample_shuttle_B_names, + storage.sample_shuttle_B_owners, + storage.sample_shuttle_B_status, + ), + ( + storage.sample_shuttle_C_placed, + storage.sample_shuttle_C_names, + storage.sample_shuttle_C_owners, + storage.sample_shuttle_C_status, + ), ): for slot_nr in shuttle_placed.component_names: getattr(shuttle_placed, slot_nr).set(0) getattr(shuttle_names, slot_nr).set("-") + getattr(shuttle_owners, slot_nr).set("") + getattr(shuttle_status, slot_nr).set(OMNYSampleStorage.STATUS_NONE) for slot_nr in storage.sample_placed.component_names: getattr(storage.sample_placed, slot_nr).set(0) getattr(storage.sample_names, slot_nr).set("-") + getattr(storage.sample_owners, slot_nr).set("") + getattr(storage.sample_status, slot_nr).set(OMNYSampleStorage.STATUS_NONE) for slot_nr in storage.parking_placed.component_names: getattr(storage.parking_placed, slot_nr).set(0) getattr(storage.parking_names, slot_nr).set("none")