From 7082b9ff7b960cb3c7cf8f2c25fc24f5d32dd88f Mon Sep 17 00:00:00 2001 From: hitz_s Date: Wed, 9 Sep 2026 11:32:05 +0200 Subject: [PATCH] wip --- .../digital_twin_core/digital_twin_core.py | 56 +- .../widgets/digital_twin/digital_twin.py | 84 ++- .../widgets/scheduler/item_dialog.py | 659 ++++++++++-------- 3 files changed, 484 insertions(+), 315 deletions(-) diff --git a/debye_bec/bec_ipython_client/plugins/digital_twin_core/digital_twin_core.py b/debye_bec/bec_ipython_client/plugins/digital_twin_core/digital_twin_core.py index b147608..243afa1 100644 --- a/debye_bec/bec_ipython_client/plugins/digital_twin_core/digital_twin_core.py +++ b/debye_bec/bec_ipython_client/plugins/digital_twin_core/digital_twin_core.py @@ -87,7 +87,7 @@ class DigitalTwinCore: self.beamline = get_beamline_id() self.offset_file = Path() match self.beamline: - case "x01da": + case BeamlineId.X01DA: self.offset_file = OFFSET_FILE_X01DA case "x10da": self.offset_file = OFFSET_FILE_X10DA @@ -115,7 +115,7 @@ class DigitalTwinCore: without error, else {"success": False, "moved": [...], "failures": {motor: error}}. """ - positions = self.calc_positions(self.beamline, config) + positions = self.calc_positions(config) positions = self.apply_offsets(positions, nested_config=True) managers: dict[int, _ExclusiveGroupManager] = {} @@ -213,8 +213,7 @@ class DigitalTwinCore: config[axis] -= axis_offsets["offset"] return config - @staticmethod - def calc_positions(beamline: BeamlineId, cfg: ConfigDict) -> dict[str, dict[str, float]]: + def calc_positions(self, cfg: ConfigDict) -> dict[str, dict[str, float]]: """ Calculates the positions of axes based on a beamline config. @@ -363,17 +362,9 @@ class DigitalTwinCore: pos["bm1_try"] = {"value": bm1_beam_height} ## Focusing Mirror - p = bl.fm.center[1] - q = (cfg["ot_es1_trz"] + cfg["es1man_trz"]) - bl.fm.center[1] - f = (p * q) / (p + q) # focal length - - # Bender radius - if cfg["fm_qy"] is None: - radius = 2 * q / np.sin(cfg["fm_rotx"]) # ideal bending radius for focused beam - else: - radius = ( - 2 * cfg["fm_qy"] / np.sin(cfg["fm_rotx"]) - ) # ideal bending radius for unfocused beam + radius = self.calc_fm_bnd_radius( + cfg["ot_es1_trz"] + cfg["es1man_trz"], cfg["fm_qy"], cfg["fm_rotx"] + ) pos["fm_bnd_radius"] = {"value": radius * 1e-6} # Convert to km # Pitch @@ -448,7 +439,7 @@ class DigitalTwinCore: ## Optical Table - if beamline == "x01da": + if self.beamline == BeamlineId.X01DA: # TRY d = bl.ehWindow.center[1] - bl.fm.center[1] ot_height = fm_beam_height - d * np.tan(-(2 * cfg["cm_pitch"] - 2 * cfg["fm_rotx"])) @@ -622,6 +613,30 @@ class DigitalTwinCore: return -(low + high) / 2 return None + @staticmethod + def calc_fm_bnd_radius(smpl: float, fm_qy: None | float, fm_rotx: float): + """ + Calculate the fm bender radius + + Args: + smpl(float): Sample position in mm + fm_qy(None|float): FM qy value + fm_rotx(float: FM rotx position + + Returns: + float: bending radius in m + """ + # p = bl.fm.center[1] + q = smpl - bl.fm.center[1] + # f = (p * q) / (p + q) # focal length + + # Bender radius + if fm_qy is None: + radius = 2 * q / np.sin(fm_rotx) # ideal bending radius for focused beam + else: + radius = 2 * fm_qy / np.sin(fm_rotx) # ideal bending radius for unfocused beam + return radius + @staticmethod def mo1_energy_resolution(xtal: Literal["Si111", "Si311"], energy: float) -> float: """ @@ -1032,6 +1047,15 @@ class DigitalTwinCore: return bl.es2.center[1] raise ValueError(f"Table {table} not found in beamline parameter file") + @staticmethod + def smpl_pos_to_table(smpl_pos: float) -> str: + """Return the table name based on the sample position.""" + if np.isclose(smpl_pos, bl.es1.center[1]): + return bl.es1.name + if np.isclose(smpl_pos, bl.es2.center[1]): + return bl.es2.name + raise ValueError(f"Sample position {smpl_pos} not found in beamline parameter file") + @staticmethod def calc_sideview(cfg: ConfigDict) -> DataDict: """ diff --git a/debye_bec/bec_widgets/widgets/digital_twin/digital_twin.py b/debye_bec/bec_widgets/widgets/digital_twin/digital_twin.py index 0901da6..ea583c7 100644 --- a/debye_bec/bec_widgets/widgets/digital_twin/digital_twin.py +++ b/debye_bec/bec_widgets/widgets/digital_twin/digital_twin.py @@ -34,9 +34,8 @@ from qtpy.QtWidgets import ( QWidget, ) -from ....bec_ipython_client.plugins.digital_twin_core.beamline import get_beamline_id from ....bec_ipython_client.plugins.digital_twin_core.digital_twin_core import DigitalTwinCore -from ....bec_ipython_client.plugins.digital_twin_core.types import ConfigDict +from ....bec_ipython_client.plugins.digital_twin_core.types import BeamlineId, ConfigDict from ..edge_selector import EdgeSelector from .panels.input_panel import InputPanel from .panels.mover_panel import MoverPanel @@ -65,7 +64,7 @@ class DigitalTwin(BECWidget, QWidget): self.core = DigitalTwinCore() - self.beamline = get_beamline_id() + self.beamline = self.core.beamline # Debugging, override beamline! # self.beamline = BeamlineId.X10DA @@ -132,7 +131,7 @@ class DigitalTwin(BECWidget, QWidget): self.input.fm_rotx.value_changed_connect(self.calc_assistant) self.input.fm_focx.value_changed_connect(self.calc_assistant) self.input.fm_focy.value_changed_connect(self.calc_assistant) - if self.beamline == "x01da": + if self.beamline == BeamlineId.X01DA: self.input.ot_es1_trz.value_changed_connect(self.calc_assistant) else: self.input.table.activated_connect(self.calc_assistant) @@ -217,7 +216,7 @@ class DigitalTwin(BECWidget, QWidget): "es0wi_try", "es1man_trz", ] - if self.beamline == "x01da": # X01DA specific devices + if self.beamline == BeamlineId.X01DA: # X01DA specific devices devices.extend( [ "cm_bnd_radius", @@ -371,7 +370,7 @@ class DigitalTwin(BECWidget, QWidget): elif fm_focus == "Focused": fm_rotx = self.input.fm_rotx_ideal.value() fm_qy = None - else: # Focused + else: # Defocused fm_rotx = self.input.fm_rotx_ideal.value() fm_qy = self.qy @@ -383,7 +382,7 @@ class DigitalTwin(BECWidget, QWidget): assert cm_trx is not None, f"No cm_trx found for given stripe {cm_stripe}!" assert fm_trx is not None, f"No fm_trx found for given stripe {fm_stripe}!" - if self.beamline == "x01da": + if self.beamline == BeamlineId.X01DA: ot_es1_trz = self.input.ot_es1_trz.value() smpl_to_xrd = self.input.smpl_to_xrd.value() else: @@ -426,6 +425,57 @@ class DigitalTwin(BECWidget, QWidget): # logger.info(f'Config created: {config}') return config + def set_assistant_config(self, config: ConfigDict) -> None: + """ + Applies an assistant config to the input Qt widgets. + + Args: + config: Assistant configuration, as returned by + ``get_assistant_config``. The acceleration and angular + values are expected to be in SI units. + """ + self.input.energy.set_number(config["energy"]) + self.input.sldi_hacc.set_number(config["h_acc"] * 1e3) + self.input.sldi_vacc.set_number(config["v_acc"] * 1e3) + self.input.cm_pitch.set_number(-config["cm_pitch"] * 1e3) + self.input.cm_stripe.set_current_text(config["cm_stripe"]) + self.input.fm_stripe.set_current_text(config["fm_stripe"]) + self.input.mo1_mode.set_current_text(config["mo1_mode"]) + self.input.mo1_xtal.set_current_text(config["mo1_xtal"]) + + self.bragg_angle = config["mo1_bragg"] + + fm_rotx = -config["fm_rotx"] * 1e3 + + self.qy = config["fm_qy"] + fm_rotx_real = 2 * config["cm_pitch"] - config["fm_rotx"] + smpl = config["ot_es1_trz"] + config["es1man_trz"] + radius = self.core.calc_fm_bnd_radius(smpl, config["fm_qy"], config["fm_rotx"]) + fm_focx, fm_focy = self.core.calc_beamsize( + config["h_acc"], config["v_acc"], config["fm_stripe"], fm_rotx_real * 1e-3, radius, smpl + ) + if config["fm_qy"] is not None: + self.input.fm_focus.set_current_text("Defocused") + self.input.fm_focx.set_number(fm_focx) + self.input.fm_focy.set_number(fm_focy) + elif fm_focx < 0.08 and fm_focy < 0.08: + self.input.fm_focus.set_current_text("Focused") + else: + self.input.fm_focus.set_current_text("Manual") + self.input.fm_rotx_ideal.setValue(fm_rotx) + + if self.beamline == BeamlineId.X01DA: + self.input.ot_es1_trz.set_number(config["ot_es1_trz"]) + if config["smpl_to_xrd"] is not None: + self.input.smpl_to_xrd.set_number(config["smpl_to_xrd"]) + else: + table = self.core.smpl_pos_to_table(config["ot_es1_trz"]) + self.input.table.set_current_text(table) + + self.input.es1man_trz.set_number(config["es1man_trz"]) + + self.calc_assistant(identifier="init") + def get_reality_config(self) -> ConfigDict: """ Assembles the digital twin config based on the real axis positions. @@ -451,7 +501,7 @@ class DigitalTwin(BECWidget, QWidget): fm_rotx_real = 2 * cm_pitch - fm_rotx es1man_trz = self.dev.es1man_trz.read(cached=True)["es1man_trz"]["value"] - if self.beamline == "x01da": + if self.beamline == BeamlineId.X01DA: ot_es1_trz = self.dev.ot_es1_trz.read(cached=True)["ot_es1_trz"]["value"] ot_es2_trz = self.dev.ot_es2_trz.read(cached=True)["ot_es2_trz"]["value"] smpl_to_xrd = ot_es2_trz - ot_es1_trz - es1man_trz + 32 @@ -519,7 +569,7 @@ class DigitalTwin(BECWidget, QWidget): self.mover.fm_rotx.set_feedback(fm_rotx) self.mover.fm_roty.set_feedback(self.dev.fm_roty.read(cached=True)["fm_roty"]["value"]) self.mover.fm_rotz.set_feedback(self.dev.fm_rotz.read(cached=True)["fm_rotz"]["value"]) - if self.beamline == "x01da": + if self.beamline == BeamlineId.X01DA: self.mover.sl2_centery.set_feedback( self.dev.sl2_centery.read(cached=True)["sl2_centery"]["value"] ) @@ -527,7 +577,7 @@ class DigitalTwin(BECWidget, QWidget): self.dev.sl2_gapy.read(cached=True)["sl2_gapy"]["value"] ) self.mover.bm2_try.set_feedback(self.dev.bm2_try.read(cached=True)["bm2_try"]["value"]) - if self.beamline == "x01da": + if self.beamline == BeamlineId.X01DA: self.mover.ot_try.set_feedback(self.dev.ot_try.read(cached=True)["ot_try"]["value"]) self.mover.ot_rotx.set_feedback(self.dev.ot_rotx.read(cached=True)["ot_rotx"]["value"]) self.mover.ot_es1_trz.set_feedback(ot_es1_trz) @@ -568,7 +618,7 @@ class DigitalTwin(BECWidget, QWidget): pos["fm_rotx"] = self.dev.fm_rotx.read(cached=True)["fm_rotx"]["value"] pos["fm_bnd_radius"] = self.dev.fm_bnd_radius.read(cached=True)["fm_bnd_radius"]["value"] pos["es1man_trz"] = self.dev.es1man_trz.read(cached=True)["es1man_trz"]["value"] - if self.beamline == "x01da": + if self.beamline == BeamlineId.X01DA: pos["ot_es1_trz"] = self.dev.ot_es1_trz.read(cached=True)["ot_es1_trz"]["value"] pos["ot_es2_trz"] = self.dev.ot_es2_trz.read(cached=True)["ot_es2_trz"]["value"] pos["smpl_to_xrd"] = pos["ot_es2_trz"] - pos["ot_es1_trz"] - pos["es1man_trz"] + 32 @@ -598,7 +648,7 @@ class DigitalTwin(BECWidget, QWidget): fm_rotx_real = 2 * pos["cm_rotx"] - pos["fm_rotx"] self.input.fm_rotx.set_number(fm_rotx_real) - if self.beamline == "x01da": + if self.beamline == BeamlineId.X01DA: self.input.ot_es1_trz.set_number(pos["ot_es1_trz"]) self.input.smpl_to_xrd.set_number(pos["smpl_to_xrd"]) else: @@ -768,7 +818,7 @@ class DigitalTwin(BECWidget, QWidget): @SafeSlot() def open_edge_selector(self, *_): match self.beamline: - case "x01da": + case BeamlineId.X01DA: dlg = EdgeSelector(self, llim=X01DA_E_MIN, hlim=X01DA_E_MAX) case "x10da": dlg = EdgeSelector(self, llim=X10DA_E_MIN, hlim=X10DA_E_MAX) @@ -858,7 +908,7 @@ class DigitalTwin(BECWidget, QWidget): Calculates the positions for the axes based on the assistant values """ config = self.get_assistant_config() - out = self.core.calc_positions(self.beamline, config) + out = self.core.calc_positions(config) out = self.core.apply_offsets(out, nested_config=True) self.mover.sldi_gapx.set_target(out["sldi_gapx"]["value"]) @@ -879,11 +929,11 @@ class DigitalTwin(BECWidget, QWidget): self.mover.fm_rotx.set_target(out["fm_rotx"]["value"]) self.mover.fm_roty.set_target(out["fm_roty"]["value"]) self.mover.fm_rotz.set_target(out["fm_rotz"]["value"]) - if self.beamline == "x01da": + if self.beamline == BeamlineId.X01DA: self.mover.sl2_centery.set_target(out["sl2_centery"]["value"]) self.mover.sl2_gapy.set_target(out["sl2_gapy"]["value"]) self.mover.bm2_try.set_target(out["bm2_try"]["value"]) - if self.beamline == "x01da": + if self.beamline == BeamlineId.X01DA: self.mover.ot_try.set_target(out["ot_try"]["value"]) self.mover.ot_rotx.set_target(out["ot_rotx"]["value"]) self.mover.ot_es1_trz.set_target(out["ot_es1_trz"]["value"]) @@ -942,7 +992,7 @@ class DigitalTwin(BECWidget, QWidget): ) fm_stripe = self.input.fm_stripe.currentText() es1man_trz = self.input.es1man_trz.value() - if self.beamline == "x01da": + if self.beamline == BeamlineId.X01DA: ot_es1_trz = self.input.ot_es1_trz.value() else: table = self.input.table.currentText() diff --git a/debye_bec/bec_widgets/widgets/scheduler/item_dialog.py b/debye_bec/bec_widgets/widgets/scheduler/item_dialog.py index 02066bf..aaae858 100644 --- a/debye_bec/bec_widgets/widgets/scheduler/item_dialog.py +++ b/debye_bec/bec_widgets/widgets/scheduler/item_dialog.py @@ -2,9 +2,13 @@ The dialog behind the schedule widget's "Add..."/"Edit..." buttons. Rather than asking the operator to remember and type -`scans.xas_simple_scan(12000, 14000, 2, 10)`-style commands, this presents: +`scans.xas_simple_scan(12000, 14000, 2, 10)`-style commands, this presents +a dropdown of item types; picking one swaps in that item's own form below +the dropdown (a `QStackedWidget` page per item, kept in the same order as +the dropdown so a combo index also works as a stack index - see +`_add_page`). The available items are: -- a "Scan" tab: BEC's own `bec_widgets` `ScanControl` widget, embedded +- "ScanControl": BEC's own `bec_widgets` `ScanControl` widget, embedded as-is - scan selection, its live-generated per-scan argument form, docs tooltips, metadata, "recall last scan parameters", all of it. Reusing it instead of a plugin-owned reimplementation means this stays @@ -13,31 +17,40 @@ Rather than asking the operator to remember and type GUI. `ScanControl.button_run_scan` ("Start") is hidden here: this dialog only ever wants the configured scan name/args/kwargs, never an immediate submission - see `_collect_scan_result`. -- a "Move" tab: pick a device and a target value/relative flag; -- a "Digital Twin" tab: the beamline-alignment `DigitalTwin` widget, - embedded the same way as ScanControl. Instead of submitting anything - itself, OK captures a *snapshot* of `DigitalTwin.get_assistant_config()` - and stores it - execution later calls `move_all_axes(...)` with that - frozen config, so editing the (possibly separately open) Digital Twin - widget afterwards never affects an already-added schedule item, exactly - like a Scan item's captured args/kwargs aren't affected by reopening +- "Move": pick a device and a target value/relative flag. +- "Digital Twin": the beamline-alignment `DigitalTwin` widget, embedded + the same way as ScanControl. Instead of submitting anything itself, OK + captures a *snapshot* of `DigitalTwin.get_assistant_config()` and stores + it - execution later calls `move_all_axes(...)` with that frozen + config, so editing the (possibly separately open) Digital Twin widget + afterwards never affects an already-added schedule item, exactly like a + Scan item's captured args/kwargs aren't affected by reopening ScanControl elsewhere. See `_collect_digital_twin_result`. -- an "Other" tab: a free-text field for anything else (including RPC - calls to other widgets), plus a couple of beamline-specific quick-fill - forms (ionization chamber gas mix, reference foil). +- "Absorber", "Ionization Chamber", "Reference foil changer", "Auto-Gain": + beamline-specific quick-fill forms. Each has its own `_collect_*_result` + that either builds a command straight from the form fields or raises + `ValueError` with a message explaining what's missing or invalid - + `_on_accept` catches that and shows it in a message box, rather than + silently doing nothing the way the old "Generate command" buttons used + to. Each also has a `_prefill_*_form` so "Edit..." can reselect the + exact same field values, not just show the generated command as text. +- "Other": a free-text field for anything else (including RPC calls to + other widgets); also the fallback shown, pre-filled with the raw + command, for any saved item this dialog can't reconstruct into a form + (see `_apply_initial`). -Whichever tab is used, the dialog's only output is the same kind of plain -command string the executor already knows how to run - this dialog adds a -friendlier way to *build* that string, it doesn't change what happens with -it afterwards. `kind`/`form_state` are carried along purely so "Edit..." -can reopen the dialog pre-filled instead of asking the user to start over -- except for the Digital Twin tab, where "Edit..." currently falls back to -showing the generated command as read/write text on the "Other" tab rather -than reloading the captured config back into DigitalTwin's input fields; -see the note on `_apply_initial`. +Whichever item is selected, the dialog's only output is the same kind of +plain command string the executor already knows how to run - this dialog +adds a friendlier way to *build* that string, it doesn't change what +happens with it afterwards. `kind`/`form_state` are carried along purely +so "Edit..." can reopen the dialog pre-filled instead of asking the user +to start over - except for the Digital Twin item, where "Edit..." +currently falls back to showing the generated command as read/write text +on the "Other" page rather than reloading the captured config back into +DigitalTwin's input fields; see the note on `_apply_initial`. The Digital Twin item reuses `kind="move"` (not a new kind): it submits -through `scans.mv(...)`, exactly like the plain Move tab, so it should be +through `scans.mv(...)`, exactly like the plain Move item, so it should be treated the same way everywhere else in the plugin that branches on kind - not guard-protected, and counted under the "Movements" notification toggle. A `form_state["source"] = "digital_twin"` marker is only used @@ -63,13 +76,12 @@ from qtpy.QtWidgets import ( QDialogButtonBox, QDoubleSpinBox, QFormLayout, - QGroupBox, QHBoxLayout, QLabel, QLineEdit, QMessageBox, QScrollArea, - QTabWidget, + QStackedWidget, QVBoxLayout, QWidget, ) @@ -78,12 +90,10 @@ from .qt_widgets import MyButton logger = bec_logger.logger -# Tab indices, named instead of magic numbers now that there are four - -# see _collect_result()/_apply_initial(). -_TAB_SCAN = 0 -_TAB_MOVE = 1 -_TAB_DIGITAL_TWIN = 2 -_TAB_OTHER = 3 +# Beamlines that get the XAS scan control, the Digital Twin item, and the +# beamline-specific quick-fill items (Absorber, Ionization Chamber, +# Reference foil changer, Auto-Gain). +_XAS_BEAMLINES = ("x01da", "x10da") class ScheduleItemDialog(QDialog): @@ -107,9 +117,13 @@ class ScheduleItemDialog(QDialog): self._client = client self.beamline = beamline - if self.beamline in ["x01da", "x10da"]: + self._is_xas_beamline = self.beamline in _XAS_BEAMLINES + + if self._is_xas_beamline: logger.info(f"Loading bl-specific modules for beamline {self.beamline}") - from ....bec_ipython_client.plugins.digital_twin_core.digital_twin_core import DigitalTwinCore + from ....bec_ipython_client.plugins.digital_twin_core.digital_twin_core import ( + DigitalTwinCore, + ) from ..digital_twin.digital_twin import DigitalTwin from ..edge_selector import EdgeSelector from ..scan_control_xas.scan_control_xas import ScanControlXAS @@ -120,14 +134,38 @@ class ScheduleItemDialog(QDialog): self.ScanControlXAS = ScanControlXAS layout = QVBoxLayout(self) - self.tabs = QTabWidget() - layout.addWidget(self.tabs) - self._build_scan_tab() - self._build_move_tab() - if self.beamline in ["x01da", "x10da"]: - self._build_digital_twin_tab() - self._build_custom_tab() + self.item_combo = QComboBox() + layout.addWidget(self.item_combo) + + self.stack = QStackedWidget() + layout.addWidget(self.stack) + + self._build_scan_page() + self._build_move_page() + if self._is_xas_beamline: + self._build_digital_twin_page() + self._build_abs_page() + self._build_ic_page() + self._build_reffoil_page() + self._build_auto_gain_page() + self._build_other_page() + + self.item_combo.currentIndexChanged.connect(self.stack.setCurrentIndex) + + # Dispatch table keyed by the dropdown item's data (see + # _add_page) - one place to register a new item type instead of + # scattering per-item checks through _collect_result(). + self._collectors = { + "scan": self._collect_scan_result, + "move": self._collect_move_result, + "digital_twin": self._collect_digital_twin_result, + "abs": self._collect_abs_result, + "ic": self._collect_ic_result, + "reffoil": self._collect_reffoil_result, + "auto_gain": self._collect_auto_gain_result, + "other": self._collect_custom_result, + } buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel) buttons.button(QDialogButtonBox.Ok).setText("Add") @@ -145,23 +183,42 @@ class ScheduleItemDialog(QDialog): self._apply_initial(initial or {}) # ------------------------------------------------------------------ # - # Scan tab - embeds BEC's own ScanControl widget + # Dropdown/page plumbing # ------------------------------------------------------------------ # - def _build_scan_tab(self): - tab = QWidget() - layout = QVBoxLayout(tab) + def _add_page(self, key: str, label: str, widget: QWidget): + # The combo and the stack are always appended to together, so a + # combo index doubles as a stack index - no separate key->index + # map needed. `key` is what _collect_result()/_apply_initial() + # dispatch on. + self.item_combo.addItem(label, key) + self.stack.addWidget(widget) + + def _select_page(self, key: str) -> bool: + """Switch the dropdown to `key`'s page; False if that item isn't available.""" + idx = self.item_combo.findData(key) + if idx < 0: + return False + self.item_combo.setCurrentIndex(idx) + return True + + # ------------------------------------------------------------------ # + # ScanControl - embeds BEC's own ScanControl widget + # ------------------------------------------------------------------ # + def _build_scan_page(self): + page = QWidget() + layout = QVBoxLayout(page) # client=None resolves to the same process-wide BEC client # (bec_dispatcher.client) our own widget uses - no second Redis # connection is opened. - if self.beamline in ["x01da", "x10da"]: - self.scan_control = self.ScanControlXAS(parent=tab, client=self._client) + if self._is_xas_beamline: + self.scan_control = self.ScanControlXAS(parent=page, client=self._client) else: - self.scan_control = ScanControl(parent=tab, client=self._client) + self.scan_control = ScanControl(parent=page, client=self._client) self.scan_control.button_run_scan.hide() layout.addWidget(self.scan_control) - self.tabs.addTab(tab, "Scan") + self._add_page("scan", "ScanControl", page) def _collect_scan_result(self) -> dict: # Same call ScanControl.run_scan() makes before actually @@ -183,7 +240,7 @@ class ScheduleItemDialog(QDialog): "form_state": {"scan_name": scan_name, "args": args, "kwargs": kwargs}, } - def _prefill_scan_tab(self, scan_name: str, args: list, kwargs: dict): + def _prefill_scan_page(self, scan_name: str, args: list, kwargs: dict): # ScanControl restores parameters for a scan from its own config # cache (see `ScanControl.restore_scan_parameters`); pre-loading # that cache before switching to the scan reuses that mechanism @@ -195,15 +252,14 @@ class ScheduleItemDialog(QDialog): self.scan_control.restore_scan_parameters(scan_name) # ------------------------------------------------------------------ # - # Move tab + # Move # ------------------------------------------------------------------ # - def _build_move_tab(self): - tab = QWidget() - form = QFormLayout(tab) + def _build_move_page(self): + page = QWidget() + form = QFormLayout(page) form.setFieldGrowthPolicy(QFormLayout.FieldGrowthPolicy.FieldsStayAtSizeHint) self.move_device_combo = DeviceComboBox(self, device_filter=[BECDeviceFilter.POSITIONER]) - form.addRow("Device", self.move_device_combo) self.move_value_spin = QDoubleSpinBox() @@ -214,13 +270,14 @@ class ScheduleItemDialog(QDialog): self.move_device_combo.currentIndexChanged.connect(self._adjust_spinbox) - self.tabs.addTab(tab, "Move") + self._add_page("move", "Move", page) def _adjust_spinbox(self, _) -> None: - prec = self._dev[self.move_device_combo.currentText()].precision - units = self._dev[self.move_device_combo.currentText()].egu() - ll = self._dev[self.move_device_combo.currentText()].low_limit - hl = self._dev[self.move_device_combo.currentText()].high_limit + device = self._dev[self.move_device_combo.currentText()] + prec = device.precision + units = device.egu() + ll = device.low_limit + hl = device.high_limit self.move_value_spin.setDecimals(prec) self.move_value_spin.setSuffix(f" {units}") @@ -245,12 +302,19 @@ class ScheduleItemDialog(QDialog): "form_state": {"device_name": device_name, "value": value, "relative": relative}, } + def _prefill_move_page(self, state: dict): + idx = self.move_device_combo.findText(state.get("device_name", "")) + if idx >= 0: + self.move_device_combo.setCurrentIndex(idx) + self.move_value_spin.setValue(float(state.get("value", 0.0))) + self.move_relative_check.setChecked(bool(state.get("relative", False))) + # ------------------------------------------------------------------ # - # Digital Twin tab - embeds the beamline-alignment DigitalTwin widget + # Digital Twin - embeds the beamline-alignment DigitalTwin widget # ------------------------------------------------------------------ # - def _build_digital_twin_tab(self): - tab = QWidget() - outer = QVBoxLayout(tab) + def _build_digital_twin_page(self): + page = QWidget() + outer = QVBoxLayout(page) # DigitalTwin defaults to a large fixed size (it's normally its # own top-level window) - wrap it in a scroll area so this dialog @@ -275,20 +339,15 @@ class ScheduleItemDialog(QDialog): hint.setStyleSheet("color: gray;") outer.addWidget(hint) - self.tabs.addTab(tab, "Digital Twin") + self._add_page("digital_twin", "Digital Twin", page) def _collect_digital_twin_result(self) -> dict: + # A frozen snapshot of the configured targets, not a live handle - + # editing the Digital Twin widget afterwards must not affect an + # already-added schedule item (see module docstring). config = self.digital_twin.get_assistant_config() - # beamline = self.digital_twin.beamline - - # # Init the class when the scheduler is opened - # digital_twin = DigitalTwinCore() - # # The command below would then execute the movement - # digital_twin.move_with_config(config) - cmd = f"digital_twin.move_with_config({config})" - - return {"command": f"{cmd}", "kind": "custom", "form_state": {"text": cmd}} + return {"command": cmd, "kind": "custom", "form_state": {"text": cmd, "config": config}} def _cleanup_digital_twin(self, *_): digital_twin = getattr(self, "digital_twin", None) @@ -303,124 +362,91 @@ class ScheduleItemDialog(QDialog): except Exception: # pylint: disable=broad-except logger.exception("Failed to clean up the embedded Digital Twin widget.") + def _prefill_digital_twin_page(self, state: dict): + config = state["config"] + self.digital_twin.set_assistant_config(config) + # ------------------------------------------------------------------ # - # Custom tab + # Absorber # ------------------------------------------------------------------ # - def _build_custom_tab(self): - tab = QWidget() - layout = QVBoxLayout(tab) - layout.addWidget( - QLabel( - "Free-form command, evaluated against `scans` and `dev` - use this for " - "anything the other tabs don't cover, e.g. an RPC call to another widget." - ) + def _build_abs_page(self): + if "abs" not in self._dev: + return + page = QWidget() + form = QFormLayout(page) + form.setFieldGrowthPolicy(QFormLayout.FieldGrowthPolicy.FieldsStayAtSizeHint) + + self.abs_selector = QComboBox() + self.abs_selector.addItems(["Open", "Force open", "Close"]) + form.addRow("Action", self.abs_selector) + + self._add_page("abs", "Absorber", page) + + def _collect_abs_result(self) -> dict: + action = self.abs_selector.currentText() + suffix = {"Open": "open()", "Force open": "open(force=True)", "Close": "close()"}.get( + action ) - self.custom_edit = QLineEdit() - # TODO Change to a different placeholder text - self.custom_edit.setPlaceholderText("scans.xas_simple_scan(12000, 14000, 2, 10)") - layout.addWidget(self.custom_edit) - - if self.beamline in ["x01da", "x10da"]: - abs_form = self._create_abs_form() - if abs_form is not None: - layout.addWidget(abs_form) - - ic_form = self._create_ionization_chamber_form() - if ic_form is not None: - layout.addWidget(ic_form) - - reffoil_form = self._create_reffoil_form() - if reffoil_form is not None: - layout.addWidget(reffoil_form) - - auto_gain_form = self._create_auto_gain_form() - if auto_gain_form is not None: - layout.addWidget(auto_gain_form) - - layout.addStretch(1) - self.tabs.addTab(tab, "Other") - - def _create_abs_form(self): - if "abs" in self._dev: - abs_group = QGroupBox("Frontend Absorber") - layout = QVBoxLayout(abs_group) - form = QFormLayout() - layout.addLayout(form) - form.setFieldGrowthPolicy(QFormLayout.FieldGrowthPolicy.FieldsStayAtSizeHint) - self.abs_selector = QComboBox() - self.abs_selector.addItems(["Open", "Force open", "Close"]) - form.addRow("Action", self.abs_selector) - - button_layout = QHBoxLayout() - generate_cmd = MyButton("Generate command", "default") - button_layout.addWidget(generate_cmd) - button_layout.addStretch(1) - layout.addLayout(button_layout) - - generate_cmd.clicked.connect(self._generate_abs_command) - - return abs_group - return None - - def _generate_abs_command(self): - match self.abs_selector.currentText(): - case "Open": - suffix = "open()" - case "Force open": - suffix = "open(force=True)" - case "Close": - suffix = "close()" + if suffix is None: + raise ValueError("Select an absorber action.") cmd = f"dev.abs.{suffix}" - self.custom_edit.setText(cmd) + return { + "command": cmd, + "kind": "custom", + "form_state": {"text": cmd, "source": "abs", "action": action}, + } - def _create_ionization_chamber_form(self): - if all(key in self._dev for key in ("ic0", "ic1", "ic2")): - ic_group = QGroupBox("Ionization chamber filling") - layout = QVBoxLayout(ic_group) - form = QFormLayout() - layout.addLayout(form) - form.setFieldGrowthPolicy(QFormLayout.FieldGrowthPolicy.FieldsStayAtSizeHint) - self.ic_selector = QComboBox() - self.ic_selector.addItems(["IC0", "IC1", "IC2"]) - gases = ["He", "N2", "Ar", "Kr"] - self.gas1 = QComboBox() - self.gas2 = QComboBox() - self.gas1.addItems(gases) - self.gas2.addItems(gases) - self.conc1 = QDoubleSpinBox() - self.conc2 = QDoubleSpinBox() - for conc in [self.conc1, self.conc2]: - conc.setDecimals(0) - conc.setSuffix(" %") - conc.setMinimum(0) - conc.setMaximum(100) - conc.setSingleStep(1) - self.pressure = QDoubleSpinBox() - self.pressure.setDecimals(3) - self.pressure.setSuffix(" bar abs") - self.pressure.setMinimum(1) - self.pressure.setMaximum(3) - self.pressure.setSingleStep(0.1) + def _prefill_abs_form(self, state: dict): + idx = self.abs_selector.findText(state.get("action", "")) + if idx >= 0: + self.abs_selector.setCurrentIndex(idx) - form.addRow("Ionization chamber", self.ic_selector) - form.addRow("Gas 1", self.gas1) - form.addRow("Concentration 1", self.conc1) - form.addRow("Gas 2", self.gas2) - form.addRow("Concentration 2", self.conc2) - form.addRow("Pressure", self.pressure) + # ------------------------------------------------------------------ # + # Ionization chamber + # ------------------------------------------------------------------ # + def _build_ic_page(self): + if not all(key in self._dev for key in ("ic0", "ic1", "ic2")): + return + page = QWidget() + form = QFormLayout(page) + form.setFieldGrowthPolicy(QFormLayout.FieldGrowthPolicy.FieldsStayAtSizeHint) - button_layout = QHBoxLayout() - generate_cmd = MyButton("Generate command", "default") - button_layout.addWidget(generate_cmd) - button_layout.addStretch(1) - layout.addLayout(button_layout) + self.ic_selector = QComboBox() + self.ic_selector.addItems(["IC0", "IC1", "IC2"]) - self.conc1.valueChanged.connect(self._equalize_ic_conc) - self.conc2.valueChanged.connect(self._equalize_ic_conc) - generate_cmd.clicked.connect(self._generate_ic_command) + gases = ["He", "N2", "Ar", "Kr"] + self.gas1 = QComboBox() + self.gas2 = QComboBox() + self.gas1.addItems(gases) + self.gas2.addItems(gases) - return ic_group - return None + self.conc1 = QDoubleSpinBox() + self.conc2 = QDoubleSpinBox() + for conc in (self.conc1, self.conc2): + conc.setDecimals(0) + conc.setSuffix(" %") + conc.setMinimum(0) + conc.setMaximum(100) + conc.setSingleStep(1) + + self.pressure = QDoubleSpinBox() + self.pressure.setDecimals(3) + self.pressure.setSuffix(" bar abs") + self.pressure.setMinimum(1) + self.pressure.setMaximum(3) + self.pressure.setSingleStep(0.1) + + form.addRow("Ionization chamber", self.ic_selector) + form.addRow("Gas 1", self.gas1) + form.addRow("Concentration 1", self.conc1) + form.addRow("Gas 2", self.gas2) + form.addRow("Concentration 2", self.conc2) + form.addRow("Pressure", self.pressure) + + self.conc1.valueChanged.connect(self._equalize_ic_conc) + self.conc2.valueChanged.connect(self._equalize_ic_conc) + + self._add_page("ic", "Ionization Chamber", page) def _equalize_ic_conc(self, new_val): if self.conc1.value() == new_val: # conc1 was changed @@ -428,66 +454,98 @@ class ScheduleItemDialog(QDialog): else: self.conc1.setValue(100 - new_val) - def _generate_ic_command(self): - if self.conc1.value() + self.conc2.value() != 100: - return - match self.ic_selector.currentText(): - case "IC0": - ic = "ic0" - case "IC1": - ic = "ic1" - case "IC2": - ic = "ic2" + def _collect_ic_result(self) -> dict: + total = self.conc1.value() + self.conc2.value() + if total != 100: + raise ValueError(f"Gas concentrations must add up to 100% (currently {total:.0f}%).") + chamber = self.ic_selector.currentText() + ic = {"IC0": "ic0", "IC1": "ic1", "IC2": "ic2"}[chamber] cmd = ( f"dev.{ic}.fill(" - + f"gas1='{self.gas1.currentText()}', conc1={self.conc1.value()}, " - + f"gas2='{self.gas2.currentText()}', conc2={self.conc2.value()}, " - + f"pressure={self.pressure.value()}, wait=True)" + f"gas1='{self.gas1.currentText()}', conc1={self.conc1.value()}, " + f"gas2='{self.gas2.currentText()}', conc2={self.conc2.value()}, " + f"pressure={self.pressure.value()}, wait=True)" ) - self.custom_edit.setText(cmd) + return { + "command": cmd, + "kind": "custom", + "form_state": { + "text": cmd, + "source": "ic", + "chamber": chamber, + "gas1": self.gas1.currentText(), + "conc1": self.conc1.value(), + "gas2": self.gas2.currentText(), + "conc2": self.conc2.value(), + "pressure": self.pressure.value(), + }, + } - def _create_reffoil_form(self): - if "reffoilchanger" in self._dev: - reffoil_group = QGroupBox("Reference foil changer") - layout = QVBoxLayout(reffoil_group) - form = QFormLayout() - layout.addLayout(form) - form.setFieldGrowthPolicy(QFormLayout.FieldGrowthPolicy.FieldsStayAtSizeHint) - self.reffoil_selector = QComboBox() - available_foils = self._dev.reffoilchanger.get_all_foils() - self.reffoil_selector.addItems(available_foils) + def _prefill_ic_form(self, state: dict): + idx = self.ic_selector.findText(state.get("chamber", "")) + if idx >= 0: + self.ic_selector.setCurrentIndex(idx) + idx = self.gas1.findText(state.get("gas1", "")) + if idx >= 0: + self.gas1.setCurrentIndex(idx) + idx = self.gas2.findText(state.get("gas2", "")) + if idx >= 0: + self.gas2.setCurrentIndex(idx) + # Set conc2 last: conc1's valueChanged->_equalize_ic_conc would + # otherwise immediately overwrite it. + self.conc1.setValue(float(state.get("conc1", 0))) + self.conc2.setValue(float(state.get("conc2", 0))) + self.pressure.setValue(float(state.get("pressure", self.pressure.minimum()))) - form.addRow("Reference foil", self.reffoil_selector) + # ------------------------------------------------------------------ # + # Reference foil changer + # ------------------------------------------------------------------ # + def _build_reffoil_page(self): + if "reffoilchanger" not in self._dev: + return + page = QWidget() + form = QFormLayout(page) + form.setFieldGrowthPolicy(QFormLayout.FieldGrowthPolicy.FieldsStayAtSizeHint) - button_layout = QHBoxLayout() - generate_cmd = MyButton("Generate command", "default") - button_layout.addWidget(generate_cmd) - button_layout.addStretch(1) - layout.addLayout(button_layout) + self.reffoil_selector = QComboBox() + self.reffoil_selector.addItems(self._dev.reffoilchanger.get_all_foils()) + form.addRow("Reference foil", self.reffoil_selector) - generate_cmd.clicked.connect(self._generate_reffoil_command) + self._add_page("reffoil", "Reference foil changer", page) - return reffoil_group - return None + def _collect_reffoil_result(self) -> dict: + foil = self.reffoil_selector.currentText() + if not foil: + raise ValueError("Select a reference foil.") + cmd = f"dev.reffoilchanger.insert(ref='{foil}', wait=True)" + return { + "command": cmd, + "kind": "custom", + "form_state": {"text": cmd, "source": "reffoil", "foil": foil}, + } - def _generate_reffoil_command(self): - cmd = f"dev.reffoilchanger.insert(ref='{self.reffoil_selector.currentText()}', wait=True)" - self.custom_edit.setText(cmd) + def _prefill_reffoil_form(self, state: dict): + idx = self.reffoil_selector.findText(state.get("foil", "")) + if idx >= 0: + self.reffoil_selector.setCurrentIndex(idx) - def _create_auto_gain_form(self): - auto_gain_group = QGroupBox("Auto Gain") - layout = QVBoxLayout(auto_gain_group) + # ------------------------------------------------------------------ # + # Auto-Gain + # ------------------------------------------------------------------ # + def _build_auto_gain_page(self): + page = QWidget() + layout = QVBoxLayout(page) edge_selector_layout = QHBoxLayout() - edge_selector_label = QLabel("Absorption edge:") + edge_selector_layout.addWidget(QLabel("Absorption edge:")) self.edge_selector_button = MyButton("Choose", "default") self.edge_label = QLabel("No edge selected") - edge_selector_layout.addWidget(edge_selector_label) edge_selector_layout.addWidget(self.edge_selector_button) edge_selector_layout.addWidget(self.edge_label) edge_selector_layout.addStretch() self.edge_element = None self.edge_edge = None + self.edge_energy = None layout.addLayout(edge_selector_layout) @@ -506,49 +564,90 @@ class ScheduleItemDialog(QDialog): self.pips_check = QCheckBox("") form.addRow("PIPS", self.pips_check) - button_layout = QHBoxLayout() - generate_cmd = MyButton("Generate command", "default") - button_layout.addWidget(generate_cmd) - button_layout.addStretch(1) - layout.addLayout(button_layout) - self.edge_selector_button.clicked.connect(self._update_edge) - generate_cmd.clicked.connect(self._generate_auto_gain_command) - return auto_gain_group + self._add_page("auto_gain", "Auto-Gain", page) def _update_edge(self, *_): - match self.beamline: - case "x01da": - dlg = self.EdgeSelector(self) - case "x10da": - dlg = self.EdgeSelector(self) - case _: - dlg = self.EdgeSelector(self) + # NOTE: EdgeSelector is currently the same dialog for every + # beamline; this hook is here so a beamline-specific selector can + # be swapped in later without touching callers. + dlg = self.EdgeSelector(self) if dlg.exec_(): + self.edge_element = dlg.selected_element + self.edge_edge = dlg.selected_edge self.edge_energy = dlg.selected_energy self.edge_label.setText( f"{dlg.selected_element}, {dlg.selected_edge}-edge, {dlg.selected_energy:0.1f} eV" ) - self.edge_element = dlg.selected_element - self.edge_edge = dlg.selected_edge - def _generate_auto_gain_command(self): - if self.edge_edge is None or self.edge_element is None: - return - amplifiers = [] - for amp, name in [ + def _collect_auto_gain_result(self) -> dict: + if self.edge_element is None or self.edge_edge is None: + raise ValueError("Choose an absorption edge first.") + amplifier_checks = [ (self.ic0_check, "ic0"), (self.ic1_check, "ic1"), (self.ic2_check, "ic2"), (self.pips_check, "pips"), - ]: - if amp.isChecked(): - amplifiers.append(name) - if amplifiers == []: - return - cmd = f"auto_gain.start(element={self.edge_element}, edge={self.edge_edge}, {amplifiers}, comp_ring_current=True)" - self.custom_edit.setText(cmd) + ] + amplifiers = [name for check, name in amplifier_checks if check.isChecked()] + if not amplifiers: + raise ValueError("Select at least one amplifier for auto gain.") + cmd = ( + f"auto_gain.start(element={self.edge_element}, edge={self.edge_edge}, " + f"{amplifiers}, comp_ring_current=True)" + ) + return { + "command": cmd, + "kind": "custom", + "form_state": { + "text": cmd, + "source": "auto_gain", + "element": self.edge_element, + "edge": self.edge_edge, + "energy": self.edge_energy, + "ic0": self.ic0_check.isChecked(), + "ic1": self.ic1_check.isChecked(), + "ic2": self.ic2_check.isChecked(), + "pips": self.pips_check.isChecked(), + }, + } + + def _prefill_auto_gain_form(self, state: dict): + element = state.get("element") + edge = state.get("edge") + if element is not None and edge is not None: + self.edge_element = element + self.edge_edge = edge + self.edge_energy = state.get("energy") + if self.edge_energy is not None: + self.edge_label.setText(f"{element}, {edge}-edge, {self.edge_energy:0.1f} eV") + else: + self.edge_label.setText(f"{element}, {edge}-edge") + self.ic0_check.setChecked(bool(state.get("ic0", False))) + self.ic1_check.setChecked(bool(state.get("ic1", False))) + self.ic2_check.setChecked(bool(state.get("ic2", False))) + self.pips_check.setChecked(bool(state.get("pips", False))) + + # ------------------------------------------------------------------ # + # Other + # ------------------------------------------------------------------ # + def _build_other_page(self): + page = QWidget() + layout = QVBoxLayout(page) + layout.addWidget( + QLabel( + "Free-form command, evaluated against `scans` and `dev` - use this for " + "anything the other items don't cover, e.g. an RPC call to another widget." + ) + ) + self.custom_edit = QLineEdit() + # TODO Change to a different placeholder text + self.custom_edit.setPlaceholderText("scans.xas_simple_scan(12000, 14000, 2, 10)") + layout.addWidget(self.custom_edit) + layout.addStretch(1) + + self._add_page("other", "Other", page) def _collect_custom_result(self) -> dict: text = self.custom_edit.text().strip() @@ -562,34 +661,33 @@ class ScheduleItemDialog(QDialog): def _apply_initial(self, initial: dict): kind = initial.get("kind") state = initial.get("form_state") or {} + source = state.get("source") if kind == "scan" and state.get("scan_name"): - self._prefill_scan_tab( + self._prefill_scan_page( state["scan_name"], state.get("args") or [], state.get("kwargs") or {} ) - self.tabs.setCurrentIndex(_TAB_SCAN) - elif kind == "move" and state.get("source") == "digital_twin": - # Reloading a captured config back into DigitalTwin's own input - # fields would need inverting get_assistant_config()'s unit - # conversions and mode branching (fm_focus, mo1_mode, ...) - # field-by-field - not implemented yet. Fall back to showing - # the generated command as read/write text instead of silently - # dropping the captured config; the "Add" (=Ok) button below - # will just resubmit that text unchanged unless it's edited. - self.custom_edit.setText(initial.get("command", "")) - self.tabs.setCurrentIndex(_TAB_OTHER) + self._select_page("scan") + elif kind == "custom" and source == "digital_twin": + self._prefill_digital_twin_page(state) + self._select_page("digital_twin") elif kind == "move" and state.get("device_name"): - idx = self.move_device_combo.findText(state["device_name"]) - if idx >= 0: - self.move_device_combo.setCurrentIndex(idx) - self.move_value_spin.setValue(float(state.get("value", 0.0))) - self.move_relative_check.setChecked(bool(state.get("relative", False))) - self.tabs.setCurrentIndex(_TAB_MOVE) + self._prefill_move_page(state) + self._select_page("move") + elif kind == "custom" and source == "abs" and self._select_page("abs"): + self._prefill_abs_form(state) + elif kind == "custom" and source == "ic" and self._select_page("ic"): + self._prefill_ic_form(state) + elif kind == "custom" and source == "reffoil" and self._select_page("reffoil"): + self._prefill_reffoil_form(state) + elif kind == "custom" and source == "auto_gain" and self._select_page("auto_gain"): + self._prefill_auto_gain_form(state) elif initial.get("command"): - # "custom" kind, or a legacy/unrecognized item - fall back to - # showing the raw command text as-is. + # "custom" kind with no (recognized/available) source, or a + # legacy/unrecognized item - fall back to showing the raw + # command text as-is. self.custom_edit.setText(state.get("text", initial["command"])) - self.tabs.setCurrentIndex(_TAB_OTHER) + self._select_page("other") else: logger.warning(f"Unknown kind: {kind}") @@ -597,20 +695,17 @@ class ScheduleItemDialog(QDialog): try: result = self._collect_result() except ValueError as exc: - QMessageBox.warning(self, "Missing input", str(exc)) + QMessageBox.warning(self, "Can't build command", str(exc)) return self._result = result self.accept() def _collect_result(self) -> dict: - current = self.tabs.currentIndex() - if current == _TAB_SCAN: - return self._collect_scan_result() - if current == _TAB_MOVE: - return self._collect_move_result() - if current == _TAB_DIGITAL_TWIN: - return self._collect_digital_twin_result() - return self._collect_custom_result() + key = self.item_combo.currentData() + handler = self._collectors.get(key) + if handler is None: + raise ValueError(f"Unknown item type: {key!r}") + return handler() def result(self) -> dict: """Valid after `exec_()` returns `QDialog.Accepted`."""