mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-10-29 16:21:27 +01:00
feat(device_input): add searchable device and signal pickers with QWidget and QML popups, shared device snapshot and faster refresh
This commit is contained in:
8 files changed
+2639
No files matched your search
@@ -0,0 +1,7 @@
|
||||
"""Searchable device and signal pickers with a QWidget and a QML popup.
|
||||
|
||||
The pickers keep the public API of :class:`DeviceComboBox` and :class:`SignalComboBox` (they are
|
||||
subclasses) and replace the plain drop-down list with a popup that searches, groups and shows
|
||||
the live value of the highlighted entry. :mod:`picker_model` holds the toolkit-independent
|
||||
model and :mod:`picker_common` the combobox and popup base classes both versions share.
|
||||
"""
|
||||
@@ -0,0 +1,603 @@
|
||||
"""Combobox and popup base classes of the device and signal pickers.
|
||||
|
||||
:class:`DevicePickerBase` and :class:`SignalPickerBase` subclass :class:`DeviceComboBox` and
|
||||
:class:`SignalComboBox`, keep their API and add a faster refresh path and the wiring of the
|
||||
searchable popup. The concrete pickers in :mod:`picker_qwidget` and :mod:`picker_qml` only choose
|
||||
the popup. The toolkit-independent model lives in :mod:`picker_model`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import weakref
|
||||
|
||||
from bec_lib.endpoints import MessageEndpoints
|
||||
from qtpy.QtCore import QRectF, Qt
|
||||
from qtpy.QtGui import QPainter, QPen
|
||||
from qtpy.QtWidgets import QComboBox, QWidget
|
||||
|
||||
from bec_widgets.utils.error_popups import SafeSlot
|
||||
from bec_widgets.utils.filter_io import get_bec_signals_for_classes
|
||||
from bec_widgets.utils.quick.host import ThemeTokens
|
||||
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
|
||||
from bec_widgets.widgets.control.device_input.picker.picker_model import (
|
||||
DEVICE_KIND_LABELS,
|
||||
DeviceSnapshot,
|
||||
PickerController,
|
||||
PickerEntry,
|
||||
device_snapshot,
|
||||
invalidate_device_snapshot,
|
||||
recent_picks,
|
||||
remember_pick,
|
||||
)
|
||||
from bec_widgets.widgets.control.device_input.signal_combobox.signal_combobox import SignalComboBox
|
||||
|
||||
# ---- combobox base classes ----------------------------------------------------------------------
|
||||
|
||||
|
||||
class PickerPopupMixin:
|
||||
"""Popup wiring and the cheaper validity styling shared by both picker comboboxes.
|
||||
|
||||
The class using it must be a :class:`QComboBox` subclass and implement
|
||||
:meth:`_picker_entries`, :meth:`_select_value` and :meth:`_create_popup`.
|
||||
"""
|
||||
|
||||
PICKER_NOUN = "items"
|
||||
RECENT_KIND = "items"
|
||||
|
||||
def _init_picker(self) -> None:
|
||||
self._picker_popup = None
|
||||
self._picker_controller: PickerController | None = None
|
||||
self._picker_text_before = ""
|
||||
self._live_topics: list = []
|
||||
self._live_entry: PickerEntry | None = None
|
||||
self._show_invalid = False
|
||||
self._last_valid_text = self.currentText() if self._is_valid_input else ""
|
||||
self.setSizeAdjustPolicy(QComboBox.SizeAdjustPolicy.AdjustToMinimumContentsLengthWithIcon)
|
||||
self.setMinimumContentsLength(8)
|
||||
line_edit = self.lineEdit()
|
||||
if line_edit is not None:
|
||||
line_edit.textEdited.connect(self._on_field_edited)
|
||||
line_edit.setPlaceholderText(f"Select {self.PICKER_NOUN[:-1]}…")
|
||||
# the validity of the current text may have been styled with the old stylesheet
|
||||
self._update_validity_style(self._is_valid_input)
|
||||
|
||||
# ---- styling -------------------------------------------------------------------------
|
||||
# The field keeps the application style; a per-widget stylesheet costs several
|
||||
# milliseconds per picker to parse and polish, so the error state is painted instead.
|
||||
|
||||
def _update_validity_style(self, is_valid: bool) -> None:
|
||||
if not hasattr(self, "_picker_popup"):
|
||||
# still inside the parent constructor; styled once in _init_picker
|
||||
return
|
||||
if is_valid and not self.is_popup_open():
|
||||
self._last_valid_text = self.currentText()
|
||||
show = bool(self.currentText()) and not is_valid and self.isEnabled()
|
||||
if show != self._show_invalid:
|
||||
self._show_invalid = show
|
||||
self.update()
|
||||
|
||||
@property
|
||||
def shows_invalid(self) -> bool:
|
||||
"""Whether the field is drawn with the error border."""
|
||||
return self._show_invalid
|
||||
|
||||
def paintEvent(self, event): # pylint: disable=invalid-name
|
||||
"""Paint the field and, for invalid text, a danger-coloured border on top."""
|
||||
super().paintEvent(event) # type: ignore[misc]
|
||||
if not self._show_invalid:
|
||||
return
|
||||
painter = QPainter(self)
|
||||
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
|
||||
pen = QPen(ThemeTokens().danger, 2)
|
||||
painter.setPen(pen)
|
||||
painter.setBrush(Qt.BrushStyle.NoBrush)
|
||||
painter.drawRoundedRect(QRectF(self.rect()).adjusted(1, 1, -1, -1), 4, 4)
|
||||
painter.end()
|
||||
|
||||
def apply_theme(self, theme: str):
|
||||
"""Re-style the popup for the new theme."""
|
||||
super().apply_theme(theme) # type: ignore[misc]
|
||||
if self._picker_popup is not None:
|
||||
self._picker_popup.refresh_theme()
|
||||
self.update()
|
||||
|
||||
# ---- popup -------------------------------------------------------------------------------
|
||||
|
||||
@property
|
||||
def picker_controller(self) -> PickerController:
|
||||
"""Controller of the popup, created on first use."""
|
||||
if self._picker_controller is None:
|
||||
self._picker_controller = PickerController(
|
||||
self.PICKER_NOUN, self._read_live_value, self._watch_live_value, self
|
||||
)
|
||||
self._picker_controller.picked.connect(self._on_picked)
|
||||
self._picker_controller.dismissed.connect(self._on_dismissed)
|
||||
return self._picker_controller
|
||||
|
||||
@property
|
||||
def picker_popup(self):
|
||||
"""The popup window, created on first use."""
|
||||
if self._picker_popup is None:
|
||||
self._picker_popup = self._create_popup()
|
||||
return self._picker_popup
|
||||
|
||||
def is_popup_open(self) -> bool:
|
||||
"""Whether the picker popup is showing."""
|
||||
return self._picker_popup is not None and self._picker_popup.isVisible()
|
||||
|
||||
def showPopup(self) -> None: # pylint: disable=invalid-name
|
||||
"""Open the searchable popup instead of the plain drop-down list."""
|
||||
self.open_picker()
|
||||
|
||||
def hidePopup(self) -> None: # pylint: disable=invalid-name
|
||||
"""Close the popup."""
|
||||
if self.is_popup_open():
|
||||
self._picker_popup.hide()
|
||||
super().hidePopup() # type: ignore[misc]
|
||||
|
||||
def open_picker(self, query: str = "") -> None:
|
||||
"""Open the popup, optionally with a search text already entered.
|
||||
|
||||
Args:
|
||||
query(str): Initial search text.
|
||||
"""
|
||||
if not self.isEnabled():
|
||||
return
|
||||
if not self.is_popup_open():
|
||||
# text typed to start a search is not a selection; cancelling reverts to the last
|
||||
# valid one
|
||||
self._picker_text_before = (
|
||||
self._last_valid_text
|
||||
if query and query == self.currentText()
|
||||
else self.currentText()
|
||||
)
|
||||
controller = self.picker_controller
|
||||
controller.open(
|
||||
self._picker_entries(), self._picker_text_before, query, recent_picks(self.RECENT_KIND)
|
||||
)
|
||||
popup = self.picker_popup
|
||||
popup.show_below(self)
|
||||
|
||||
@SafeSlot(str)
|
||||
def _on_field_edited(self, text: str) -> None:
|
||||
if text and not self.is_popup_open():
|
||||
self.open_picker(query=text)
|
||||
|
||||
@SafeSlot(str)
|
||||
def _on_picked(self, value: str) -> None:
|
||||
self._picker_text_before = value
|
||||
if self._picker_popup is not None:
|
||||
self._picker_popup.hide()
|
||||
remember_pick(self.RECENT_KIND, value)
|
||||
self._select_value(value)
|
||||
self.setFocus()
|
||||
|
||||
@SafeSlot()
|
||||
def _on_dismissed(self) -> None:
|
||||
if self._picker_popup is not None:
|
||||
self._picker_popup.hide()
|
||||
self.setFocus()
|
||||
|
||||
def _on_popup_hidden(self) -> None:
|
||||
"""Called by the popup window when it hides for any reason, e.g. a click outside.
|
||||
|
||||
Text typed into the field to start a search is reverted unless an entry was picked.
|
||||
"""
|
||||
if self._picker_controller is not None:
|
||||
self._picker_controller.close()
|
||||
if self.currentText() != self._picker_text_before:
|
||||
self.setCurrentText(self._picker_text_before)
|
||||
|
||||
# ---- live value --------------------------------------------------------------------------
|
||||
|
||||
def _read_live_value(self, entry: PickerEntry) -> dict | None:
|
||||
connector = getattr(self.client, "connector", None)
|
||||
if connector is None or not entry.device:
|
||||
return None
|
||||
for endpoint in (
|
||||
MessageEndpoints.device_readback(entry.device),
|
||||
MessageEndpoints.device_read_configuration(entry.device),
|
||||
):
|
||||
msg = connector.get(endpoint)
|
||||
signals = getattr(msg, "signals", None) or {}
|
||||
reading = _pick_reading(signals, entry)
|
||||
if reading is not None:
|
||||
return reading
|
||||
return None
|
||||
|
||||
def _watch_live_value(self, entry: PickerEntry | None) -> None:
|
||||
if self._live_topics:
|
||||
self.bec_dispatcher.disconnect_slot(self._on_live_message, self._live_topics)
|
||||
self._live_topics = []
|
||||
self._live_entry = entry
|
||||
if entry is None or not entry.device:
|
||||
return
|
||||
self._live_topics = [
|
||||
MessageEndpoints.device_readback(entry.device),
|
||||
MessageEndpoints.device_read_configuration(entry.device),
|
||||
]
|
||||
self.bec_dispatcher.connect_slot(self._on_live_message, self._live_topics, owner=self)
|
||||
|
||||
@SafeSlot(dict, dict)
|
||||
def _on_live_message(self, content: dict, _metadata: dict) -> None:
|
||||
entry = self._live_entry
|
||||
if entry is None or self._picker_controller is None:
|
||||
return
|
||||
reading = _pick_reading(content.get("signals", {}) or {}, entry)
|
||||
if reading is not None:
|
||||
self._picker_controller.update_live_value(entry, reading)
|
||||
|
||||
def cleanup(self):
|
||||
"""Close the popup and drop the live subscription."""
|
||||
if self._picker_controller is not None:
|
||||
self._picker_controller.close()
|
||||
if self._picker_popup is not None:
|
||||
self._picker_popup.hide()
|
||||
self._picker_popup.release()
|
||||
self._picker_popup.deleteLater()
|
||||
self._picker_popup = None
|
||||
super().cleanup() # type: ignore[misc]
|
||||
|
||||
# ---- to implement ------------------------------------------------------------------------
|
||||
|
||||
def _picker_entries(self) -> list[PickerEntry]:
|
||||
raise NotImplementedError
|
||||
|
||||
def _select_value(self, value: str) -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
def _create_popup(self):
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
def _pick_reading(signals: dict, entry: PickerEntry) -> dict | None:
|
||||
if not signals:
|
||||
return None
|
||||
key = entry.signal
|
||||
if not key:
|
||||
key = entry.device if entry.device in signals else next(iter(signals))
|
||||
reading = signals.get(key)
|
||||
return reading if isinstance(reading, dict) else None
|
||||
|
||||
|
||||
class DevicePickerBase(PickerPopupMixin, DeviceComboBox): # pylint: disable=abstract-method
|
||||
""":class:`DeviceComboBox` with the searchable popup and a cheaper refresh.
|
||||
|
||||
Differences to the base class, all invisible to callers:
|
||||
|
||||
* enabling several filters at once (e.g. the five readout priorities in the constructor)
|
||||
refreshes the list once instead of once per filter;
|
||||
* the enabled devices are read from a snapshot shared by all pickers and only re-read when
|
||||
the device configuration changes;
|
||||
* an unchanged device list does not rebuild the items or re-emit the selection signals;
|
||||
* the error state is a dynamic property, not a new stylesheet per validity check.
|
||||
"""
|
||||
|
||||
PLUGIN = False
|
||||
RPC = False
|
||||
PICKER_NOUN = "devices"
|
||||
RECENT_KIND = "devices"
|
||||
|
||||
_batching_filters = False
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.device_config_update.disconnect(self.update_devices_from_filters)
|
||||
self.device_config_update.connect(self._refresh_after_config_update)
|
||||
self._init_picker()
|
||||
|
||||
# ---- faster refresh --------------------------------------------------------------------
|
||||
|
||||
def set_device_filter(self, filter_selection):
|
||||
self._batching_filters = True
|
||||
try:
|
||||
super().set_device_filter(filter_selection)
|
||||
finally:
|
||||
self._batching_filters = False
|
||||
self.update_devices_from_filters()
|
||||
|
||||
def set_readout_priority_filter(self, filter_selection):
|
||||
self._batching_filters = True
|
||||
try:
|
||||
super().set_readout_priority_filter(filter_selection)
|
||||
finally:
|
||||
self._batching_filters = False
|
||||
self.update_devices_from_filters()
|
||||
|
||||
@SafeSlot()
|
||||
def update_devices_from_filters(self):
|
||||
"""Refresh the available device list from the current filters."""
|
||||
self._refresh(force=False)
|
||||
|
||||
def on_device_update(self, action: str, content: dict) -> None:
|
||||
"""Invalidate the shared snapshot, then refresh like :class:`DeviceComboBox`."""
|
||||
invalidate_device_snapshot(self.client)
|
||||
super().on_device_update(action, content)
|
||||
|
||||
@SafeSlot()
|
||||
def _refresh_after_config_update(self):
|
||||
self._refresh(force=False)
|
||||
|
||||
def _refresh(self, force: bool) -> None:
|
||||
if self._batching_filters:
|
||||
return
|
||||
self.config.device_filter = [entry.value for entry in self.device_filter]
|
||||
self.config.readout_filter = [entry.value for entry in self.readout_filter]
|
||||
self.config.signal_class_filter = self.signal_class_filter
|
||||
if not self.apply_filter:
|
||||
return
|
||||
snapshot = device_snapshot(self.client, force=force)
|
||||
records = list(snapshot.records.values())
|
||||
if self.config.signal_class_filter:
|
||||
allowed = self._devices_with_signal_classes(snapshot)
|
||||
records = [record for record in records if record.name in allowed]
|
||||
wanted = set(self.device_filter)
|
||||
readout = self.readout_filter
|
||||
names = [
|
||||
record.name
|
||||
for record in records
|
||||
if wanted <= record.filters and record.readout in readout
|
||||
]
|
||||
if self.include_signals_with_write_access:
|
||||
seen = set(names)
|
||||
names.extend(
|
||||
record.name
|
||||
for record in snapshot.records.values()
|
||||
if record.write_access and record.name not in seen
|
||||
)
|
||||
self.devices = names
|
||||
|
||||
def _devices_with_signal_classes(self, snapshot: DeviceSnapshot) -> set[str]:
|
||||
key = tuple(self.config.signal_class_filter)
|
||||
cached = snapshot.signal_classes.get(key)
|
||||
if cached is None:
|
||||
signals = get_bec_signals_for_classes(client=self.client, signal_class_filter=list(key))
|
||||
cached = snapshot.signal_classes[key] = [name for name, _, _ in signals]
|
||||
return set(cached)
|
||||
|
||||
def _replace_items(self, devices: list[str]):
|
||||
items = [""] + list(devices) if self._set_first_element_as_empty else list(devices)
|
||||
if self.count() == len(items) and all(
|
||||
self.itemText(row) == text for row, text in enumerate(items)
|
||||
):
|
||||
# nothing changed: keep the items, the completer and the emitted state
|
||||
return
|
||||
current = self.currentText()
|
||||
self.blockSignals(True)
|
||||
try:
|
||||
self.clear()
|
||||
self.addItems(items)
|
||||
self.setCurrentText(current)
|
||||
finally:
|
||||
self.blockSignals(False)
|
||||
self._completer_model.setStringList(list(devices))
|
||||
self.check_validity(self.currentText())
|
||||
|
||||
def validate_device(self, device: str | None) -> bool:
|
||||
"""Validate a device against the current filtered device selection."""
|
||||
if not device:
|
||||
return False
|
||||
device_name = self._device_name_from_text(device)
|
||||
if device_name not in self.devices:
|
||||
return False
|
||||
return device_name in device_snapshot(self.client).records
|
||||
|
||||
# ---- popup -------------------------------------------------------------------------------
|
||||
|
||||
def _picker_entries(self) -> list[PickerEntry]:
|
||||
records = device_snapshot(self.client).records
|
||||
entries = []
|
||||
for name in self.devices:
|
||||
record = records.get(name)
|
||||
if record is None:
|
||||
entries.append(PickerEntry(value=name, title=name, device=name))
|
||||
continue
|
||||
group, icon = DEVICE_KIND_LABELS[record.kind]
|
||||
readout = getattr(record.readout, "value", record.readout) or ""
|
||||
details = record.description or ", ".join(record.tags)
|
||||
entries.append(
|
||||
PickerEntry(
|
||||
value=name,
|
||||
title=name,
|
||||
subtitle=record.device_class,
|
||||
group=group,
|
||||
badge=str(readout),
|
||||
icon=icon,
|
||||
device=name,
|
||||
details=details,
|
||||
)
|
||||
)
|
||||
return entries
|
||||
|
||||
def _select_value(self, value: str) -> None:
|
||||
self.setCurrentText(value)
|
||||
|
||||
|
||||
SIGNAL_KIND_ICONS = {"hinted": "star", "normal": "sensors", "config": "tune"}
|
||||
|
||||
|
||||
class SignalPickerBase(PickerPopupMixin, SignalComboBox): # pylint: disable=abstract-method
|
||||
""":class:`SignalComboBox` with the searchable popup and cheaper validity styling.
|
||||
|
||||
Signals of one device are grouped by kind (hinted, normal, config); signals listed by class
|
||||
are grouped by their device.
|
||||
"""
|
||||
|
||||
PLUGIN = False
|
||||
RPC = False
|
||||
PICKER_NOUN = "signals"
|
||||
RECENT_KIND = "signals"
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self._init_picker()
|
||||
|
||||
def _on_device_update(self, action: str, content: dict) -> None:
|
||||
invalidate_device_snapshot(self.client)
|
||||
super()._on_device_update(action, content)
|
||||
|
||||
def _replace_signal_items(self, items=None):
|
||||
combo_items = self._signals if items is None else items
|
||||
display_items = [""] + combo_items if self._set_first_element_as_empty else combo_items
|
||||
current = self.currentText()
|
||||
keep_text = bool(current)
|
||||
self.blockSignals(True)
|
||||
try:
|
||||
self.clear()
|
||||
plain = [entry for entry in display_items if isinstance(entry, str)]
|
||||
if len(plain) == len(display_items):
|
||||
self.addItems(plain)
|
||||
else:
|
||||
for entry in display_items:
|
||||
if isinstance(entry, str):
|
||||
self.addItem(entry)
|
||||
else:
|
||||
self.addItem(*entry)
|
||||
if keep_text:
|
||||
self.setCurrentText(current)
|
||||
finally:
|
||||
self.blockSignals(False)
|
||||
self._completer_model.setStringList(
|
||||
[entry if isinstance(entry, str) else entry[0] for entry in combo_items]
|
||||
)
|
||||
|
||||
def _picker_entries(self) -> list[PickerEntry]:
|
||||
entries: list[PickerEntry] = []
|
||||
if self._signal_class_filter:
|
||||
device_names = sorted(device_snapshot(self.client).records, key=len, reverse=True)
|
||||
for item in self._signals:
|
||||
text, info = (item, {}) if isinstance(item, str) else item
|
||||
obj_name = info.get("obj_name") or text
|
||||
device = self._device or next(
|
||||
(name for name in device_names if obj_name.startswith(name)), ""
|
||||
)
|
||||
describe = info.get("describe", {}).get("signal_info", {}) or {}
|
||||
ndim = describe.get("ndim")
|
||||
subtitle = info.get("signal_class", "")
|
||||
if ndim is not None:
|
||||
subtitle = f"{subtitle} · {ndim}D" if subtitle else f"{ndim}D"
|
||||
entries.append(
|
||||
PickerEntry(
|
||||
value=text,
|
||||
title=text,
|
||||
subtitle=subtitle,
|
||||
group=device or "Signals",
|
||||
badge=info.get("kind_str", ""),
|
||||
icon=SIGNAL_KIND_ICONS.get(info.get("kind_str", ""), "sensors"),
|
||||
device=device,
|
||||
signal=obj_name,
|
||||
details=info.get("storage_name", ""),
|
||||
)
|
||||
)
|
||||
return entries
|
||||
groups = (
|
||||
("Hinted", "hinted", self._hinted_signals),
|
||||
("Normal", "normal", self._normal_signals),
|
||||
("Config", "config", self._config_signals),
|
||||
)
|
||||
for group, kind, signals in groups:
|
||||
for text, info in signals:
|
||||
obj_name = info.get("obj_name") or text
|
||||
component = info.get("component_name") or ""
|
||||
subtitle = obj_name if obj_name != text else component
|
||||
if subtitle == text:
|
||||
subtitle = ""
|
||||
entries.append(
|
||||
PickerEntry(
|
||||
value=text,
|
||||
title=text,
|
||||
subtitle=subtitle,
|
||||
group=group,
|
||||
badge=kind,
|
||||
icon=SIGNAL_KIND_ICONS[kind],
|
||||
device=self._device or "",
|
||||
signal=obj_name,
|
||||
details=info.get("signal_class", "") or info.get("doc", "") or "",
|
||||
)
|
||||
)
|
||||
if not entries and self._signals:
|
||||
# a plain signal device lists itself without signal info
|
||||
for item in self._signals:
|
||||
text = item if isinstance(item, str) else item[0]
|
||||
entries.append(
|
||||
PickerEntry(value=text, title=text, icon="sensors", device=self._device or text)
|
||||
)
|
||||
return entries
|
||||
|
||||
def _select_value(self, value: str) -> None:
|
||||
index = self.findText(value)
|
||||
if index >= 0:
|
||||
self.setCurrentIndex(index)
|
||||
else:
|
||||
self.setCurrentText(value)
|
||||
|
||||
|
||||
class PickerPopupBase(QWidget):
|
||||
"""Top-level popup window shared by the QWidget and QML popups.
|
||||
|
||||
Subclasses fill the window; this class positions it under the field, follows the theme
|
||||
and reports when it hides.
|
||||
"""
|
||||
|
||||
WIDTH = 380
|
||||
MAX_HEIGHT = 440
|
||||
MIN_HEIGHT = 240
|
||||
# search row, separators and footer around the list
|
||||
CHROME_HEIGHT = 136
|
||||
ROW_HEIGHTS = {"header": 28, "empty": 96, "item": 34}
|
||||
|
||||
def __init__(self, picker: PickerPopupMixin):
|
||||
# parented so it is destroyed with the picker; the Popup flag keeps it a separate window
|
||||
super().__init__(picker, Qt.WindowType.Popup | Qt.WindowType.FramelessWindowHint)
|
||||
self._picker = weakref.ref(picker)
|
||||
self.setAttribute(Qt.WidgetAttribute.WA_TranslucentBackground, False)
|
||||
self.setObjectName("pickerPopup")
|
||||
|
||||
@property
|
||||
def controller(self) -> PickerController:
|
||||
"""Controller of the picker this popup belongs to."""
|
||||
return self._picker().picker_controller
|
||||
|
||||
def show_below(self, anchor: QWidget) -> None:
|
||||
"""Show under ``anchor``, flipping above it if there is no room below."""
|
||||
width = max(anchor.width(), self.WIDTH)
|
||||
height = self.preferred_height()
|
||||
origin = anchor.mapToGlobal(anchor.rect().bottomLeft())
|
||||
screen = anchor.screen().availableGeometry() if anchor.screen() is not None else None
|
||||
x, y = origin.x(), origin.y() + 2
|
||||
if screen is not None:
|
||||
if y + height > screen.bottom():
|
||||
y = max(screen.top(), anchor.mapToGlobal(anchor.rect().topLeft()).y() - height - 2)
|
||||
x = max(screen.left(), min(x, screen.right() - width))
|
||||
self.setGeometry(x, y, width, height)
|
||||
if not self.isVisible():
|
||||
self.show()
|
||||
self.activateWindow()
|
||||
self.focus_search()
|
||||
|
||||
def preferred_height(self) -> int:
|
||||
"""Height that fits the current rows, between ``MIN_HEIGHT`` and ``MAX_HEIGHT``."""
|
||||
model = self.controller.model
|
||||
rows = sum(
|
||||
self.ROW_HEIGHTS.get(model.row(row).get("rowType"), 34)
|
||||
for row in range(model.rowCount())
|
||||
)
|
||||
return max(self.MIN_HEIGHT, min(self.MAX_HEIGHT, rows + self.CHROME_HEIGHT + 8))
|
||||
|
||||
def focus_search(self) -> None:
|
||||
"""Move keyboard focus to the search field."""
|
||||
|
||||
def refresh_theme(self) -> None:
|
||||
"""Apply the current theme."""
|
||||
|
||||
def release(self) -> None:
|
||||
"""Release resources before deletion."""
|
||||
|
||||
def hideEvent(self, event): # pylint: disable=invalid-name
|
||||
picker = self._picker()
|
||||
if picker is not None:
|
||||
picker._on_popup_hidden() # pylint: disable=protected-access
|
||||
super().hideEvent(event)
|
||||
|
||||
|
||||
__all__ = ["DevicePickerBase", "PickerPopupBase", "PickerPopupMixin", "SignalPickerBase"]
|
||||
@@ -0,0 +1,685 @@
|
||||
"""Toolkit-independent model of the device and signal pickers.
|
||||
|
||||
* :func:`device_snapshot` reads the enabled devices once per device configuration and shares
|
||||
the result between all pickers of a client, so a refresh no longer walks the device manager
|
||||
for every combobox.
|
||||
* :class:`PickerEntry`, :func:`rank_entry` and :class:`PickerController` hold the searchable,
|
||||
grouped rows, the highlighted row and the details (with the live value) shown in the footer.
|
||||
Both the QWidget and the QML popup are thin views over the controller.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
import time
|
||||
import weakref
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Callable, Sequence
|
||||
|
||||
from bec_lib.device import ComputedSignal, Device, Positioner
|
||||
from bec_lib.device import Signal as BECSignal
|
||||
from bec_lib.logger import bec_logger
|
||||
from qtpy.QtCore import QObject, Qt, QTimer, Signal
|
||||
|
||||
from bec_widgets.utils.quick.host import ThemeTokens
|
||||
from bec_widgets.utils.quick.list_model import DictListModel
|
||||
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import BECDeviceFilter
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
MAX_RECENT = 4
|
||||
|
||||
# ---- shared device snapshot ------------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass
|
||||
class DeviceRecord:
|
||||
"""What the pickers need to know about one enabled device, read once per configuration."""
|
||||
|
||||
name: str
|
||||
device: Any
|
||||
kind: str
|
||||
filters: frozenset
|
||||
readout: Any
|
||||
device_class: str
|
||||
description: str
|
||||
tags: tuple[str, ...]
|
||||
write_access: bool
|
||||
|
||||
|
||||
@dataclass
|
||||
class DeviceSnapshot:
|
||||
"""All enabled devices of a client at one device configuration generation."""
|
||||
|
||||
generation: int
|
||||
fingerprint: tuple = ()
|
||||
records: dict[str, DeviceRecord] = field(default_factory=dict)
|
||||
signal_classes: dict[tuple, list] = field(default_factory=dict)
|
||||
|
||||
|
||||
class _CatalogState: # pylint: disable=too-few-public-methods
|
||||
"""Generation counter bumped by device configuration updates of one device manager.
|
||||
|
||||
The pickers bump it from their device update callbacks. Updates that happen while no picker
|
||||
listens are caught by the fingerprint of the device container (see :func:`device_snapshot`).
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.generation = 0
|
||||
self.snapshot: DeviceSnapshot | None = None
|
||||
|
||||
def bump(self, *_args, **_kwargs) -> None:
|
||||
"""Invalidate the snapshot; called from the BEC callback thread."""
|
||||
self.generation += 1
|
||||
|
||||
|
||||
_CATALOGS: "weakref.WeakKeyDictionary[Any, _CatalogState]" = weakref.WeakKeyDictionary()
|
||||
|
||||
_KIND_ORDER = (
|
||||
(BECDeviceFilter.POSITIONER, "positioner"),
|
||||
(BECDeviceFilter.COMPUTED_SIGNAL, "computed"),
|
||||
(BECDeviceFilter.SIGNAL, "signal"),
|
||||
(BECDeviceFilter.DEVICE, "device"),
|
||||
)
|
||||
_FILTER_CLASSES = {
|
||||
BECDeviceFilter.DEVICE: Device,
|
||||
BECDeviceFilter.POSITIONER: Positioner,
|
||||
BECDeviceFilter.SIGNAL: BECSignal,
|
||||
BECDeviceFilter.COMPUTED_SIGNAL: ComputedSignal,
|
||||
}
|
||||
|
||||
DEVICE_KIND_LABELS = {
|
||||
"positioner": ("Positioners", "open_with"),
|
||||
"computed": ("Computed signals", "functions"),
|
||||
"signal": ("Signals", "sensors"),
|
||||
"device": ("Devices", "memory"),
|
||||
}
|
||||
|
||||
|
||||
def _catalog_state(client) -> _CatalogState | None:
|
||||
device_manager = getattr(client, "device_manager", None)
|
||||
if device_manager is None:
|
||||
return None
|
||||
try:
|
||||
state = _CATALOGS.get(device_manager)
|
||||
except TypeError:
|
||||
return None
|
||||
if state is None:
|
||||
state = _CatalogState()
|
||||
try:
|
||||
_CATALOGS[device_manager] = state
|
||||
except TypeError:
|
||||
return None
|
||||
return state
|
||||
|
||||
|
||||
def _record_for(device) -> DeviceRecord:
|
||||
filters = frozenset(entry for entry, cls in _FILTER_CLASSES.items() if isinstance(device, cls))
|
||||
kind = next((label for entry, label in _KIND_ORDER if entry in filters), "device")
|
||||
config = getattr(device, "_config", None) or {}
|
||||
info = getattr(device, "_info", None) or {}
|
||||
device_class = str(config.get("deviceClass") or "")
|
||||
tags = config.get("deviceTags") or ()
|
||||
return DeviceRecord(
|
||||
name=device.name,
|
||||
device=device,
|
||||
kind=kind,
|
||||
filters=filters,
|
||||
readout=getattr(device, "readout_priority", None),
|
||||
device_class=device_class.rsplit(".", 1)[-1],
|
||||
description=str(config.get("description") or ""),
|
||||
tags=tuple(sorted(str(tag) for tag in tags)),
|
||||
write_access=bool(info.get("write_access")),
|
||||
)
|
||||
|
||||
|
||||
def _fingerprint(devices) -> tuple:
|
||||
"""Cheap identity of a device container: changes when devices are added, removed or reloaded."""
|
||||
if not isinstance(devices, dict):
|
||||
return ()
|
||||
values = dict.values(devices)
|
||||
return (len(devices), hash(tuple(id(device) for device in values)))
|
||||
|
||||
|
||||
def device_snapshot(client, force: bool = False) -> DeviceSnapshot:
|
||||
"""Return the enabled devices of ``client``, shared by all pickers until the config changes.
|
||||
|
||||
The snapshot is rebuilt when a picker saw a device configuration update, when devices were
|
||||
added, removed or reloaded (the device objects change), or when ``force`` is set.
|
||||
|
||||
Args:
|
||||
client: BEC client.
|
||||
force(bool): Re-read the device manager even if nothing seems to have changed.
|
||||
|
||||
Returns:
|
||||
DeviceSnapshot: Records of all enabled devices in device manager order.
|
||||
"""
|
||||
devices = getattr(getattr(client, "device_manager", None), "devices", None)
|
||||
fingerprint = _fingerprint(devices)
|
||||
state = _catalog_state(client)
|
||||
if state is not None and not force:
|
||||
snapshot = state.snapshot
|
||||
if (
|
||||
snapshot is not None
|
||||
and snapshot.generation == state.generation
|
||||
and snapshot.fingerprint == fingerprint
|
||||
):
|
||||
return snapshot
|
||||
generation = state.generation if state is not None else -1
|
||||
snapshot = DeviceSnapshot(generation=generation, fingerprint=fingerprint)
|
||||
for device in getattr(devices, "enabled_devices", None) or []:
|
||||
snapshot.records[device.name] = _record_for(device)
|
||||
if state is not None:
|
||||
state.snapshot = snapshot
|
||||
return snapshot
|
||||
|
||||
|
||||
def invalidate_device_snapshot(client) -> None:
|
||||
"""Drop the shared snapshot of ``client``, e.g. after changing devices in a test."""
|
||||
state = _catalog_state(client)
|
||||
if state is not None:
|
||||
state.bump()
|
||||
|
||||
|
||||
# ---- entries, search and rows ----------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PickerEntry:
|
||||
"""One selectable row of a picker popup.
|
||||
|
||||
Attributes:
|
||||
value: Text the combobox receives when the entry is picked.
|
||||
title: Primary label.
|
||||
subtitle: Muted secondary label, e.g. the device class.
|
||||
group: Section the entry is listed under when nothing is searched.
|
||||
badge: Short chip text, e.g. the readout priority or signal kind.
|
||||
icon: Material icon name.
|
||||
device: Device whose live value is shown for the entry.
|
||||
signal: Signal object name of the live value; empty for the device's main signal.
|
||||
details: Extra text for the footer, e.g. the description or tags.
|
||||
"""
|
||||
|
||||
value: str
|
||||
title: str
|
||||
subtitle: str = ""
|
||||
group: str = ""
|
||||
badge: str = ""
|
||||
icon: str = ""
|
||||
device: str = ""
|
||||
signal: str = ""
|
||||
details: str = ""
|
||||
|
||||
|
||||
def _word_starts(text: str) -> list[int]:
|
||||
starts = [0]
|
||||
for index in range(1, len(text)):
|
||||
if text[index - 1] in "_.-: " or (text[index].isupper() and text[index - 1].islower()):
|
||||
starts.append(index)
|
||||
return starts
|
||||
|
||||
|
||||
def rank_entry( # pylint: disable=too-many-return-statements
|
||||
query: str, entry: PickerEntry
|
||||
) -> tuple[int, int, int] | None:
|
||||
"""Score how well ``entry`` matches ``query``; lower is better.
|
||||
|
||||
Ranks: 0 title prefix, 1 prefix of a word in the title (``x`` matches ``sam_x``),
|
||||
2 substring of the title, 3 characters of the query in order (``smx`` matches ``samx``),
|
||||
4 substring of the subtitle, badge or details.
|
||||
|
||||
Args:
|
||||
query(str): Search text; matching ignores case and surrounding blanks.
|
||||
entry(PickerEntry): Entry to score.
|
||||
|
||||
Returns:
|
||||
tuple[int, int, int] | None: ``(rank, match_start, match_length)`` in the title, or None
|
||||
if the entry does not match. Ranks 3 and 4 report no title highlight (``-1, 0``).
|
||||
"""
|
||||
needle = query.strip().lower()
|
||||
if not needle:
|
||||
return (0, -1, 0)
|
||||
title = entry.title.lower()
|
||||
position = title.find(needle)
|
||||
if position == 0:
|
||||
return (0, 0, len(needle))
|
||||
if position > 0:
|
||||
for start in _word_starts(entry.title):
|
||||
if title.startswith(needle, start):
|
||||
return (1, start, len(needle))
|
||||
return (2, position, len(needle))
|
||||
cursor = 0
|
||||
for char in needle:
|
||||
cursor = title.find(char, cursor)
|
||||
if cursor < 0:
|
||||
break
|
||||
cursor += 1
|
||||
else:
|
||||
return (3, -1, 0)
|
||||
haystack = f"{entry.subtitle} {entry.badge} {entry.details} {entry.signal}".lower()
|
||||
if needle in haystack:
|
||||
return (4, -1, 0)
|
||||
return None
|
||||
|
||||
|
||||
def rich_title(title: str, start: int, length: int, color: str) -> str:
|
||||
"""Return ``title`` as Qt rich text with the matched part bold and coloured."""
|
||||
if start < 0 or length <= 0:
|
||||
return html.escape(title)
|
||||
before, match, after = title[:start], title[start : start + length], title[start + length :]
|
||||
return (
|
||||
f"{html.escape(before)}<b><font color='{color}'>{html.escape(match)}</font></b>"
|
||||
f"{html.escape(after)}"
|
||||
)
|
||||
|
||||
|
||||
ROW_ROLES = (
|
||||
"rowType",
|
||||
"value",
|
||||
"title",
|
||||
"titleRich",
|
||||
"subtitle",
|
||||
"badge",
|
||||
"icon",
|
||||
"matchStart",
|
||||
"matchLength",
|
||||
"isCurrent",
|
||||
"count",
|
||||
)
|
||||
|
||||
|
||||
class PickerRowsModel(DictListModel):
|
||||
"""Rows of a picker popup: ``header``, ``item`` and ``empty`` rows.
|
||||
|
||||
The display role returns the title, so plain item views and tests can read the rows too.
|
||||
"""
|
||||
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(ROW_ROLES, parent)
|
||||
|
||||
def data(self, index, role: int = Qt.ItemDataRole.DisplayRole):
|
||||
if role == Qt.ItemDataRole.DisplayRole and index.isValid():
|
||||
role = self.role_id("title")
|
||||
return super().data(index, role)
|
||||
|
||||
def role_id(self, key: str) -> int:
|
||||
"""Qt role id of a row key."""
|
||||
return Qt.ItemDataRole.UserRole + 1 + ROW_ROLES.index(key)
|
||||
|
||||
def row(self, row: int) -> dict:
|
||||
"""Row dictionary at ``row`` (not a copy; do not modify)."""
|
||||
return self._items[row] # pylint: disable=protected-access
|
||||
|
||||
|
||||
def _row(row_type: str, title: str, **values) -> dict:
|
||||
"""Row dictionary with every role of :data:`ROW_ROLES` set."""
|
||||
row = {
|
||||
"rowType": row_type,
|
||||
"value": "",
|
||||
"title": title,
|
||||
"titleRich": html.escape(title),
|
||||
"subtitle": "",
|
||||
"badge": "",
|
||||
"icon": "",
|
||||
"matchStart": -1,
|
||||
"matchLength": 0,
|
||||
"isCurrent": False,
|
||||
"count": 0,
|
||||
}
|
||||
row.update(values)
|
||||
return row
|
||||
|
||||
|
||||
def format_value(value: Any) -> str: # pylint: disable=too-many-return-statements
|
||||
"""Short text for a reading shown in the footer."""
|
||||
if value is None:
|
||||
return ""
|
||||
if isinstance(value, bool):
|
||||
return "true" if value else "false"
|
||||
if isinstance(value, int):
|
||||
return str(value)
|
||||
if isinstance(value, float):
|
||||
return f"{value:.6g}"
|
||||
shape = getattr(value, "shape", None)
|
||||
if shape is not None and len(shape) > 0:
|
||||
return f"array {'x'.join(str(dim) for dim in shape)}"
|
||||
if isinstance(value, (list, tuple)):
|
||||
return f"list [{len(value)}]"
|
||||
if isinstance(value, dict):
|
||||
return "{…}"
|
||||
text = str(value)
|
||||
return text if len(text) <= 40 else f"{text[:39]}…"
|
||||
|
||||
|
||||
class PickerController(QObject):
|
||||
"""Search, grouping, highlight and footer details of an open picker popup.
|
||||
|
||||
Both popups are thin views over this object; all behaviour that the user can observe lives
|
||||
here so the QWidget and QML versions behave identically.
|
||||
|
||||
Args:
|
||||
noun(str): Plural used in texts, e.g. ``"devices"``.
|
||||
live_reader: Called with an entry, returns the cached reading dict (``value``,
|
||||
``timestamp``) or None. Runs debounced when the highlight settles.
|
||||
live_watcher: Called with the highlighted entry (or None when the popup closes) so the
|
||||
picker can subscribe to live updates of that device only.
|
||||
parent: Parent QObject.
|
||||
"""
|
||||
|
||||
rows_changed = Signal()
|
||||
highlight_changed = Signal()
|
||||
detail_changed = Signal()
|
||||
query_changed = Signal()
|
||||
picked = Signal(str)
|
||||
dismissed = Signal()
|
||||
|
||||
LIVE_DELAY_MS = 120
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
noun: str,
|
||||
live_reader: Callable[[PickerEntry], dict | None] | None = None,
|
||||
live_watcher: Callable[[PickerEntry | None], None] | None = None,
|
||||
parent: QObject | None = None,
|
||||
):
|
||||
super().__init__(parent)
|
||||
self._noun = noun
|
||||
self._live_reader = live_reader
|
||||
self._live_watcher = live_watcher
|
||||
self._entries: list[PickerEntry] = []
|
||||
self._recent: list[str] = []
|
||||
self._current = ""
|
||||
self._query = ""
|
||||
self._row_entries: list[PickerEntry | None] = []
|
||||
self._highlight = -1
|
||||
self._match_count = 0
|
||||
self._detail: dict = {}
|
||||
self._accent = ""
|
||||
self._watched: PickerEntry | None = None
|
||||
self.model = PickerRowsModel(self)
|
||||
self._live_timer = QTimer(self)
|
||||
self._live_timer.setSingleShot(True)
|
||||
self._live_timer.setInterval(self.LIVE_DELAY_MS)
|
||||
self._live_timer.timeout.connect(self._load_live_value)
|
||||
|
||||
# ---- state -----------------------------------------------------------------------------
|
||||
|
||||
def open(
|
||||
self,
|
||||
entries: Sequence[PickerEntry],
|
||||
current: str = "",
|
||||
query: str = "",
|
||||
recent: Sequence[str] = (),
|
||||
) -> None:
|
||||
"""Load the entries shown by a popup that is about to open.
|
||||
|
||||
Args:
|
||||
entries: All selectable entries in display order.
|
||||
current: Value of the current selection; it is marked and highlighted.
|
||||
query: Initial search text, e.g. what the user started typing in the field.
|
||||
recent: Recently picked values listed first while nothing is searched.
|
||||
"""
|
||||
self._entries = list(entries)
|
||||
self._current = current or ""
|
||||
self._recent = [value for value in recent if any(e.value == value for e in self._entries)]
|
||||
self._query = query
|
||||
self._rebuild()
|
||||
self.query_changed.emit()
|
||||
|
||||
def close(self) -> None:
|
||||
"""Stop live updates; called when the popup hides."""
|
||||
self._live_timer.stop()
|
||||
if self._watched is not None and self._live_watcher is not None:
|
||||
self._live_watcher(None)
|
||||
self._watched = None
|
||||
|
||||
def set_query(self, query: str) -> None:
|
||||
"""Filter the rows by ``query``."""
|
||||
if query == self._query:
|
||||
return
|
||||
self._query = query
|
||||
self._rebuild()
|
||||
self.query_changed.emit()
|
||||
|
||||
def move_highlight(self, step: int) -> None:
|
||||
"""Move the highlight by ``step`` selectable rows, wrapping at the ends."""
|
||||
selectable = [row for row, entry in enumerate(self._row_entries) if entry is not None]
|
||||
if not selectable:
|
||||
return
|
||||
if self._highlight not in selectable:
|
||||
target = selectable[0] if step > 0 else selectable[-1]
|
||||
else:
|
||||
position = selectable.index(self._highlight) + step
|
||||
target = selectable[position % len(selectable)]
|
||||
self.set_highlight(target)
|
||||
|
||||
def page_highlight(self, step: int, rows: int = 8) -> None:
|
||||
"""Move the highlight by a page of ``rows`` rows without wrapping."""
|
||||
selectable = [row for row, entry in enumerate(self._row_entries) if entry is not None]
|
||||
if not selectable:
|
||||
return
|
||||
position = selectable.index(self._highlight) if self._highlight in selectable else 0
|
||||
position = max(0, min(len(selectable) - 1, position + step * rows))
|
||||
self.set_highlight(selectable[position])
|
||||
|
||||
def set_highlight(self, row: int) -> None:
|
||||
"""Highlight ``row`` if it is selectable."""
|
||||
if not 0 <= row < len(self._row_entries) or self._row_entries[row] is None:
|
||||
return
|
||||
if row == self._highlight:
|
||||
return
|
||||
self._highlight = row
|
||||
self.highlight_changed.emit()
|
||||
self._update_detail()
|
||||
|
||||
def accept(self, row: int = -1) -> bool:
|
||||
"""Pick ``row``, or the highlighted row when ``row`` is -1.
|
||||
|
||||
Returns:
|
||||
bool: True if an entry was picked.
|
||||
"""
|
||||
row = self._highlight if row < 0 else row
|
||||
if not 0 <= row < len(self._row_entries):
|
||||
return False
|
||||
entry = self._row_entries[row]
|
||||
if entry is None:
|
||||
return False
|
||||
self.picked.emit(entry.value)
|
||||
return True
|
||||
|
||||
def dismiss(self) -> None:
|
||||
"""Close without picking."""
|
||||
self.dismissed.emit()
|
||||
|
||||
# ---- read-only state used by the views ---------------------------------------------------
|
||||
|
||||
@property
|
||||
def query(self) -> str:
|
||||
"""Current search text."""
|
||||
return self._query
|
||||
|
||||
@property
|
||||
def highlight(self) -> int:
|
||||
"""Highlighted row, -1 if none."""
|
||||
return self._highlight
|
||||
|
||||
@property
|
||||
def detail(self) -> dict:
|
||||
"""Footer contents for the highlighted entry."""
|
||||
return self._detail
|
||||
|
||||
@property
|
||||
def summary(self) -> str:
|
||||
"""Count text shown next to the search field."""
|
||||
total = len(self._entries)
|
||||
if not self._query.strip():
|
||||
return f"{total} {self._noun}"
|
||||
return f"{self._match_count} of {total}"
|
||||
|
||||
@property
|
||||
def placeholder(self) -> str:
|
||||
"""Placeholder of the search field."""
|
||||
return f"Search {self._noun}…"
|
||||
|
||||
def entry_at(self, row: int) -> PickerEntry | None:
|
||||
"""Entry shown in ``row`` (None for headers and the empty row)."""
|
||||
if 0 <= row < len(self._row_entries):
|
||||
return self._row_entries[row]
|
||||
return None
|
||||
|
||||
# ---- internals ---------------------------------------------------------------------------
|
||||
|
||||
def _item_row(self, entry: PickerEntry, start: int = -1, length: int = 0) -> tuple:
|
||||
accent = self._accent
|
||||
return (
|
||||
_row(
|
||||
"item",
|
||||
entry.title,
|
||||
value=entry.value,
|
||||
titleRich=rich_title(entry.title, start, length, accent),
|
||||
subtitle=entry.subtitle,
|
||||
badge=entry.badge,
|
||||
icon=entry.icon,
|
||||
matchStart=start,
|
||||
matchLength=length,
|
||||
isCurrent=entry.value == self._current,
|
||||
),
|
||||
entry,
|
||||
)
|
||||
|
||||
def _ranked_rows(self, query: str) -> list[tuple]:
|
||||
ranked = []
|
||||
for order, entry in enumerate(self._entries):
|
||||
score = rank_entry(query, entry)
|
||||
if score is not None:
|
||||
ranked.append((score[0], order, entry, score))
|
||||
ranked.sort(key=lambda item: (item[0], item[1]))
|
||||
self._match_count = len(ranked)
|
||||
return [self._item_row(entry, score[1], score[2]) for _, _, entry, score in ranked]
|
||||
|
||||
def _grouped_rows(self) -> list[tuple]:
|
||||
self._match_count = len(self._entries)
|
||||
rows = []
|
||||
if self._recent and len(self._entries) > MAX_RECENT * 2:
|
||||
by_value = {entry.value: entry for entry in self._entries}
|
||||
rows.append((_row("header", "Recent", icon="history", count=len(self._recent)), None))
|
||||
rows.extend(self._item_row(by_value[value]) for value in self._recent)
|
||||
groups: dict[str, list[PickerEntry]] = {}
|
||||
for entry in self._entries:
|
||||
groups.setdefault(entry.group, []).append(entry)
|
||||
show_headers = len(groups) > 1 or bool(rows)
|
||||
for group, entries in groups.items():
|
||||
if show_headers and group:
|
||||
rows.append((_row("header", group, count=len(entries)), None))
|
||||
rows.extend(self._item_row(entry) for entry in entries)
|
||||
return rows
|
||||
|
||||
def _rebuild(self) -> None:
|
||||
self._accent = ThemeTokens().primary.name()
|
||||
query = self._query.strip()
|
||||
rows = self._ranked_rows(query) if query else self._grouped_rows()
|
||||
if not any(entry is not None for _, entry in rows):
|
||||
text = f"No {self._noun} match “{query}”" if query else f"No {self._noun} available"
|
||||
rows = [(_row("empty", text, icon="search_off" if query else "block"), None)]
|
||||
self._row_entries = [entry for _, entry in rows]
|
||||
self.model.set_items([row for row, _ in rows])
|
||||
self.rows_changed.emit()
|
||||
|
||||
selectable = [row for row, entry in enumerate(self._row_entries) if entry is not None]
|
||||
wanted = None if query else self._current
|
||||
target = next(
|
||||
(row for row in selectable if self._row_entries[row].value == wanted),
|
||||
selectable[0] if selectable else -1,
|
||||
)
|
||||
self._highlight = target
|
||||
self.highlight_changed.emit()
|
||||
self._update_detail()
|
||||
|
||||
def _update_detail(self) -> None:
|
||||
entry = self.entry_at(self._highlight)
|
||||
if entry is None:
|
||||
self._detail = {}
|
||||
self._live_timer.stop()
|
||||
else:
|
||||
self._detail = {
|
||||
"title": entry.title,
|
||||
"subtitle": entry.subtitle,
|
||||
"details": entry.details,
|
||||
"icon": entry.icon,
|
||||
"badge": entry.badge,
|
||||
"value": "",
|
||||
"valueTime": "",
|
||||
"valueState": "loading" if self._live_reader is not None else "none",
|
||||
}
|
||||
if self._live_reader is not None:
|
||||
self._live_timer.start()
|
||||
self.detail_changed.emit()
|
||||
|
||||
def _load_live_value(self) -> None:
|
||||
entry = self.entry_at(self._highlight)
|
||||
if entry is None:
|
||||
return
|
||||
reading = None
|
||||
try:
|
||||
reading = self._live_reader(entry) if self._live_reader is not None else None
|
||||
except Exception as exc: # pylint: disable=broad-except
|
||||
logger.debug(f"Could not read cached value of {entry.device}: {exc}")
|
||||
self.update_live_value(entry, reading)
|
||||
if self._live_watcher is not None and entry != self._watched:
|
||||
self._watched = entry
|
||||
self._live_watcher(entry)
|
||||
|
||||
def update_live_value(self, entry: PickerEntry, reading: dict | None) -> None:
|
||||
"""Show ``reading`` in the footer if ``entry`` is still highlighted.
|
||||
|
||||
Args:
|
||||
entry(PickerEntry): Entry the reading belongs to.
|
||||
reading(dict | None): Reading with ``value`` and optional ``timestamp``.
|
||||
"""
|
||||
if self.entry_at(self._highlight) != entry or not self._detail:
|
||||
return
|
||||
detail = dict(self._detail)
|
||||
if reading is None or "value" not in reading:
|
||||
detail.update({"value": "", "valueTime": "", "valueState": "none"})
|
||||
else:
|
||||
stamp = reading.get("timestamp")
|
||||
detail.update(
|
||||
{
|
||||
"value": format_value(reading.get("value")),
|
||||
"valueTime": (
|
||||
time.strftime("%H:%M:%S", time.localtime(stamp))
|
||||
if isinstance(stamp, (int, float)) and stamp > 0
|
||||
else ""
|
||||
),
|
||||
"valueState": "live",
|
||||
}
|
||||
)
|
||||
self._detail = detail
|
||||
self.detail_changed.emit()
|
||||
|
||||
|
||||
# ---- recent picks ------------------------------------------------------------------------------
|
||||
|
||||
_RECENT: dict[str, list[str]] = {"devices": [], "signals": []}
|
||||
|
||||
|
||||
def remember_pick(kind: str, value: str) -> None:
|
||||
"""Remember ``value`` as recently picked for pickers of ``kind`` in this session."""
|
||||
if not value:
|
||||
return
|
||||
recent = _RECENT.setdefault(kind, [])
|
||||
if value in recent:
|
||||
recent.remove(value)
|
||||
recent.insert(0, value)
|
||||
del recent[MAX_RECENT:]
|
||||
|
||||
|
||||
def recent_picks(kind: str) -> list[str]:
|
||||
"""Recently picked values for pickers of ``kind``, newest first."""
|
||||
return list(_RECENT.get(kind, []))
|
||||
|
||||
|
||||
def clear_recent_picks() -> None:
|
||||
"""Forget the recent picks of all pickers."""
|
||||
for values in _RECENT.values():
|
||||
values.clear()
|
||||
@@ -0,0 +1,133 @@
|
||||
"""Device and signal pickers whose popup is rendered with Qt Quick (QML).
|
||||
|
||||
Same behaviour as the QWidget pickers in :mod:`picker_qwidget`; both popups are views over
|
||||
:class:`~bec_widgets.widgets.control.device_input.picker.picker_common.PickerController`.
|
||||
The field itself stays a :class:`QComboBox` so the pickers remain drop-in replacements for
|
||||
:class:`DeviceComboBox` and :class:`SignalComboBox`; only the popup is QML.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from qtpy.QtCore import Property, QObject, Qt, Signal, Slot
|
||||
from qtpy.QtWidgets import QVBoxLayout
|
||||
|
||||
from bec_widgets.utils.quick import create_quick_widget, release_quick_widget
|
||||
from bec_widgets.widgets.control.device_input.picker.picker_common import (
|
||||
DevicePickerBase,
|
||||
PickerPopupBase,
|
||||
SignalPickerBase,
|
||||
)
|
||||
from bec_widgets.widgets.control.device_input.picker.picker_model import PickerController
|
||||
|
||||
QML_FILE = Path(__file__).parent / "qml" / "PickerPopupView.qml"
|
||||
|
||||
|
||||
class PickerBackend(QObject):
|
||||
"""Bridge between :class:`PickerController` and ``PickerPopupView.qml``."""
|
||||
|
||||
rowsChanged = Signal() # pylint: disable=invalid-name
|
||||
highlightChanged = Signal() # pylint: disable=invalid-name
|
||||
detailChanged = Signal() # pylint: disable=invalid-name
|
||||
queryChanged = Signal() # pylint: disable=invalid-name
|
||||
focusRequested = Signal() # pylint: disable=invalid-name
|
||||
|
||||
def __init__(self, controller: PickerController, parent: QObject | None = None):
|
||||
super().__init__(parent)
|
||||
self._controller = controller
|
||||
controller.rows_changed.connect(self.rowsChanged)
|
||||
controller.highlight_changed.connect(self.highlightChanged)
|
||||
controller.detail_changed.connect(self.detailChanged)
|
||||
controller.query_changed.connect(self.queryChanged)
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
@Slot(str)
|
||||
def setQuery(self, text: str) -> None:
|
||||
"""Filter the rows."""
|
||||
self._controller.set_query(text)
|
||||
|
||||
@Slot(int)
|
||||
def move(self, step: int) -> None:
|
||||
"""Move the highlight."""
|
||||
self._controller.move_highlight(step)
|
||||
|
||||
@Slot(int)
|
||||
def page(self, step: int) -> None:
|
||||
"""Move the highlight by a page."""
|
||||
self._controller.page_highlight(step)
|
||||
|
||||
@Slot(int)
|
||||
def hover(self, row: int) -> None:
|
||||
"""Highlight the row under the mouse."""
|
||||
self._controller.set_highlight(row)
|
||||
|
||||
@Slot(int)
|
||||
def accept(self, row: int) -> None:
|
||||
"""Pick a row (-1 for the highlighted one)."""
|
||||
self._controller.accept(row)
|
||||
|
||||
@Slot()
|
||||
def dismiss(self) -> None:
|
||||
"""Close without picking."""
|
||||
self._controller.dismiss()
|
||||
|
||||
@property
|
||||
def controller(self) -> PickerController:
|
||||
"""Controller this backend forwards."""
|
||||
return self._controller
|
||||
|
||||
rows = Property(QObject, lambda self: self.controller.model, constant=True)
|
||||
highlight = Property(int, lambda self: self.controller.highlight, notify=highlightChanged)
|
||||
query = Property(str, lambda self: self.controller.query, notify=queryChanged)
|
||||
placeholder = Property(str, lambda self: self.controller.placeholder, notify=queryChanged)
|
||||
summary = Property(str, lambda self: self.controller.summary, notify=rowsChanged)
|
||||
detail = Property("QVariantMap", lambda self: self.controller.detail, notify=detailChanged)
|
||||
|
||||
|
||||
class PickerPopupQml(PickerPopupBase):
|
||||
"""Popup window that hosts ``PickerPopupView.qml``."""
|
||||
|
||||
def __init__(self, picker):
|
||||
super().__init__(picker)
|
||||
self.backend = PickerBackend(self.controller, self)
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
self.view = create_quick_widget(self, QML_FILE, {"backend": self.backend})
|
||||
self.view.setFocusPolicy(Qt.FocusPolicy.StrongFocus)
|
||||
layout.addWidget(self.view)
|
||||
self.refresh_theme()
|
||||
|
||||
def focus_search(self) -> None:
|
||||
self.view.setFocus()
|
||||
self.backend.focusRequested.emit()
|
||||
|
||||
def refresh_theme(self) -> None:
|
||||
self.view.setClearColor(self.palette().window().color())
|
||||
|
||||
def release(self) -> None:
|
||||
release_quick_widget(self.view)
|
||||
|
||||
|
||||
class DevicePickerQML(DevicePickerBase):
|
||||
"""Device combobox with a searchable, grouped popup rendered in QML.
|
||||
|
||||
Drop-in replacement for :class:`DeviceComboBox`; see :class:`DevicePickerBase`.
|
||||
"""
|
||||
|
||||
ICON_NAME = "manage_search"
|
||||
|
||||
def _create_popup(self):
|
||||
return PickerPopupQml(self)
|
||||
|
||||
|
||||
class SignalPickerQML(SignalPickerBase):
|
||||
"""Signal combobox with a searchable, grouped popup rendered in QML.
|
||||
|
||||
Drop-in replacement for :class:`SignalComboBox`; see :class:`SignalPickerBase`.
|
||||
"""
|
||||
|
||||
ICON_NAME = "manage_search"
|
||||
|
||||
def _create_popup(self):
|
||||
return PickerPopupQml(self)
|
||||
@@ -0,0 +1,470 @@
|
||||
"""Device and signal pickers whose popup is built from QWidgets.
|
||||
|
||||
Same behaviour as the QML pickers in :mod:`picker_qml`; both popups are views over
|
||||
:class:`~bec_widgets.widgets.control.device_input.picker.picker_common.PickerController`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from bec_qthemes import material_icon
|
||||
from qtpy.QtCore import QEvent, QModelIndex, QRect, QRectF, QSize, Qt
|
||||
from qtpy.QtGui import QColor, QFont, QFontMetrics, QPainter, QPainterPath, QPixmap, QRegion
|
||||
from qtpy.QtWidgets import (
|
||||
QAbstractItemView,
|
||||
QFrame,
|
||||
QHBoxLayout,
|
||||
QLabel,
|
||||
QLineEdit,
|
||||
QListView,
|
||||
QStyledItemDelegate,
|
||||
QStyleOptionViewItem,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
|
||||
from bec_widgets.utils.quick.host import ThemeTokens
|
||||
from bec_widgets.utils.ux_kit import rgba
|
||||
from bec_widgets.widgets.control.device_input.picker.picker_common import (
|
||||
DevicePickerBase,
|
||||
PickerPopupBase,
|
||||
SignalPickerBase,
|
||||
)
|
||||
from bec_widgets.widgets.control.device_input.picker.picker_model import PickerRowsModel
|
||||
|
||||
ROW_HEIGHT = 34
|
||||
HEADER_HEIGHT = 28
|
||||
EMPTY_HEIGHT = 96
|
||||
|
||||
_ICON_CACHE: dict[tuple[str, str, int], QPixmap] = {}
|
||||
|
||||
|
||||
def cached_icon(name: str, color: QColor, size: int = 16) -> QPixmap:
|
||||
"""Material icon pixmap, cached per name, colour and size."""
|
||||
key = (name, color.name(), size)
|
||||
pixmap = _ICON_CACHE.get(key)
|
||||
if pixmap is None:
|
||||
pixmap = material_icon(name, size=(size * 2, size * 2), color=color)
|
||||
pixmap.setDevicePixelRatio(2.0)
|
||||
_ICON_CACHE[key] = pixmap
|
||||
return pixmap
|
||||
|
||||
|
||||
class PickerRowDelegate(QStyledItemDelegate):
|
||||
"""Paints header, item and empty rows like ``PickerPopupView.qml``."""
|
||||
|
||||
def __init__(self, popup: "PickerPopupWidget"):
|
||||
super().__init__(popup)
|
||||
self._popup = popup
|
||||
self.tokens = ThemeTokens()
|
||||
|
||||
def _row(self, index: QModelIndex) -> dict:
|
||||
model = index.model()
|
||||
if isinstance(model, PickerRowsModel):
|
||||
return model.row(index.row())
|
||||
return {}
|
||||
|
||||
def sizeHint(self, option, index): # pylint: disable=invalid-name
|
||||
row_type = self._row(index).get("rowType")
|
||||
height = {"header": HEADER_HEIGHT, "empty": EMPTY_HEIGHT}.get(row_type, ROW_HEIGHT)
|
||||
return QSize(option.rect.width(), height)
|
||||
|
||||
def paint(self, painter: QPainter, option: QStyleOptionViewItem, index: QModelIndex):
|
||||
row = self._row(index)
|
||||
painter.save()
|
||||
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
|
||||
rect = option.rect
|
||||
row_type = row.get("rowType")
|
||||
if row_type == "header":
|
||||
self._paint_header(painter, rect, row)
|
||||
elif row_type == "empty":
|
||||
self._paint_empty(painter, rect, row)
|
||||
else:
|
||||
highlighted = index.row() == self._popup.controller.highlight
|
||||
self._paint_item(painter, rect, row, highlighted)
|
||||
painter.restore()
|
||||
|
||||
def _paint_header(self, painter: QPainter, rect: QRect, row: dict) -> None:
|
||||
tokens = self.tokens
|
||||
font = QFont(painter.font())
|
||||
font.setPixelSize(11)
|
||||
font.setWeight(QFont.Weight.DemiBold)
|
||||
font.setLetterSpacing(QFont.SpacingType.AbsoluteSpacing, 0.6)
|
||||
painter.setFont(font)
|
||||
x = rect.left() + 12
|
||||
if row.get("icon"):
|
||||
painter.drawPixmap(
|
||||
x,
|
||||
rect.top() + (rect.height() - 14) // 2 + 2,
|
||||
cached_icon(row["icon"], tokens.fg_subtle, 14),
|
||||
)
|
||||
x += 20
|
||||
painter.setPen(tokens.fg_subtle)
|
||||
title = row.get("title", "").upper()
|
||||
text_rect = QRect(x, rect.top() + 4, rect.width() - x - 12, rect.height() - 4)
|
||||
painter.drawText(text_rect, Qt.AlignmentFlag.AlignVCenter, title)
|
||||
width = QFontMetrics(font).horizontalAdvance(title)
|
||||
count = row.get("count") or 0
|
||||
if count:
|
||||
font.setWeight(QFont.Weight.Normal)
|
||||
painter.setFont(font)
|
||||
painter.drawText(
|
||||
text_rect.adjusted(width + 8, 0, 0, 0), Qt.AlignmentFlag.AlignVCenter, str(count)
|
||||
)
|
||||
|
||||
def _paint_empty(self, painter: QPainter, rect: QRect, row: dict) -> None:
|
||||
tokens = self.tokens
|
||||
icon = cached_icon(row.get("icon") or "search_off", tokens.fg_subtle, 24)
|
||||
painter.drawPixmap(rect.center().x() - 12, rect.top() + 18, icon)
|
||||
font = QFont(painter.font())
|
||||
font.setPixelSize(13)
|
||||
painter.setFont(font)
|
||||
painter.setPen(tokens.fg_muted)
|
||||
painter.drawText(
|
||||
rect.adjusted(12, 48, -12, -8),
|
||||
Qt.AlignmentFlag.AlignHCenter | Qt.AlignmentFlag.AlignTop,
|
||||
row.get("title", ""),
|
||||
)
|
||||
|
||||
def _paint_item(self, painter: QPainter, rect: QRect, row: dict, highlighted: bool) -> None:
|
||||
tokens = self.tokens
|
||||
card = QRectF(rect.adjusted(4, 1, -4, -1))
|
||||
painter.setPen(Qt.PenStyle.NoPen)
|
||||
if highlighted:
|
||||
painter.setBrush(tokens.hover)
|
||||
painter.drawRoundedRect(card, 6, 6)
|
||||
if row.get("isCurrent"):
|
||||
painter.setBrush(tokens.primary)
|
||||
painter.drawRoundedRect(
|
||||
QRectF(card.left(), card.top() + 7, 3, card.height() - 14), 1.5, 1.5
|
||||
)
|
||||
left = int(card.left()) + 10
|
||||
if row.get("icon"):
|
||||
color = tokens.primary if row.get("isCurrent") else tokens.fg_muted
|
||||
painter.drawPixmap(
|
||||
left, rect.top() + (rect.height() - 16) // 2, cached_icon(row["icon"], color)
|
||||
)
|
||||
right = self._paint_badge(painter, rect, int(card.right()) - 8, row.get("badge") or "")
|
||||
self._paint_title(painter, rect, left + 24, right, row)
|
||||
|
||||
def _paint_badge(self, painter: QPainter, rect: QRect, right: int, badge: str) -> int:
|
||||
"""Paint the chip right-aligned at ``right``; return the x where the text must end."""
|
||||
if not badge:
|
||||
return right
|
||||
font = QFont(painter.font())
|
||||
font.setPixelSize(11)
|
||||
font.setWeight(QFont.Weight.Medium)
|
||||
width = QFontMetrics(font).horizontalAdvance(badge) + 14
|
||||
chip = QRectF(right - width, rect.top() + (rect.height() - 18) / 2, width, 18)
|
||||
tone = _badge_tone(self.tokens, badge)
|
||||
painter.setPen(Qt.PenStyle.NoPen)
|
||||
painter.setBrush(self.tokens.soft(tone, 0.18))
|
||||
painter.drawRoundedRect(chip, 9, 9)
|
||||
painter.setFont(font)
|
||||
painter.setPen(tone)
|
||||
painter.drawText(chip, Qt.AlignmentFlag.AlignCenter, badge)
|
||||
return int(chip.left()) - 8
|
||||
|
||||
def _paint_title(self, painter: QPainter, rect: QRect, left: int, right: int, row: dict):
|
||||
"""Paint the title with the matched part highlighted, then the muted subtitle."""
|
||||
tokens = self.tokens
|
||||
font = QFont(painter.font())
|
||||
font.setPixelSize(13)
|
||||
font.setWeight(QFont.Weight.Medium if row.get("isCurrent") else QFont.Weight.Normal)
|
||||
bold = QFont(font)
|
||||
bold.setWeight(QFont.Weight.Bold)
|
||||
title = row.get("title", "")
|
||||
start, length = row.get("matchStart", -1), row.get("matchLength", 0)
|
||||
segments = [(title, font, tokens.fg)]
|
||||
if start >= 0 and length > 0:
|
||||
segments = [
|
||||
(title[:start], font, tokens.fg),
|
||||
(title[start : start + length], bold, tokens.primary),
|
||||
(title[start + length :], font, tokens.fg),
|
||||
]
|
||||
cursor = left
|
||||
for text, segment_font, color in segments:
|
||||
if not text:
|
||||
continue
|
||||
metrics = QFontMetrics(segment_font)
|
||||
text = metrics.elidedText(text, Qt.TextElideMode.ElideRight, max(0, right - cursor))
|
||||
painter.setFont(segment_font)
|
||||
painter.setPen(color)
|
||||
painter.drawText(
|
||||
QRect(cursor, rect.top(), max(0, right - cursor), rect.height()),
|
||||
Qt.AlignmentFlag.AlignVCenter,
|
||||
text,
|
||||
)
|
||||
cursor += metrics.horizontalAdvance(text)
|
||||
subtitle = row.get("subtitle") or ""
|
||||
if not subtitle or cursor + 38 >= right:
|
||||
return
|
||||
font.setPixelSize(12)
|
||||
font.setWeight(QFont.Weight.Normal)
|
||||
cursor += 8
|
||||
painter.setFont(font)
|
||||
painter.setPen(tokens.fg_subtle)
|
||||
painter.drawText(
|
||||
QRect(cursor, rect.top(), right - cursor, rect.height()),
|
||||
Qt.AlignmentFlag.AlignVCenter,
|
||||
QFontMetrics(font).elidedText(subtitle, Qt.TextElideMode.ElideRight, right - cursor),
|
||||
)
|
||||
|
||||
|
||||
def _badge_tone(tokens: ThemeTokens, badge: str) -> QColor:
|
||||
"""Chip colour of a readout priority or signal kind badge."""
|
||||
return {
|
||||
"monitored": tokens.primary,
|
||||
"hinted": tokens.primary,
|
||||
"async": tokens.highlight,
|
||||
"baseline": tokens.fg_muted,
|
||||
"config": tokens.fg_muted,
|
||||
"on_request": tokens.warning,
|
||||
"continuous": tokens.success,
|
||||
"normal": tokens.success,
|
||||
}.get(badge, tokens.fg_muted)
|
||||
|
||||
|
||||
class _SearchField(QLineEdit):
|
||||
"""Search input that forwards navigation keys to the controller."""
|
||||
|
||||
def __init__(self, popup: "PickerPopupWidget"):
|
||||
super().__init__(popup)
|
||||
self._popup = popup
|
||||
|
||||
def keyPressEvent(self, event): # pylint: disable=invalid-name
|
||||
controller = self._popup.controller
|
||||
key = event.key()
|
||||
if key == Qt.Key.Key_Down or (key == Qt.Key.Key_Tab and not event.modifiers()):
|
||||
controller.move_highlight(1)
|
||||
elif key in (Qt.Key.Key_Up, Qt.Key.Key_Backtab):
|
||||
controller.move_highlight(-1)
|
||||
elif key == Qt.Key.Key_PageDown:
|
||||
controller.page_highlight(1)
|
||||
elif key == Qt.Key.Key_PageUp:
|
||||
controller.page_highlight(-1)
|
||||
elif key in (Qt.Key.Key_Return, Qt.Key.Key_Enter):
|
||||
controller.accept()
|
||||
elif key == Qt.Key.Key_Escape:
|
||||
controller.dismiss()
|
||||
else:
|
||||
super().keyPressEvent(event)
|
||||
|
||||
|
||||
class PickerPopupWidget(PickerPopupBase): # pylint: disable=too-many-instance-attributes
|
||||
"""Popup with search field, grouped list and live-value footer, built from QWidgets."""
|
||||
|
||||
def __init__(self, picker):
|
||||
super().__init__(picker)
|
||||
self.card = QFrame(self)
|
||||
self.card.setObjectName("pickerCard")
|
||||
outer = QVBoxLayout(self)
|
||||
outer.setContentsMargins(0, 0, 0, 0)
|
||||
outer.addWidget(self.card)
|
||||
layout = QVBoxLayout(self.card)
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
layout.setSpacing(0)
|
||||
layout.addLayout(self._build_search())
|
||||
layout.addWidget(self._separator())
|
||||
layout.addWidget(self._build_list(), 1)
|
||||
layout.addWidget(self._separator())
|
||||
layout.addWidget(self._build_footer())
|
||||
|
||||
self.search.textChanged.connect(self.controller.set_query)
|
||||
self.controller.rows_changed.connect(self._sync_rows)
|
||||
self.controller.highlight_changed.connect(self._sync_highlight)
|
||||
self.controller.detail_changed.connect(self._sync_detail)
|
||||
self.controller.query_changed.connect(self._sync_query)
|
||||
self.refresh_theme()
|
||||
self._sync_query()
|
||||
self._sync_rows()
|
||||
self._sync_highlight()
|
||||
|
||||
def _build_search(self) -> QHBoxLayout:
|
||||
header = QHBoxLayout()
|
||||
header.setContentsMargins(12, 7, 12, 7)
|
||||
header.setSpacing(8)
|
||||
self.search_icon = QLabel(self.card)
|
||||
self.search = _SearchField(self)
|
||||
self.search.setObjectName("pickerSearch")
|
||||
self.search.setClearButtonEnabled(True)
|
||||
self.summary = QLabel(self.card)
|
||||
self.summary.setObjectName("pickerSummary")
|
||||
header.addWidget(self.search_icon)
|
||||
header.addWidget(self.search, 1)
|
||||
header.addWidget(self.summary)
|
||||
return header
|
||||
|
||||
def _build_list(self) -> QListView:
|
||||
self.list = QListView(self.card)
|
||||
self.list.setObjectName("pickerList")
|
||||
self.list.setFrameShape(QFrame.Shape.NoFrame)
|
||||
self.list.setMouseTracking(True)
|
||||
self.list.setSelectionMode(QAbstractItemView.SelectionMode.NoSelection)
|
||||
self.list.setFocusPolicy(Qt.FocusPolicy.NoFocus)
|
||||
self.list.setVerticalScrollMode(QAbstractItemView.ScrollMode.ScrollPerPixel)
|
||||
self.list.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
|
||||
self.list.viewport().installEventFilter(self)
|
||||
self.delegate = PickerRowDelegate(self)
|
||||
self.list.setItemDelegate(self.delegate)
|
||||
self.list.setModel(self.controller.model)
|
||||
self.list.clicked.connect(lambda index: self.controller.accept(index.row()))
|
||||
return self.list
|
||||
|
||||
def _build_footer(self) -> QWidget:
|
||||
self.footer = QWidget(self.card)
|
||||
footer = QVBoxLayout(self.footer)
|
||||
footer.setContentsMargins(12, 8, 12, 8)
|
||||
footer.setSpacing(2)
|
||||
top = QHBoxLayout()
|
||||
top.setSpacing(8)
|
||||
self.detail_icon = QLabel(self.footer)
|
||||
self.detail_title = QLabel(self.footer)
|
||||
self.detail_title.setObjectName("pickerDetailTitle")
|
||||
self.detail_value = QLabel(self.footer)
|
||||
self.detail_value.setObjectName("pickerDetailValue")
|
||||
self.detail_value.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter)
|
||||
top.addWidget(self.detail_icon)
|
||||
top.addWidget(self.detail_title, 1)
|
||||
top.addWidget(self.detail_value)
|
||||
footer.addLayout(top)
|
||||
self.detail_sub = QLabel(self.footer)
|
||||
self.detail_sub.setObjectName("pickerDetailSub")
|
||||
footer.addWidget(self.detail_sub)
|
||||
self.hints = QLabel("↑↓ move ↵ select esc close", self.footer)
|
||||
self.hints.setObjectName("pickerHints")
|
||||
footer.addWidget(self.hints)
|
||||
return self.footer
|
||||
|
||||
def _separator(self) -> QFrame:
|
||||
line = QFrame(self.card)
|
||||
line.setObjectName("pickerSeparator")
|
||||
line.setFixedHeight(1)
|
||||
return line
|
||||
|
||||
# ---- PickerPopupBase -----------------------------------------------------------------------
|
||||
|
||||
def focus_search(self) -> None:
|
||||
self.search.setFocus()
|
||||
self.search.end(False)
|
||||
|
||||
def refresh_theme(self) -> None:
|
||||
tokens = ThemeTokens()
|
||||
self.delegate.tokens = tokens
|
||||
self.search_icon.setPixmap(cached_icon("search", tokens.fg_muted, 18))
|
||||
self.setStyleSheet(f"""
|
||||
QFrame#pickerCard {{ background: {tokens.card.name()};
|
||||
border: 1px solid {tokens.border.name()}; border-radius: 10px; }}
|
||||
QFrame#pickerSeparator {{ background: {tokens.border.name()}; border: none; }}
|
||||
QLineEdit#pickerSearch {{ background: transparent; border: none; color: {tokens.fg.name()};
|
||||
font-size: 14px; padding: 0px; min-height: 32px; max-height: 32px;
|
||||
selection-background-color: {tokens.primary.name()}; }}
|
||||
QLabel {{ background: transparent; }}
|
||||
QLabel#pickerSummary {{ color: {tokens.fg_subtle.name()}; font-size: 12px; }}
|
||||
QLabel#pickerDetailTitle {{ color: {tokens.fg.name()}; font-size: 13px; font-weight: 600; }}
|
||||
QLabel#pickerDetailSub {{ color: {tokens.fg_muted.name()}; font-size: 12px; }}
|
||||
QLabel#pickerHints {{ color: {tokens.fg_subtle.name()}; font-size: 11px; padding-top: 4px; }}
|
||||
QListView#pickerList {{ background: transparent; border: none; outline: none; }}
|
||||
QScrollBar:vertical {{ background: transparent; width: 8px; margin: 2px; }}
|
||||
QScrollBar::handle:vertical {{ background: {rgba(tokens.fg, 0.25)}; border-radius: 3px;
|
||||
min-height: 24px; }}
|
||||
QScrollBar::add-line, QScrollBar::sub-line {{ height: 0px; }}
|
||||
QScrollBar::add-page, QScrollBar::sub-page {{ background: none; }}
|
||||
""")
|
||||
self.search.setPlaceholderText(self.controller.placeholder)
|
||||
self._sync_detail()
|
||||
self.list.viewport().update()
|
||||
|
||||
# ---- sync from the controller ----------------------------------------------------------------
|
||||
|
||||
def _sync_query(self) -> None:
|
||||
if self.search.text() != self.controller.query:
|
||||
self.search.blockSignals(True)
|
||||
self.search.setText(self.controller.query)
|
||||
self.search.blockSignals(False)
|
||||
self.search.setPlaceholderText(self.controller.placeholder)
|
||||
|
||||
def _sync_rows(self) -> None:
|
||||
self.summary.setText(self.controller.summary)
|
||||
self.list.viewport().update()
|
||||
|
||||
def _sync_highlight(self) -> None:
|
||||
row = self.controller.highlight
|
||||
model = self.controller.model
|
||||
if row >= 0:
|
||||
# keep the section header of the highlighted row in view as well
|
||||
if row > 0 and model.row(row - 1).get("rowType") == "header":
|
||||
self.list.scrollTo(model.index(row - 1, 0))
|
||||
self.list.scrollTo(model.index(row, 0))
|
||||
self.list.viewport().update()
|
||||
|
||||
def showEvent(self, event): # pylint: disable=invalid-name
|
||||
super().showEvent(event)
|
||||
self.list.doItemsLayout()
|
||||
self._sync_highlight()
|
||||
|
||||
def _sync_detail(self) -> None:
|
||||
detail = self.controller.detail
|
||||
tokens = self.delegate.tokens
|
||||
self.footer.setVisible(bool(detail))
|
||||
if not detail:
|
||||
return
|
||||
self.detail_icon.setPixmap(
|
||||
cached_icon(detail.get("icon") or "sensors", tokens.fg_muted, 16)
|
||||
)
|
||||
self.detail_title.setText(detail.get("title", ""))
|
||||
parts = [part for part in (detail.get("subtitle"), detail.get("details")) if part]
|
||||
self.detail_sub.setText(" · ".join(parts))
|
||||
self.detail_sub.setVisible(bool(parts))
|
||||
state = detail.get("valueState")
|
||||
if state == "live" and detail.get("value"):
|
||||
stamp = detail.get("valueTime")
|
||||
self.detail_value.setText(detail["value"] + (f" · {stamp}" if stamp else ""))
|
||||
color = tokens.success
|
||||
elif state == "loading":
|
||||
self.detail_value.setText("reading…")
|
||||
color = tokens.fg_subtle
|
||||
else:
|
||||
self.detail_value.setText("no value yet")
|
||||
color = tokens.fg_subtle
|
||||
self.detail_value.setStyleSheet(
|
||||
f"color: {color.name()}; font-size: 12px; font-family: monospace; background: none;"
|
||||
)
|
||||
|
||||
# ---- events -------------------------------------------------------------------------------
|
||||
|
||||
def eventFilter(self, watched, event): # pylint: disable=invalid-name
|
||||
if watched is self.list.viewport() and event.type() == QEvent.Type.MouseMove:
|
||||
index = self.list.indexAt(event.position().toPoint())
|
||||
if index.isValid():
|
||||
self.controller.set_highlight(index.row())
|
||||
return super().eventFilter(watched, event)
|
||||
|
||||
def resizeEvent(self, event): # pylint: disable=invalid-name
|
||||
super().resizeEvent(event)
|
||||
path = QPainterPath()
|
||||
path.addRoundedRect(QRectF(self.rect()), 10, 10)
|
||||
self.setMask(QRegion(path.toFillPolygon().toPolygon()))
|
||||
|
||||
|
||||
class DevicePicker(DevicePickerBase):
|
||||
"""Device combobox with a searchable, grouped popup built from QWidgets.
|
||||
|
||||
Drop-in replacement for :class:`DeviceComboBox`; see :class:`DevicePickerBase`.
|
||||
"""
|
||||
|
||||
ICON_NAME = "manage_search"
|
||||
|
||||
def _create_popup(self):
|
||||
return PickerPopupWidget(self)
|
||||
|
||||
|
||||
class SignalPicker(SignalPickerBase):
|
||||
"""Signal combobox with a searchable, grouped popup built from QWidgets.
|
||||
|
||||
Drop-in replacement for :class:`SignalComboBox`; see :class:`SignalPickerBase`.
|
||||
"""
|
||||
|
||||
ICON_NAME = "manage_search"
|
||||
|
||||
def _create_popup(self):
|
||||
return PickerPopupWidget(self)
|
||||
@@ -0,0 +1,288 @@
|
||||
import QtQuick
|
||||
import QtQuick.Controls.Basic
|
||||
import QtQuick.Layouts
|
||||
import BecUi
|
||||
|
||||
// Searchable, grouped device/signal list with a live-value footer.
|
||||
// All state lives in the Python PickerController; `backend` forwards it.
|
||||
Rectangle {
|
||||
id: root
|
||||
required property var backend
|
||||
|
||||
color: theme.card
|
||||
radius: 10
|
||||
border.color: theme.border
|
||||
border.width: 1
|
||||
|
||||
function badgeTone(badge) {
|
||||
switch (badge) {
|
||||
case "monitored":
|
||||
case "hinted": return theme.primary
|
||||
case "async": return theme.highlight
|
||||
case "on_request": return theme.warning
|
||||
case "continuous":
|
||||
case "normal": return theme.success
|
||||
default: return theme.fgMuted
|
||||
}
|
||||
}
|
||||
|
||||
Connections {
|
||||
target: root.backend
|
||||
function onFocusRequested() { search.forceActiveFocus(); search.cursorPosition = search.text.length }
|
||||
function onHighlightChanged() {
|
||||
const row = root.backend.highlight
|
||||
if (row < 0) return
|
||||
// keep the section header of the highlighted row in view as well
|
||||
if (row > 0) list.positionViewAtIndex(row - 1, ListView.Contain)
|
||||
list.positionViewAtIndex(row, ListView.Contain)
|
||||
}
|
||||
}
|
||||
|
||||
ColumnLayout {
|
||||
anchors.fill: parent
|
||||
anchors.margins: 1
|
||||
spacing: 0
|
||||
|
||||
// ---- search -------------------------------------------------------------------
|
||||
RowLayout {
|
||||
Layout.fillWidth: true
|
||||
Layout.leftMargin: 12
|
||||
Layout.rightMargin: 12
|
||||
Layout.topMargin: 7
|
||||
Layout.bottomMargin: 7
|
||||
spacing: 8
|
||||
Icon { name: "search"; size: 18; color: theme.fgMuted }
|
||||
TextField {
|
||||
id: search
|
||||
Layout.fillWidth: true
|
||||
text: root.backend.query
|
||||
placeholderText: root.backend.placeholder
|
||||
placeholderTextColor: theme.fgSubtle
|
||||
color: theme.fg
|
||||
selectionColor: theme.primary
|
||||
selectedTextColor: theme.onPrimary
|
||||
font.pixelSize: 14
|
||||
leftPadding: 0
|
||||
implicitHeight: 32
|
||||
verticalAlignment: TextInput.AlignVCenter
|
||||
background: null
|
||||
selectByMouse: true
|
||||
onTextEdited: root.backend.setQuery(text)
|
||||
Keys.onPressed: (event) => {
|
||||
switch (event.key) {
|
||||
case Qt.Key_Down:
|
||||
case Qt.Key_Tab: root.backend.move(1); break
|
||||
case Qt.Key_Up:
|
||||
case Qt.Key_Backtab: root.backend.move(-1); break
|
||||
case Qt.Key_PageDown: root.backend.page(1); break
|
||||
case Qt.Key_PageUp: root.backend.page(-1); break
|
||||
case Qt.Key_Return:
|
||||
case Qt.Key_Enter: root.backend.accept(-1); break
|
||||
case Qt.Key_Escape: root.backend.dismiss(); break
|
||||
default: return
|
||||
}
|
||||
event.accepted = true
|
||||
}
|
||||
}
|
||||
Text {
|
||||
text: root.backend.summary
|
||||
color: theme.fgSubtle
|
||||
font.pixelSize: 12
|
||||
}
|
||||
}
|
||||
Rectangle { Layout.fillWidth: true; height: 1; color: theme.border }
|
||||
|
||||
// ---- rows -----------------------------------------------------------------------
|
||||
ListView {
|
||||
id: list
|
||||
Layout.fillWidth: true
|
||||
Layout.fillHeight: true
|
||||
clip: true
|
||||
model: root.backend.rows
|
||||
boundsBehavior: Flickable.StopAtBounds
|
||||
reuseItems: true
|
||||
ScrollBar.vertical: ScrollBar { policy: ScrollBar.AsNeeded; width: 8 }
|
||||
|
||||
delegate: Item {
|
||||
id: row
|
||||
required property int index
|
||||
required property string rowType
|
||||
required property string titleRich
|
||||
required property string title
|
||||
required property string subtitle
|
||||
required property string badge
|
||||
required property string icon
|
||||
required property bool isCurrent
|
||||
required property int count
|
||||
readonly property bool highlighted: index === root.backend.highlight
|
||||
|
||||
width: ListView.view.width
|
||||
height: rowType === "header" ? 28 : rowType === "empty" ? 96 : 34
|
||||
|
||||
// header
|
||||
RowLayout {
|
||||
visible: row.rowType === "header"
|
||||
anchors.fill: parent
|
||||
anchors.leftMargin: 12
|
||||
anchors.rightMargin: 12
|
||||
anchors.topMargin: 4
|
||||
spacing: 6
|
||||
Icon { visible: row.icon !== ""; name: row.icon; size: 14; color: theme.fgSubtle }
|
||||
Text {
|
||||
text: row.title.toUpperCase()
|
||||
color: theme.fgSubtle
|
||||
font.pixelSize: 11
|
||||
font.weight: Font.DemiBold
|
||||
font.letterSpacing: 0.6
|
||||
}
|
||||
Text {
|
||||
visible: row.count > 0
|
||||
text: row.count
|
||||
color: theme.fgSubtle
|
||||
font.pixelSize: 11
|
||||
font.letterSpacing: 0.6
|
||||
}
|
||||
Item { Layout.fillWidth: true }
|
||||
}
|
||||
|
||||
// empty state
|
||||
Column {
|
||||
visible: row.rowType === "empty"
|
||||
anchors.centerIn: parent
|
||||
spacing: 8
|
||||
Icon { anchors.horizontalCenter: parent.horizontalCenter; name: row.icon; size: 24; color: theme.fgSubtle }
|
||||
Text { text: row.title; color: theme.fgMuted; font.pixelSize: 13 }
|
||||
}
|
||||
|
||||
// item
|
||||
Rectangle {
|
||||
visible: row.rowType === "item"
|
||||
anchors.fill: parent
|
||||
anchors.leftMargin: 4
|
||||
anchors.rightMargin: 4
|
||||
anchors.topMargin: 1
|
||||
anchors.bottomMargin: 1
|
||||
radius: 6
|
||||
color: row.highlighted ? theme.hover : "transparent"
|
||||
|
||||
Rectangle {
|
||||
visible: row.isCurrent
|
||||
width: 3; radius: 1.5
|
||||
height: parent.height - 14
|
||||
anchors.verticalCenter: parent.verticalCenter
|
||||
color: theme.primary
|
||||
}
|
||||
RowLayout {
|
||||
anchors.fill: parent
|
||||
anchors.leftMargin: 10
|
||||
anchors.rightMargin: 8
|
||||
spacing: 8
|
||||
Icon {
|
||||
name: row.icon
|
||||
size: 16
|
||||
color: row.isCurrent ? theme.primary : theme.fgMuted
|
||||
}
|
||||
Item {
|
||||
Layout.fillWidth: true
|
||||
Layout.fillHeight: true
|
||||
Text {
|
||||
id: titleText
|
||||
anchors.verticalCenter: parent.verticalCenter
|
||||
width: Math.min(implicitWidth, parent.width)
|
||||
text: row.titleRich
|
||||
textFormat: Text.StyledText
|
||||
color: theme.fg
|
||||
font.pixelSize: 13
|
||||
font.weight: row.isCurrent ? Font.Medium : Font.Normal
|
||||
elide: Text.ElideRight
|
||||
}
|
||||
Text {
|
||||
anchors.verticalCenter: parent.verticalCenter
|
||||
x: titleText.width + 8
|
||||
width: Math.max(0, parent.width - x)
|
||||
visible: width > 24
|
||||
text: row.subtitle
|
||||
color: theme.fgSubtle
|
||||
font.pixelSize: 12
|
||||
elide: Text.ElideRight
|
||||
}
|
||||
}
|
||||
Rectangle {
|
||||
visible: row.badge !== ""
|
||||
implicitWidth: badgeLabel.implicitWidth + 14
|
||||
implicitHeight: 18
|
||||
radius: 9
|
||||
readonly property color tone: root.badgeTone(row.badge)
|
||||
color: Qt.rgba(tone.r, tone.g, tone.b, 0.18)
|
||||
Text {
|
||||
id: badgeLabel
|
||||
anchors.centerIn: parent
|
||||
text: row.badge
|
||||
color: parent.tone
|
||||
font.pixelSize: 11
|
||||
font.weight: Font.Medium
|
||||
}
|
||||
}
|
||||
}
|
||||
MouseArea {
|
||||
anchors.fill: parent
|
||||
hoverEnabled: true
|
||||
cursorShape: Qt.PointingHandCursor
|
||||
onEntered: root.backend.hover(row.index)
|
||||
onClicked: root.backend.accept(row.index)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- footer ---------------------------------------------------------------------
|
||||
Rectangle { Layout.fillWidth: true; height: 1; color: theme.border; visible: footer.visible }
|
||||
ColumnLayout {
|
||||
id: footer
|
||||
readonly property var detail: root.backend.detail
|
||||
visible: detail.title !== undefined
|
||||
Layout.fillWidth: true
|
||||
Layout.leftMargin: 12
|
||||
Layout.rightMargin: 12
|
||||
Layout.topMargin: 8
|
||||
Layout.bottomMargin: 8
|
||||
spacing: 2
|
||||
RowLayout {
|
||||
Layout.fillWidth: true
|
||||
spacing: 8
|
||||
Icon { name: footer.detail.icon || "sensors"; size: 16; color: theme.fgMuted }
|
||||
Text {
|
||||
Layout.fillWidth: true
|
||||
text: footer.detail.title || ""
|
||||
color: theme.fg
|
||||
font.pixelSize: 13
|
||||
font.weight: Font.DemiBold
|
||||
elide: Text.ElideRight
|
||||
}
|
||||
Text {
|
||||
readonly property bool live: footer.detail.valueState === "live" && footer.detail.value !== ""
|
||||
text: live ? footer.detail.value + (footer.detail.valueTime ? " · " + footer.detail.valueTime : "")
|
||||
: footer.detail.valueState === "loading" ? "reading…" : "no value yet"
|
||||
color: live ? theme.success : theme.fgSubtle
|
||||
font.pixelSize: 12
|
||||
font.family: "monospace"
|
||||
}
|
||||
}
|
||||
Text {
|
||||
readonly property string line: [footer.detail.subtitle, footer.detail.details].filter(p => p).join(" · ")
|
||||
visible: line !== ""
|
||||
Layout.fillWidth: true
|
||||
text: line
|
||||
color: theme.fgMuted
|
||||
font.pixelSize: 12
|
||||
elide: Text.ElideRight
|
||||
}
|
||||
Text {
|
||||
Layout.topMargin: 4
|
||||
text: "↑↓ move ↵ select esc close"
|
||||
color: theme.fgSubtle
|
||||
font.pixelSize: 11
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
"""Benchmarks of the device pickers against DeviceComboBox with a large device list."""
|
||||
|
||||
# pylint: disable=redefined-outer-name, missing-function-docstring
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
from bec_widgets.tests.client_mocks import mocked_client # noqa: F401
|
||||
from bec_widgets.tests.fake_devices import FakeDevice, FakePositioner
|
||||
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
|
||||
from bec_widgets.widgets.control.device_input.picker.picker_qml import DevicePickerQML
|
||||
from bec_widgets.widgets.control.device_input.picker.picker_qwidget import DevicePicker
|
||||
|
||||
WIDGETS = [DeviceComboBox, DevicePicker, DevicePickerQML]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def many_devices(mocked_client):
|
||||
devices = []
|
||||
for index in range(300):
|
||||
devices.append(FakePositioner(f"mot{index:03d}", limits=[-1, 1], read_value=1.0))
|
||||
devices.append(FakeDevice(f"det{index:03d}"))
|
||||
mocked_client.device_manager.add_devices(devices)
|
||||
yield mocked_client
|
||||
|
||||
|
||||
def _close(widgets):
|
||||
for widget in widgets:
|
||||
widget.close()
|
||||
widget.deleteLater()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("widget_cls", WIDGETS, ids=lambda cls: cls.__name__)
|
||||
def test_create_device_input(benchmark, qtbot, many_devices, widget_cls):
|
||||
"""Create one device input with about 600 devices."""
|
||||
created = []
|
||||
|
||||
def create():
|
||||
created.append(widget_cls(client=many_devices))
|
||||
|
||||
benchmark(create)
|
||||
assert created[-1].devices
|
||||
_close(created)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("widget_cls", WIDGETS, ids=lambda cls: cls.__name__)
|
||||
def test_device_update_refreshes_twenty_inputs(benchmark, qtbot, many_devices, widget_cls):
|
||||
"""Deliver one device configuration update to 20 device inputs, as on a config reload."""
|
||||
widgets = [widget_cls(client=many_devices) for _ in range(20)]
|
||||
|
||||
def update():
|
||||
thread = threading.Thread(
|
||||
target=lambda: [widget.on_device_update("reload", {}) for widget in widgets]
|
||||
)
|
||||
thread.start()
|
||||
thread.join()
|
||||
QApplication.processEvents()
|
||||
|
||||
benchmark(update)
|
||||
_close(widgets)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("widget_cls", [DevicePicker, DevicePickerQML], ids=lambda c: c.__name__)
|
||||
def test_search_popup_keystroke(benchmark, qtbot, many_devices, widget_cls):
|
||||
"""Filter the open popup by one more character."""
|
||||
widget = widget_cls(client=many_devices)
|
||||
widget.open_picker()
|
||||
controller = widget.picker_controller
|
||||
queries = iter(["m", "mo", "mot", "mot1", "mot12", ""] * 1000)
|
||||
|
||||
benchmark(lambda: controller.set_query(next(queries)))
|
||||
widget.hidePopup()
|
||||
_close([widget])
|
||||
@@ -0,0 +1,377 @@
|
||||
"""Tests for the searchable device and signal pickers (QWidget and QML popups)."""
|
||||
|
||||
# pylint: disable=redefined-outer-name, protected-access, missing-function-docstring
|
||||
import time
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
from bec_lib import messages
|
||||
from bec_lib.device import ReadoutPriority
|
||||
from bec_lib.endpoints import MessageEndpoints
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
from bec_widgets.tests.fake_devices import FakeDevice
|
||||
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
|
||||
BECDeviceFilter,
|
||||
DeviceComboBox,
|
||||
)
|
||||
from bec_widgets.widgets.control.device_input.picker import picker_common, picker_model
|
||||
from bec_widgets.widgets.control.device_input.picker.picker_model import (
|
||||
PickerController,
|
||||
PickerEntry,
|
||||
format_value,
|
||||
rank_entry,
|
||||
)
|
||||
from bec_widgets.widgets.control.device_input.picker.picker_qml import (
|
||||
DevicePickerQML,
|
||||
PickerPopupQml,
|
||||
SignalPickerQML,
|
||||
)
|
||||
from bec_widgets.widgets.control.device_input.picker.picker_qwidget import (
|
||||
DevicePicker,
|
||||
PickerPopupWidget,
|
||||
SignalPicker,
|
||||
)
|
||||
|
||||
from .conftest import create_widget
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clear_recent():
|
||||
picker_model.clear_recent_picks()
|
||||
yield
|
||||
picker_model.clear_recent_picks()
|
||||
|
||||
|
||||
@pytest.fixture(params=[DevicePicker, DevicePickerQML])
|
||||
def device_picker(request, qtbot, mocked_client):
|
||||
yield create_widget(qtbot, request.param, client=mocked_client)
|
||||
|
||||
|
||||
@pytest.fixture(params=[SignalPicker, SignalPickerQML])
|
||||
def signal_picker(request, qtbot, mocked_client):
|
||||
yield create_widget(qtbot, request.param, client=mocked_client, device="samx")
|
||||
|
||||
|
||||
def _titles(controller: PickerController) -> list[tuple[str, str]]:
|
||||
model = controller.model
|
||||
return [(model.row(r)["rowType"], model.row(r)["title"]) for r in range(model.rowCount())]
|
||||
|
||||
|
||||
def _wait_closed(qtbot, picker):
|
||||
qtbot.waitUntil(lambda: not picker.is_popup_open())
|
||||
|
||||
|
||||
# ---- pure logic ---------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_rank_entry_orders_prefix_word_substring_fuzzy():
|
||||
def rank(query, title, **kwargs):
|
||||
score = rank_entry(query, PickerEntry(value=title, title=title, **kwargs))
|
||||
return None if score is None else score[0]
|
||||
|
||||
assert rank("sam", "samx") == 0
|
||||
assert rank("x", "slit1_xc") == 1
|
||||
assert rank("pm", "bpm4i") == 2
|
||||
assert rank("smx", "samx") == 3
|
||||
assert rank("motor", "samx", subtitle="EpicsMotor") == 4
|
||||
assert rank("zzz", "samx") is None
|
||||
assert rank_entry("", PickerEntry(value="a", title="a")) == (0, -1, 0)
|
||||
assert rank_entry("BPM", PickerEntry(value="x", title="gauss_bpm")) == (1, 6, 3)
|
||||
|
||||
|
||||
def test_format_value():
|
||||
assert format_value(None) == ""
|
||||
assert format_value(True) == "true"
|
||||
assert format_value(3) == "3"
|
||||
assert format_value(1.23456789) == "1.23457"
|
||||
assert format_value([1, 2]) == "list [2]"
|
||||
assert format_value("x" * 50).endswith("…")
|
||||
|
||||
|
||||
def test_controller_groups_searches_and_navigates(qtbot):
|
||||
entries = [
|
||||
PickerEntry(value="samx", title="samx", group="Positioners"),
|
||||
PickerEntry(value="samy", title="samy", group="Positioners"),
|
||||
PickerEntry(value="bpm4i", title="bpm4i", group="Devices"),
|
||||
]
|
||||
controller = PickerController("devices")
|
||||
controller.open(entries, current="samy")
|
||||
assert _titles(controller) == [
|
||||
("header", "Positioners"),
|
||||
("item", "samx"),
|
||||
("item", "samy"),
|
||||
("header", "Devices"),
|
||||
("item", "bpm4i"),
|
||||
]
|
||||
# the current value is highlighted and marked
|
||||
assert controller.highlight == 2
|
||||
assert controller.model.row(2)["isCurrent"]
|
||||
assert controller.summary == "3 devices"
|
||||
|
||||
# navigation skips headers and wraps around
|
||||
controller.move_highlight(1)
|
||||
assert controller.entry_at(controller.highlight).value == "bpm4i"
|
||||
controller.move_highlight(1)
|
||||
assert controller.entry_at(controller.highlight).value == "samx"
|
||||
controller.move_highlight(-1)
|
||||
assert controller.entry_at(controller.highlight).value == "bpm4i"
|
||||
|
||||
controller.set_query("y")
|
||||
assert _titles(controller) == [("item", "samy")]
|
||||
assert controller.summary == "1 of 3"
|
||||
assert "<b>" in controller.model.row(0)["titleRich"]
|
||||
|
||||
controller.set_query("nothing")
|
||||
assert _titles(controller)[0][0] == "empty"
|
||||
assert controller.entry_at(0) is None
|
||||
assert not controller.accept()
|
||||
|
||||
controller.set_query("bpm")
|
||||
with qtbot.waitSignal(controller.picked) as blocker:
|
||||
assert controller.accept()
|
||||
assert blocker.args == ["bpm4i"]
|
||||
controller.close()
|
||||
|
||||
|
||||
def test_controller_lists_recent_first():
|
||||
entries = [PickerEntry(value=f"dev{i}", title=f"dev{i}", group="Devices") for i in range(12)]
|
||||
controller = PickerController("devices")
|
||||
controller.open(entries, recent=["dev7", "unknown"])
|
||||
titles = _titles(controller)
|
||||
assert titles[:3] == [("header", "Recent"), ("item", "dev7"), ("header", "Devices")]
|
||||
controller.close()
|
||||
|
||||
|
||||
# ---- device picker -------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_device_picker_lists_same_devices_as_combobox(qtbot, mocked_client, device_picker):
|
||||
combo = create_widget(qtbot, DeviceComboBox, client=mocked_client)
|
||||
assert device_picker.devices == combo.devices
|
||||
|
||||
for widget in (combo, device_picker):
|
||||
widget.set_device_filter(BECDeviceFilter.POSITIONER)
|
||||
assert device_picker.devices == combo.devices
|
||||
assert "samx" in device_picker.devices and "bpm4i" not in device_picker.devices
|
||||
|
||||
for widget in (combo, device_picker):
|
||||
widget.readout_monitored = False
|
||||
assert device_picker.devices == combo.devices
|
||||
|
||||
for widget in (combo, device_picker):
|
||||
widget.set_available_devices(["samx", "samy"])
|
||||
assert device_picker.devices == combo.devices == ["samx", "samy"]
|
||||
|
||||
|
||||
ASYNC_SIGNALS = [
|
||||
(
|
||||
"eiger",
|
||||
"eiger",
|
||||
{
|
||||
"obj_name": "eiger",
|
||||
"component_name": "eiger",
|
||||
"kind_str": "hinted",
|
||||
"signal_class": "AsyncSignal",
|
||||
},
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def async_signals(mocked_client):
|
||||
with mock.patch.object(
|
||||
mocked_client.device_manager, "get_bec_signals", return_value=ASYNC_SIGNALS, create=True
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
def test_device_picker_signal_class_and_write_access_filters(qtbot, mocked_client, async_signals):
|
||||
combo = create_widget(
|
||||
qtbot,
|
||||
DeviceComboBox,
|
||||
client=mocked_client,
|
||||
signal_class_filter=["AsyncSignal"],
|
||||
include_signals_with_write_access=True,
|
||||
)
|
||||
picker = create_widget(
|
||||
qtbot,
|
||||
DevicePicker,
|
||||
client=mocked_client,
|
||||
signal_class_filter=["AsyncSignal"],
|
||||
include_signals_with_write_access=True,
|
||||
)
|
||||
assert picker.devices == combo.devices
|
||||
assert "eiger" in picker.devices
|
||||
|
||||
|
||||
def test_readout_filters_refresh_once(qtbot, mocked_client):
|
||||
with mock.patch.object(
|
||||
picker_common, "device_snapshot", wraps=picker_common.device_snapshot
|
||||
) as snapshot:
|
||||
picker = create_widget(qtbot, DevicePicker, client=mocked_client)
|
||||
reads_in_init = [c for c in snapshot.call_args_list if c.kwargs.get("force") is not None]
|
||||
assert len(reads_in_init) == 1
|
||||
snapshot.reset_mock()
|
||||
picker.set_readout_priority_filter([ReadoutPriority.MONITORED, ReadoutPriority.BASELINE])
|
||||
assert len([c for c in snapshot.call_args_list if "force" in c.kwargs]) == 1
|
||||
|
||||
|
||||
def test_unchanged_device_list_does_not_reemit(qtbot, device_picker):
|
||||
device_picker.set_device("samx")
|
||||
with qtbot.assertNotEmitted(device_picker.device_selected):
|
||||
device_picker.update_devices_from_filters()
|
||||
|
||||
|
||||
def test_device_update_lists_new_device(qtbot, mocked_client, device_picker):
|
||||
assert "new_det" not in device_picker.devices
|
||||
mocked_client.device_manager.add_devices([FakeDevice("new_det")])
|
||||
device_picker.on_device_update("add", {})
|
||||
qtbot.waitUntil(lambda: "new_det" in device_picker.devices)
|
||||
|
||||
|
||||
def test_invalid_text_is_painted_not_restyled(device_picker):
|
||||
sheet = device_picker.styleSheet()
|
||||
device_picker.setCurrentText("not_a_device")
|
||||
assert not device_picker.is_valid_input
|
||||
assert device_picker.shows_invalid
|
||||
device_picker.setCurrentText("")
|
||||
assert not device_picker.shows_invalid
|
||||
device_picker.setCurrentText("samx")
|
||||
assert device_picker.is_valid_input and not device_picker.shows_invalid
|
||||
assert device_picker.styleSheet() == sheet
|
||||
device_picker.setCurrentText("not_a_device")
|
||||
device_picker.setEnabled(False)
|
||||
assert not device_picker.shows_invalid
|
||||
|
||||
|
||||
def test_pick_from_popup_selects_device(qtbot, device_picker):
|
||||
device_picker.showPopup()
|
||||
assert device_picker.is_popup_open()
|
||||
popup_cls = PickerPopupQml if isinstance(device_picker, DevicePickerQML) else PickerPopupWidget
|
||||
assert isinstance(device_picker.picker_popup, popup_cls)
|
||||
controller = device_picker.picker_controller
|
||||
assert ("header", "Positioners") in _titles(controller)
|
||||
|
||||
controller.set_query("samy")
|
||||
with qtbot.waitSignal(device_picker.device_selected) as blocker:
|
||||
controller.accept()
|
||||
assert blocker.args == ["samy"]
|
||||
assert device_picker.currentText() == "samy"
|
||||
_wait_closed(qtbot, device_picker)
|
||||
assert picker_model.recent_picks("devices") == ["samy"]
|
||||
|
||||
|
||||
def test_typing_opens_popup_with_query_and_escape_restores(qtbot, device_picker):
|
||||
device_picker.set_device("samx")
|
||||
device_picker.lineEdit().textEdited.emit("bpm")
|
||||
assert device_picker.is_popup_open()
|
||||
controller = device_picker.picker_controller
|
||||
assert controller.query == "bpm"
|
||||
assert all("bpm" in title for kind, title in _titles(controller) if kind == "item")
|
||||
device_picker.setCurrentText("bpm")
|
||||
controller.dismiss()
|
||||
_wait_closed(qtbot, device_picker)
|
||||
assert device_picker.currentText() == "samx"
|
||||
|
||||
|
||||
def test_closing_popup_by_clicking_outside_reverts_typed_text(qtbot, device_picker):
|
||||
device_picker.set_device("samx")
|
||||
device_picker.setCurrentText("bp")
|
||||
device_picker.lineEdit().textEdited.emit("bp")
|
||||
assert device_picker.is_popup_open()
|
||||
device_picker.picker_popup.hide()
|
||||
assert device_picker.currentText() == "samx"
|
||||
|
||||
|
||||
def test_footer_shows_cached_and_live_value(qtbot, mocked_client, device_picker):
|
||||
now = time.time()
|
||||
mocked_client.connector.set_and_publish(
|
||||
MessageEndpoints.device_readback("samx"),
|
||||
messages.DeviceMessage(signals={"samx": {"value": 2.5, "timestamp": now}}, metadata={}),
|
||||
)
|
||||
device_picker.set_device("samx")
|
||||
device_picker.showPopup()
|
||||
controller = device_picker.picker_controller
|
||||
assert controller.detail["title"] == "samx"
|
||||
qtbot.waitUntil(lambda: controller.detail.get("valueState") == "live", timeout=2000)
|
||||
assert controller.detail["value"] == "2.5"
|
||||
assert device_picker._live_entry.device == "samx"
|
||||
|
||||
device_picker._on_live_message({"signals": {"samx": {"value": 3.0, "timestamp": now}}}, {})
|
||||
assert controller.detail["value"] == "3"
|
||||
|
||||
device_picker.hidePopup()
|
||||
_wait_closed(qtbot, device_picker)
|
||||
assert device_picker._live_entry is None
|
||||
|
||||
|
||||
def test_popup_follows_theme(qtbot, device_picker):
|
||||
from bec_widgets.utils.colors import apply_theme
|
||||
|
||||
device_picker.showPopup()
|
||||
apply_theme("dark")
|
||||
QApplication.processEvents()
|
||||
device_picker.hidePopup()
|
||||
apply_theme("light")
|
||||
|
||||
|
||||
def test_qml_popup_loads_without_errors(qtbot, mocked_client):
|
||||
picker = create_widget(qtbot, DevicePickerQML, client=mocked_client)
|
||||
picker.showPopup()
|
||||
view = picker.picker_popup.view
|
||||
assert view.errors() == []
|
||||
assert view.rootObject() is not None
|
||||
picker.hidePopup()
|
||||
|
||||
|
||||
def test_cleanup_releases_popup(qtbot, mocked_client):
|
||||
picker = DevicePicker(client=mocked_client)
|
||||
qtbot.addWidget(picker)
|
||||
picker.showPopup()
|
||||
popup = picker.picker_popup
|
||||
picker.close()
|
||||
assert picker._picker_popup is None
|
||||
assert not popup.isVisible()
|
||||
|
||||
|
||||
# ---- signal picker -------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_signal_picker_groups_by_kind(signal_picker):
|
||||
signal_picker.showPopup()
|
||||
titles = _titles(signal_picker.picker_controller)
|
||||
headers = [title for kind, title in titles if kind == "header"]
|
||||
assert headers == ["Hinted", "Normal", "Config"]
|
||||
assert ("item", "samx (readback)") in titles
|
||||
signal_picker.hidePopup()
|
||||
|
||||
|
||||
def test_signal_picker_pick_sets_index_and_config(qtbot, signal_picker):
|
||||
signal_picker.showPopup()
|
||||
controller = signal_picker.picker_controller
|
||||
controller.set_query("velo")
|
||||
with qtbot.waitSignal(signal_picker.device_signal_changed):
|
||||
controller.accept()
|
||||
assert signal_picker.currentText() == "velocity"
|
||||
assert signal_picker.get_signal_name() == "samx_velocity"
|
||||
assert signal_picker.get_signal_config()["obj_name"] == "samx_velocity"
|
||||
_wait_closed(qtbot, signal_picker)
|
||||
|
||||
|
||||
def test_signal_picker_by_class_groups_by_device(qtbot, mocked_client, async_signals):
|
||||
picker = create_widget(
|
||||
qtbot, SignalPicker, client=mocked_client, signal_class_filter=["AsyncSignal"]
|
||||
)
|
||||
picker.showPopup()
|
||||
titles = _titles(picker.picker_controller)
|
||||
# a single group is listed without a header
|
||||
assert titles == [("item", "eiger")]
|
||||
entry = next(
|
||||
picker.picker_controller.entry_at(row)
|
||||
for row in range(len(titles))
|
||||
if titles[row] == ("item", "eiger")
|
||||
)
|
||||
assert entry.device == "eiger"
|
||||
assert entry.group == "eiger"
|
||||
picker.hidePopup()
|
||||
Reference in new issue
Block a user