wip
CI for debye_bec / test (pull_request) Successful in 54s
CI for debye_bec / test (push) Successful in 56s

This commit is contained in:
2026-09-09 11:32:05 +02:00
parent 578be60a2d
commit 7082b9ff7b
3 changed files with 484 additions and 315 deletions
@@ -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:
"""
@@ -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()
@@ -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`."""