feat(devices_views): let users change device settings and see readings in Devices, move Config next to Admin

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016TCZEjCSXwMdY9sTEqPQWx
This commit is contained in:
Claude committed 2026-10-09 22:04:52 +00:00
1 parent b10639e345
commit 64594b6572
8 files changed
+579 -22

No files matched your search

+6 -1
View File
@@ -178,8 +178,13 @@ class BECMainApp(BECMainWindow):
self.devices_view.browser.open_in_workspace.connect(self._open_device_in_workspace)
self.device_config_view = ConfigView(parent=self)
self.add_view(icon="memory", title="Devices", widget=self.devices_view, mini_text="Devices")
# Adding and configuring devices is beamline staff work: it sits with Admin View.
self.add_view(
icon="tune", title="Device Config", widget=self.device_config_view, mini_text="Config"
icon="tune",
title="Device Config",
widget=self.device_config_view,
mini_text="Config",
from_top=False,
)
def _open_device_in_workspace(self, widget_name: str, device: str):
@@ -29,6 +29,8 @@ from bec_lib.device import ReadoutPriority
from bec_lib.endpoints import MessageEndpoints
from bec_lib.messages import DeviceMessage, ScanStatusMessage
from bec_widgets.applications.views.devices_views.devices_core import device_kind
DEMO_CONFIG = Path(__file__).resolve().parents[5] / "bec" / "bec_lib" / "bec_lib" / "configs"
@@ -41,6 +43,85 @@ def _demo_config_path() -> Path:
return DEMO_CONFIG / "demo_config.yaml"
class _DemoSignal:
"""Settable demo signal: ``set`` publishes the new value as the device configuration."""
def __init__(self, connector, device: str, obj: str, values: dict):
self.connector = connector
self.device = device
self.obj = obj
self.values = values
def set(self, value):
"""Store and publish ``value``; negative values are refused like a soft limit would."""
if isinstance(value, (int, float)) and value < 0:
raise ValueError(f"{self.obj} must be positive, got {value}")
self.values[self.obj] = {"value": value}
self.connector.set_and_publish(
MessageEndpoints.device_read_configuration(self.device),
DeviceMessage(signals=dict(self.values), metadata={}),
)
def _signal(kind: str, obj: str, cls: str = "Signal", units: str = "", doc: str = "") -> dict:
return {
"kind_str": kind,
"component_name": obj,
"obj_name": obj,
"signal_class": cls,
"doc": doc,
"describe": {"units": units} if units else {},
}
def _demo_signals(client, name: str, dev, kind: str) -> None:
"""Give demo devices the readings and settings real ones report."""
connector = client.connector
if kind == "positioner":
settings = {
"velocity": (2.0, "mm/s", "Speed of a move"),
"acceleration": (0.5, "s", "Time to reach full speed"),
"tolerance": (0.01, "mm", "How close counts as arrived"),
}
signals = {
"readback": _signal("hinted", name),
"setpoint": _signal("normal", f"{name}_setpoint"),
}
readonly = {}
elif kind == "detector":
settings = {
"exp_time": (0.1, "s", "Exposure time per frame"),
"frames_per_trigger": (1, "", "Frames taken for each trigger"),
}
readonly = {"image_shape": ([2048, 2048], "SignalRO")}
signals = {
"image": _signal("hinted", f"{name}_image", "PreviewSignal"),
"progress": _signal("normal", f"{name}_progress", "ProgressSignal"),
}
else:
return
values = {}
for key, (value, units, doc) in settings.items():
obj = f"{name}_{key}"
signals[key] = _signal("config", obj, "Signal", units, doc)
values[obj] = {"value": value}
setattr(dev, key, _DemoSignal(connector, name, obj, values))
for key, (value, cls) in readonly.items():
obj = f"{name}_{key}"
signals[key] = _signal("config", obj, cls)
values[obj] = {"value": value}
# Like real devices, the container reports no write access of its own.
dev._info = { # pylint: disable=protected-access
**(dev._info or {}),
"write_access": None,
"signals": signals,
}
connector.set_and_publish(
MessageEndpoints.device_read_configuration(name),
DeviceMessage(signals=dict(values), metadata={}),
)
def fake_session():
"""Return a BEC dispatcher whose client holds the demo config devices on fakeredis."""
from bec_widgets.tests.fake_devices import ( # pylint: disable=import-outside-toplevel
@@ -71,10 +152,14 @@ def fake_session():
dev._config = {"name": name, **cfg} # pylint: disable=protected-access
devices[name] = dev
value = round(math.sin(i) * 5, 3) if isinstance(dev, FakePositioner) else 1000 + i * 7.5
signals = {name: {"value": value}}
if isinstance(dev, FakePositioner):
signals[f"{name}_setpoint"] = {"value": value}
client.connector.set_and_publish(
MessageEndpoints.device_readback(name),
DeviceMessage(signals={name: {"value": value}}, metadata={}),
MessageEndpoints.device_readback(name), DeviceMessage(signals=signals, metadata={})
)
kind = device_kind(dev)
_demo_signals(client, name, dev, kind)
return dispatcher
@@ -152,6 +237,9 @@ def screenshots(app, client, out: Path, theme: str) -> None:
widget.resize(1500, 900)
widget.show()
_settle(app, 900)
if view != "legacy" and page == "devices":
widget.browser.set_signal("velocity", "3")
_settle(app, 500)
widget.grab().save(str(out / f"{theme}_{view}_{page}.png"))
if view != "legacy" and page == "config":
sheet = widget.open_review()
@@ -171,6 +259,13 @@ def screenshots(app, client, out: Path, theme: str) -> None:
)
_settle(app, 600)
widget.grab().save(str(out / f"{theme}_{view}_devices_monitor.png"))
widget.browser.set_kind("detector")
widget.browser.select("eiger")
_settle(app, 400)
widget.browser.set_signal("exp_time", "0,5")
widget.browser.set_signal("frames_per_trigger", "-2")
_settle(app, 600)
widget.grab().save(str(out / f"{theme}_{view}_devices_detector.png"))
widget.close()
widget.deleteLater()
_settle(app, 200)
@@ -957,6 +957,42 @@ def device_kind(device) -> str:
return "monitor"
READ_ONLY_CLASS_HINTS = ("RO", "Computed", "Async", "Preview", "Progress", "FileEvent", "Dynamic")
MAX_READINGS = 30
def signal_entries(device) -> list[dict]:
"""Signals a user can see on a device: ``readings`` (hinted/normal) and ``settings`` (config).
A setting is offered for editing unless the device reports no write access or its signal
class is read-only (``EpicsSignalRO``, computed and async signals).
"""
info = getattr(device, "_info", {}) or {}
device_writable = info.get("write_access") is not False
out = []
for key, sig in (info.get("signals") or {}).items():
kind = str(sig.get("kind_str", ""))
if kind not in ("hinted", "normal", "config"):
continue
cls = str(sig.get("signal_class", ""))
describe = sig.get("describe") or {}
precision = describe.get("precision")
out.append(
{
"key": key,
"obj": str(sig.get("obj_name") or key),
"section": "setting" if kind == "config" else "reading",
"settable": kind == "config"
and device_writable
and not any(h in cls for h in READ_ONLY_CLASS_HINTS),
"doc": str(sig.get("doc") or ""),
"units": str(describe.get("units") or describe.get("egu") or ""),
"precision": precision if isinstance(precision, int) else None,
}
)
return out
def readback_value(name: str, content: dict | None) -> Any:
"""The main value of a device readback message (its own signal, else the first one)."""
signals = (content or {}).get("signals") or {}
@@ -1016,6 +1052,10 @@ class DeviceBrowser(QObject):
self._workers: list[QRunnable] = []
self._visible: list[dict] = []
self._total = 0
self.contents: dict[str, dict] = {} # device -> latest readback signals
self.config_values: dict[str, dict] = {} # device -> latest read_configuration signals
self.set_status: dict[tuple[str, str], tuple[str, str]] = {} # (device, key) -> tone, text
self._config_subscribed: str | None = None
# ------------------------------------------------------------------ data
def _devices(self) -> dict:
@@ -1086,6 +1126,10 @@ class DeviceBrowser(QObject):
out.sort(key=lambda r: (pref.get(r["name"], len(pref)), r["name"]))
self._total = len(out)
self._visible = out[:MAX_ROWS]
if self.selected is None and self._visible:
self.selected = self._visible[0]["name"]
self._visible[0]["selected"] = True
self._follow_config(self.selected)
self._resubscribe([r["name"] for r in self._visible])
return self._visible
@@ -1102,11 +1146,7 @@ class DeviceBrowser(QObject):
return {"name": ""}
info = self._info(name, device)
kind = device_kind(device)
facts = [
("Class", info["deviceClass"]),
("Readout", info["readout"]),
("Tags", ", ".join(info["tags"]) or "—"),
]
facts = [("Type", info["deviceClass"]), ("Tags", ", ".join(info["tags"]) or "—")]
if not info["enabled"]:
facts.append(("Enabled", "No — disabled in the session"))
return {
@@ -1122,6 +1162,77 @@ class DeviceBrowser(QObject):
}.get(kind, f"Plot {name} in the workspace"),
}
def signal_rows(self) -> list[dict]:
"""Readings and settings of the selected device with their live values.
Motors leave out readings because the positioner card already shows them.
"""
name = self.selected
device = self._devices().get(name) if name else None
if device is None:
return []
positioner = device_kind(device) == "positioner"
live = self.contents.get(name, {})
conf = self.config_values.get(name, {})
rows, readings = [], 0
for entry in signal_entries(device):
if entry["section"] == "reading":
if positioner or readings >= MAX_READINGS:
continue
readings += 1
source = conf if entry["section"] == "setting" else live
value = (source.get(entry["obj"]) or {}).get("value")
tone, status = self.set_status.get((name, entry["key"]), ("", ""))
rows.append(
{
"key": entry["key"],
"section": entry["section"],
"settable": entry["settable"],
"doc": entry["doc"],
"valueText": format_value(value, entry["precision"], entry["units"]),
"statusText": status,
"statusTone": tone,
}
)
return rows
def set_signal(self, key: str, text: str) -> None:
"""Set a setting of the selected device to the typed value, off the GUI thread."""
name = self.selected
device = self._devices().get(name) if name else None
if device is None or not text.strip():
return
entry = next((e for e in signal_entries(device) if e["key"] == key), None)
if entry is None or not entry["settable"]:
return
value = parse_value(text)
target = device
for part in key.split("."):
target = getattr(target, part)
self.set_status[(name, key)] = (
"busy",
f"Setting to {format_value(value, entry['precision'], entry['units'])}…",
)
self._dirty = True
worker = CallWorker(lambda: target.set(value))
worker.setAutoDelete(False)
shown = format_value(value, entry["precision"], entry["units"])
worker.signals.done.connect(
lambda result, n=name, k=key, t=shown: self._on_set(n, k, t, result)
)
self._workers.append(worker)
QThreadPool.globalInstance().start(worker)
def _on_set(self, name: str, key: str, text: str, result) -> None:
self._workers = [w for w in self._workers if w.signals is not self.sender()]
if isinstance(result, Exception):
message = str(result).strip().splitlines()[-1] if str(result).strip() else ""
self.set_status[(name, key)] = ("err", f"Not set: {message or type(result).__name__}")
else:
self.set_status[(name, key)] = ("ok", f"Set to {text}")
self._dirty = True
self._flush_values()
def spark_points(self) -> list[float]:
"""Recent numeric values of the selected monitor (oldest first)."""
return [float(v) for v in self.spark]
@@ -1144,6 +1255,7 @@ class DeviceBrowser(QObject):
if name == self.selected:
return
self.selected = name
self._follow_config(name)
self.spark.clear()
value = self.values.get(name)
if isinstance(value, (int, float)) and not isinstance(value, bool):
@@ -1162,6 +1274,57 @@ class DeviceBrowser(QObject):
self.open_in_workspace.emit(widget, self.selected)
# ------------------------------------------------------------------ live values
def _follow_config(self, name: str | None) -> None:
"""Follow the configuration values (settings) of the selected device only."""
if self.dispatcher is None or name == self._config_subscribed:
return
if self._config_subscribed is not None:
self.dispatcher.disconnect_slot(
self.on_config, MessageEndpoints.device_read_configuration(self._config_subscribed)
)
self._config_subscribed = name
if name is None:
return
self.dispatcher.connect_slot(
self.on_config, MessageEndpoints.device_read_configuration(name)
)
connector = self.client.connector
def read():
out = {}
for key, endpoint in (
("live", MessageEndpoints.device_readback(name)),
("conf", MessageEndpoints.device_read_configuration(name)),
):
msg = connector.get(endpoint)
if msg is not None:
out[key] = (msg.content or {}).get("signals") or {}
return name, out
worker = CallWorker(read)
worker.setAutoDelete(False)
worker.signals.done.connect(self._on_selected_initial)
self._workers.append(worker)
QThreadPool.globalInstance().start(worker)
def _on_selected_initial(self, result) -> None:
self._workers = [w for w in self._workers if w.signals is not self.sender()]
if not isinstance(result, tuple):
return
name, out = result
if "live" in out:
self.contents.setdefault(name, out["live"])
if "conf" in out:
self.config_values.setdefault(name, out["conf"])
self._dirty = True
def on_config(self, content: dict, metadata: dict) -> None:
"""Store the configuration values of the selected device."""
name = (metadata or {}).get("device") or self._config_subscribed
if name is not None:
self.config_values[name] = (content or {}).get("signals") or {}
self._dirty = True
def _resubscribe(self, names: list[str]) -> None:
if self.dispatcher is None or names == self._subscribed:
return
@@ -1215,6 +1378,7 @@ class DeviceBrowser(QObject):
name = next((n for n in self._subscribed if first.startswith(f"{n}_")), first)
if name is None:
return
self.contents[name] = signals
value = readback_value(name, content)
self.values[name] = value
if name == self.selected and isinstance(value, (int, float)):
@@ -1235,3 +1399,4 @@ class DeviceBrowser(QObject):
self.on_readback, [MessageEndpoints.device_readback(n) for n in self._subscribed]
)
self._subscribed = []
self._follow_config(None)
@@ -49,6 +49,7 @@ CONFIG_ROLES = [
"selected",
]
DEVICE_ROLES = ["name", "what", "kind", "valueText", "selected"]
SIGNAL_ROLES = ["key", "section", "settable", "doc", "valueText", "statusText", "statusTone"]
# ----------------------------------------------------------------------------------- Devices view
@@ -79,6 +80,9 @@ class DevicesBackend(QObject):
self.browser = browser
self._rows = DictListModel(DEVICE_ROLES, self)
self._detail: dict = {}
self._settings = DictListModel(SIGNAL_ROLES, self)
self._readings = DictListModel(SIGNAL_ROLES, self)
self._signal_device: str | None = None
self._motor = None
browser.changed.connect(self.refresh)
browser.values_changed.connect(self._refresh_values)
@@ -92,14 +96,26 @@ class DevicesBackend(QObject):
rows = self.browser.rows()
self._rows.set_items(rows)
self._detail = self.browser.detail()
self._refresh_signals()
self.changed.emit()
self.values_changed.emit()
def _refresh_values(self) -> None:
self._rows.set_items(self.browser.rows())
self._detail = self.browser.detail()
self._refresh_signals()
self.values_changed.emit()
def _refresh_signals(self) -> None:
rows = self.browser.signal_rows()
if self._signal_device != self.browser.selected:
# New device: drop the old delegates so typed text does not carry over.
self._signal_device = self.browser.selected
self._settings.set_items([])
self._readings.set_items([])
self._settings.set_items([r for r in rows if r["section"] == "setting"])
self._readings.set_items([r for r in rows if r["section"] == "reading"])
def set_motor(self, motor) -> None:
"""Positioner box shown for motors."""
self._motor = motor
@@ -122,8 +138,14 @@ class DevicesBackend(QObject):
def openInWorkspace(self) -> None:
self.browser.request_open()
@Slot(str, str)
def setSignal(self, key: str, text: str) -> None:
self.browser.set_signal(key, text)
rows = Property(QObject, lambda self: self._rows, constant=True)
detail = Property("QVariantMap", lambda self: self._detail, notify=values_changed)
settings = Property(QObject, lambda self: self._settings, constant=True)
readings = Property(QObject, lambda self: self._readings, constant=True)
detailName = Property(str, lambda self: self._detail.get("name", ""), notify=changed)
kind = Property(str, lambda self: self.browser.kind, notify=changed)
kinds = Property(
@@ -173,6 +173,7 @@ def view_qss(t: ThemeTokens) -> str:
QLabel[role="big"] {{ font-size: 28px; font-weight: 600; font-family: {MONO}; }}
QLabel[role="was"] {{ color: {t.warning.name()}; font-size: 11px; }}
QLabel[role="err"] {{ color: {t.danger.name()}; font-size: 12px; }}
QLabel[role="ok"] {{ color: {t.success.name()}; font-size: 12px; }}
QLabel[role="link"] {{ color: {t.primary.name()}; font-size: 12px; }}
QToolButton[tb="ghost"] {{ color: {t.fg.name()}; background: transparent; border: none;
border-radius: 6px; padding: 4px 8px; font-size: 12px; }}
@@ -715,8 +716,8 @@ class DevicesViewQWidget(BECWidget, QWidget):
self.kinds.changed.connect(self.browser.set_kind)
row.addWidget(self.kinds)
row.addStretch(1)
lock = Chip("Read-only configuration", "neutral", "lock")
lock.setToolTip("Editing the device session is in Config (staff)")
lock = Chip("Device setup: staff only", "neutral", "lock")
lock.setToolTip("Adding devices and changing the session is in Device Config")
row.addWidget(lock)
root.addWidget(bar)
@@ -772,14 +773,27 @@ class DevicesViewQWidget(BECWidget, QWidget):
d.addWidget(self.head)
self.motor_holder = QVBoxLayout()
d.addLayout(self.motor_holder)
self.action = TextButton("", "neutral", "add")
self.action.clicked.connect(self.browser.request_open)
d.addWidget(self.action, 0, Qt.AlignmentFlag.AlignLeft)
self._signal_widgets: dict[str, dict] = {}
self._signal_device: str | None = None
self.settings_title = label("Settings", "upper")
self.settings_grid = self._signal_grid()
self.readings_title = label("Readings", "upper")
self.readings_grid = self._signal_grid()
d.addSpacing(6)
d.addWidget(self.settings_title)
d.addLayout(self.settings_grid)
d.addSpacing(6)
d.addWidget(self.readings_title)
d.addLayout(self.readings_grid)
d.addSpacing(6)
self.facts = QGridLayout()
self.facts.setHorizontalSpacing(12)
self.facts.setVerticalSpacing(3)
self.facts.setColumnStretch(1, 1)
d.addLayout(self.facts)
self.action = TextButton("", "neutral", "add")
self.action.clicked.connect(self.browser.request_open)
d.addWidget(self.action, 0, Qt.AlignmentFlag.AlignLeft)
self.empty = label("Select a device to see it here.", "muted")
d.addWidget(self.empty)
d.addStretch(1)
@@ -797,6 +811,78 @@ class DevicesViewQWidget(BECWidget, QWidget):
if current.isValid():
self.browser.select(self.model.rows[current.row()]["name"])
@staticmethod
def _signal_grid() -> QGridLayout:
grid = QGridLayout()
grid.setHorizontalSpacing(10)
grid.setVerticalSpacing(4)
grid.setColumnStretch(1, 1)
return grid
def _build_signals(self) -> None:
"""Lay out the settings and readings of the selected device (on selection change)."""
clear_layout(self.settings_grid)
clear_layout(self.readings_grid)
self._signal_widgets = {}
self._signal_device = self.browser.selected
counts = {"setting": 0, "reading": 0}
next_row = {"setting": 0, "reading": 0}
for row in self.browser.signal_rows():
grid = self.settings_grid if row["section"] == "setting" else self.readings_grid
r = next_row[row["section"]]
next_row[row["section"]] += 2 if row["settable"] else 1
counts[row["section"]] += 1
name = label(row["key"], "mono")
name.setToolTip(row["doc"] or row["key"])
value = label(row["valueText"], "mono")
value.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
grid.addWidget(name, r, 0)
grid.addWidget(value, r, 1)
widgets = {"value": value}
if row["settable"]:
edit = QLineEdit()
edit.setPlaceholderText("New value")
edit.setAccessibleName(f"New value for {row['key']}")
edit.setFixedWidth(110)
button = TextButton("Set", "neutral")
key = row["key"]
submit = lambda _=False, k=key, e=edit: self.browser.set_signal(k, e.text())
edit.returnPressed.connect(submit)
button.clicked.connect(submit)
grid.addWidget(edit, r, 2)
grid.addWidget(button, r, 3)
status = label("", "faint")
status.setWordWrap(True)
status.hide()
grid.addWidget(status, r + 1, 0, 1, 4)
widgets.update(edit=edit, status=status)
self._signal_widgets[row["key"]] = widgets
self.settings_title.setVisible(counts["setting"] > 0)
self.readings_title.setVisible(counts["reading"] > 0)
def _update_signals(self) -> None:
"""Refresh values and set results in place, keeping typed text and focus."""
if self._signal_device != self.browser.selected:
self._build_signals()
return
for row in self.browser.signal_rows():
widgets = self._signal_widgets.get(row["key"])
if widgets is None:
continue
widgets["value"].setText(row["valueText"])
status = widgets.get("status")
if status is None:
continue
role = {"ok": "ok", "err": "err"}.get(row["statusTone"], "faint")
if status.property("role") != role:
status.setProperty("role", role)
status.style().unpolish(status)
status.style().polish(status)
status.setText(row["statusText"])
status.setVisible(bool(row["statusText"]))
if row["statusTone"] == "ok" and widgets["edit"].text():
widgets["edit"].clear()
@SafeSlot()
def _sync(self) -> None:
rows = self.browser.rows()
@@ -816,6 +902,7 @@ class DevicesViewQWidget(BECWidget, QWidget):
detail = self.browser.detail()
if detail.get("name") and detail["kind"] != "positioner":
self.d_value.setText(self._value_html(detail))
self._update_signals()
@staticmethod
def _value_html(detail: dict) -> str:
@@ -835,6 +922,10 @@ class DevicesViewQWidget(BECWidget, QWidget):
clear_layout(self.facts)
if not has:
self._show_motor(None)
clear_layout(self.settings_grid)
clear_layout(self.readings_grid)
self.settings_title.hide()
self.readings_title.hide()
return
kind = detail["kind"]
self._show_motor(detail["name"] if kind == "positioner" else None)
@@ -853,6 +944,7 @@ class DevicesViewQWidget(BECWidget, QWidget):
self.action.setText(detail["actionText"])
self.action._icon_name = {"detector": "image", "monitor": "show_chart"}.get(kind, "add")
self.action.set_variant("primary" if kind == "detector" else "neutral")
self._build_signals()
def _show_motor(self, name: str | None) -> None:
if name is None:
@@ -2,6 +2,7 @@ import QtQuick
import QtQuick.Controls.Basic
import QtQuick.Layouts
import BecUi
import "Tones.js" as Tones
// Devices for everyone: find a device by kind and text, see it live, move motors.
// backend: DevicesBackend.
@@ -45,11 +46,11 @@ Rectangle {
}
Item { Layout.fillWidth: true }
Chip {
text: "Read-only configuration"
text: "Device setup: staff only"
iconName: "lock"
HoverHandler { id: lockHover }
ToolTip.visible: lockHover.hovered
ToolTip.text: "Editing the device session is in Config (staff)"
ToolTip.text: "Adding devices and changing the session is in Device Config"
}
}
}
@@ -211,8 +212,105 @@ Rectangle {
onActiveChanged: if (active) setSource(root.backend.motorSource, { backend: root.backend.motor })
Component.onCompleted: if (active) setSource(root.backend.motorSource, { backend: root.backend.motor })
}
TextButton {
visible: root.backend.detailName !== ""
text: root.d.actionText || ""
iconName: root.d.kind === "detector" ? "image" : root.d.kind === "monitor" ? "show_chart" : "add"
variant: root.d.kind === "detector" ? "primary" : "neutral"
onClicked: root.backend.openInWorkspace()
}
SectionTitle { text: "Settings"; visible: root.backend.settings.count > 0 }
Repeater {
model: root.backend.settings
delegate: ColumnLayout {
id: setting
required property string key
required property bool settable
required property string doc
required property string valueText
required property string statusText
required property string statusTone
Layout.fillWidth: true
spacing: 2
RowLayout {
Layout.fillWidth: true
spacing: 8
Text {
text: setting.key
color: theme.fg
font.family: root.mono
font.pixelSize: 12
Layout.preferredWidth: 130
elide: Text.ElideRight
HoverHandler { id: keyHover }
ToolTip.visible: keyHover.hovered && setting.doc !== ""
ToolTip.text: setting.doc
}
Text {
text: setting.valueText
color: theme.fg
font.family: root.mono
font.pixelSize: 12
Layout.fillWidth: true
elide: Text.ElideRight
}
InputField {
id: newValue
visible: setting.settable
Layout.preferredWidth: 110
placeholderText: "New value"
Accessible.name: "New value for " + setting.key
onAccepted: root.backend.setSignal(setting.key, text)
}
TextButton {
visible: setting.settable
text: "Set"
onClicked: root.backend.setSignal(setting.key, newValue.text)
}
}
Text {
visible: setting.statusText !== ""
text: setting.statusText
color: Tones.color(theme, setting.statusTone)
font.pixelSize: 12
wrapMode: Text.WordWrap
Layout.fillWidth: true
onTextChanged: if (setting.statusTone === "ok") newValue.clear()
}
}
}
SectionTitle { text: "Readings"; visible: root.backend.readings.count > 0 }
Repeater {
model: root.backend.readings
delegate: RowLayout {
id: reading
required property string key
required property string doc
required property string valueText
Layout.fillWidth: true
spacing: 8
Text {
text: reading.key
color: theme.fg
font.family: root.mono
font.pixelSize: 12
Layout.preferredWidth: 130
elide: Text.ElideRight
}
Text {
text: reading.valueText
color: theme.fg
font.family: root.mono
font.pixelSize: 12
Layout.fillWidth: true
elide: Text.ElideRight
}
}
}
GridLayout {
visible: root.backend.detailName !== ""
Layout.topMargin: 10
columns: 2
columnSpacing: 12
rowSpacing: 3
@@ -243,13 +341,6 @@ Rectangle {
}
}
}
TextButton {
visible: root.backend.detailName !== ""
text: root.d.actionText || ""
iconName: root.d.kind === "detector" ? "image" : root.d.kind === "monitor" ? "show_chart" : "add"
variant: root.d.kind === "detector" ? "primary" : "neutral"
onClicked: root.backend.openInWorkspace()
}
}
}
}
@@ -0,0 +1,12 @@
import QtQuick
import QtQuick.Layouts
// Small upper-case heading for a group in the detail panel.
Text {
color: theme.fgSubtle
font.pixelSize: 11
font.weight: Font.Bold
font.capitalization: Font.AllUppercase
font.letterSpacing: 0.4
Layout.topMargin: 10
}
+75
View File
@@ -5,6 +5,7 @@ import copy
from unittest import mock
import pytest
from bec_lib.endpoints import MessageEndpoints
from bec_widgets.applications.views.devices_views import devices_core
from bec_widgets.applications.views.devices_views.devices_core import (
@@ -17,6 +18,7 @@ from bec_widgets.applications.views.devices_views.devices_core import (
format_value,
parse_value,
plan_requests,
signal_entries,
)
from bec_widgets.applications.views.devices_views.devices_qml import (
DeviceConfigViewQML,
@@ -276,6 +278,59 @@ def test_browser_readback_and_open_request(mocked_client):
browser.cleanup()
def test_signal_entries_split_readings_and_settable_settings(mocked_client, monkeypatch):
samx = mocked_client.device_manager.devices["samx"]
entries = {e["key"]: e for e in signal_entries(samx)}
assert entries["readback"]["section"] == "reading"
assert entries["velocity"]["section"] == "setting"
assert entries["velocity"]["settable"]
info = copy.deepcopy(samx._info)
info["signals"]["velocity"]["signal_class"] = "EpicsSignalRO"
monkeypatch.setattr(samx, "_info", info)
assert not {e["key"]: e for e in signal_entries(samx)}["velocity"]["settable"]
info = copy.deepcopy(info)
info["signals"]["velocity"]["signal_class"] = "Signal"
info["write_access"] = False
monkeypatch.setattr(samx, "_info", info)
assert not {e["key"]: e for e in signal_entries(samx)}["velocity"]["settable"]
def test_browser_shows_settings_values_and_sets_them(mocked_client, sync_pool):
browser = DeviceBrowser(mocked_client)
browser.select("samx")
browser.on_config({"signals": {"samx_velocity": {"value": 2.0}}}, {"device": "samx"})
rows = {r["key"]: r for r in browser.signal_rows()}
assert "readback" not in rows # the positioner card shows it
assert rows["velocity"]["valueText"] == "2"
with mock.patch.object(browser._devices()["samx"], "velocity", create=True) as velocity:
browser.set_signal("velocity", "2,5")
velocity.set.assert_called_once_with(2.5)
row = {r["key"]: r for r in browser.signal_rows()}["velocity"]
assert (row["statusTone"], row["statusText"]) == ("ok", "Set to 2.5")
velocity.set.side_effect = RuntimeError("Velocity above limit 10")
browser.set_signal("velocity", "20")
row = {r["key"]: r for r in browser.signal_rows()}["velocity"]
assert row["statusTone"] == "err"
assert "Velocity above limit 10" in row["statusText"]
browser.cleanup()
def test_browser_follows_config_of_selected_device_only(mocked_client):
dispatcher = mock.MagicMock()
browser = DeviceBrowser(mocked_client, dispatcher)
browser.select("samx")
browser.select("samy")
endpoints = [c.args[1] for c in dispatcher.connect_slot.call_args_list]
assert MessageEndpoints.device_read_configuration("samy") in endpoints
dispatcher.disconnect_slot.assert_any_call(
browser.on_config, MessageEndpoints.device_read_configuration("samx")
)
browser.cleanup()
dispatcher.disconnect_slot.assert_any_call(
browser.on_config, MessageEndpoints.device_read_configuration("samy")
)
# ------------------------------------------------------------------------------------ both views
@pytest.fixture(params=[DeviceConfigViewQWidget, DeviceConfigViewQML])
def config_view(request, qtbot, mocked_client):
@@ -379,3 +434,23 @@ def test_main_app_adds_device_views_when_opted_in(flavor, monkeypatch, qtbot, mo
new.assert_called_once_with("PositionerBox")
new.return_value.set_positioner.assert_called_once_with("samx")
assert app.stack.currentIndex() == app._view_index["Docks"]
def test_devices_view_lets_users_change_a_setting(devices_view, qtbot, sync_pool):
browser = devices_view.browser
browser.on_config({"signals": {"samx_velocity": {"value": 2.0}}}, {"device": "samx"})
with mock.patch.object(browser._devices()["samx"], "velocity", create=True) as velocity:
if isinstance(devices_view, DevicesViewQML):
qtbot.waitUntil(lambda: devices_view.backend.settings.count == 1, timeout=1000)
devices_view.backend.setSignal("velocity", "3")
qtbot.waitUntil(
lambda: devices_view.backend.settings.items[0]["statusText"] == "Set to 3",
timeout=1000,
)
else:
qtbot.waitUntil(lambda: "velocity" in devices_view._signal_widgets, timeout=1000)
widgets = devices_view._signal_widgets["velocity"]
widgets["edit"].setText("3")
widgets["edit"].returnPressed.emit()
qtbot.waitUntil(lambda: widgets["status"].text() == "Set to 3", timeout=1000)
velocity.set.assert_called_once_with(3)