Source code for QInstrument.lib.QInstrumentWidget
from pathlib import Path
import inspect
import logging
from qtpy import uic, QtWidgets, QtCore
from QInstrument.lib.QSerialInstrument import QSerialInstrument
from .Configure import Configure
from .QReconcileDialog import QReconcileDialog
from .lazy import find_fake_cls, values_differ
logger = logging.getLogger(__name__)
[docs]
class QInstrumentWidget(QtWidgets.QWidget):
'''Widget that auto-binds a Qt Designer UI to a QAbstractInstrument.
A named widget in the UI is linked to a registered device property
when their names match and the widget type appears in :attr:`wsetter`,
:attr:`wgetter`, and :attr:`wsignal`. User interaction with a linked
widget calls :meth:`device.set`; the device value is read back and
the widget is updated without re-triggering the signal.
On first show, saved settings are reconciled with the hardware state
via :class:`QReconcileDialog`, and the device is moved to a dedicated
worker thread so that serial I/O does not block the GUI. Settings are
saved on close.
Subclass this, declare :attr:`UIFILE`, and supply a device:
.. code-block:: python
class QDS345Widget(QInstrumentWidget):
UIFILE = 'DS345.ui'
def __init__(self, *args, **kwargs):
super().__init__(*args, device=QDS345(), **kwargs)
The ``.ui`` file is resolved relative to the subclass's source
directory, so it works regardless of the working directory.
Class Attributes
----------------
UIFILE : str | None
Filename of the Qt Designer ``.ui`` file. Must be set by each
subclass.
INSTRUMENT : type | None
Concrete instrument class to instantiate and search for when no
``device`` is supplied to ``__init__``. When set, the base class
calls ``INSTRUMENT().find()`` automatically so subclasses need not
override ``__init__`` solely to locate the device.
wsetter : dict[str, str]
Maps widget class name to its value-setter method name.
wgetter : dict[str, str]
Maps widget class name to its value-getter method name.
wsignal : dict[str, str]
Maps widget class name to the signal emitted on user interaction.
Signals
-------
propertyChanged(str, object)
Emitted after a linked widget updates the device, carrying
the property name and the new value.
'''
wsetter = {'QCheckBox': 'setChecked',
'QComboBox': 'setCurrentIndex',
'QDoubleSpinBox': 'setValue',
'QGroupBox': 'setChecked',
'QLabel': 'setText',
'QLineEdit': 'setText',
'QPushButton': 'setChecked',
'QRadioButton': 'setChecked',
'QSpinBox': 'setValue'}
wgetter = {'QCheckBox': 'isChecked',
'QComboBox': 'currentIndex',
'QDoubleSpinBox': 'value',
'QGroupBox': 'isChecked',
'QLabel': 'text',
'QLineEdit': 'text',
'QPushButton': 'isChecked',
'QRadioButton': 'isChecked',
'QSpinBox': 'value'}
wsignal = {'QCheckBox': 'toggled',
'QComboBox': 'currentIndexChanged',
'QDoubleSpinBox': 'valueChanged',
'QGroupBox': 'toggled',
'QLineEdit': 'editingFinished',
'QPushButton': 'toggled',
'QRadioButton': 'toggled',
'QSpinBox': 'valueChanged'}
UIFILE: str | None = None
INSTRUMENT: type | None = None
HARDWARE_DOMINANT: bool = False
propertyChanged = QtCore.Signal(str, object)
closeRequested = QtCore.Signal()
def __init__(self, *args, device=None, **kwargs) -> None:
super().__init__(*args, **kwargs)
self._device = None
self._configure = Configure()
self._restored = False
self._thread = None
uic.loadUi(self._uiPath(), self)
if device is None and self.INSTRUMENT is not None:
device = self.INSTRUMENT().find()
self.device = device
@QtCore.Property(object)
def device(self):
'''QAbstractInstrument: instrument bound to this widget.
Setting this property identifies matching properties and methods,
syncs the UI to current device values (if open), and connects
widget signals to the device. Setting to ``None`` is a no-op.
The widget is disabled if the device is not open.
'''
return self._device
@device.setter
def device(self, device):
if device is None:
return
self._device = device
self._identifyProperties()
if self._device.isOpen():
self._connectSignals()
self._syncProperties()
else:
self.setEnabled(False)
@QtCore.Property(list)
def properties(self) -> list[str]:
'''list[str]: device property names managed by this widget.'''
return self._properties
@QtCore.Property(list)
def methods(self) -> list[str]:
'''list[str]: device method names managed by this widget.'''
return self._methods
@QtCore.Property(dict)
def settings(self) -> dict:
'''dict: current values of all managed properties.
Getting reads each linked widget. Setting calls :meth:`set`
for each key present in the supplied dict.
'''
return {key: self.get(key) for key in self.properties}
@settings.setter
def settings(self, settings):
for key, value in settings.items():
self.set(key, value)
[docs]
def get(self, key: str):
'''Return the current value of a named widget.
Parameters
----------
key : str
Name of the property whose widget value to read.
Returns
-------
object or None
Current widget value, or ``None`` if *key* is not found.
'''
widget = self.__dict__.get(key)
if isinstance(widget, QtWidgets.QWidget):
getter = self._wmethod(widget, self.wgetter)
if getter is None:
return None
return getter()
logger.error(f'Unknown property {key}')
return None
[docs]
def set(self, key: str, value=None) -> None:
'''Set the value of a named widget.
When *value* is provided, sets the widget directly; the widget
then emits its signal, which propagates the change to the device.
When *value* is ``None``, requests the current value from the
device via :meth:`device.get`; the result arrives via
:attr:`device.propertyValue` and is applied by
:meth:`_onPropertyValue` with signals blocked.
Parameters
----------
key : str
Name of the property to set.
value : bool | int | float | str | None, optional
Value to apply. ``None`` (default) syncs the widget from
the device.
'''
widget = self.__dict__.get(key)
if not isinstance(widget, QtWidgets.QWidget):
logger.error(f'Unknown property {key}')
return
if value is None:
self.device.get(key)
return
setter = self._wmethod(widget, self.wsetter)
if setter is None:
logger.debug(f'No setter for widget type of {key!r}; skipping')
return
try:
setter(value)
except Exception as ex:
logger.error(f'Could not set {key} to {value}: {ex}')
def _wmethod(self,
widget: QtWidgets.QWidget,
method: dict) -> 'callable | None':
'''Return the bound method named by *method* for *widget*'s type.
Returns ``None`` if the widget's class is not in *method*, so
callers can skip unknown widget types without raising.
Parameters
----------
widget : QWidget
The target widget.
method : dict
One of :attr:`wsetter`, :attr:`wgetter`, or :attr:`wsignal`,
mapping widget class name to method name.
'''
typeName = widget.metaObject().className()
name = method.get(typeName)
if name is None:
return None
return getattr(widget, name)
@classmethod
def _uiPath(cls) -> Path:
'''Return the absolute path to this class's UI file.
Resolves :attr:`UIFILE` relative to the directory of the class
in the MRO that defines it, so subclasses that inherit
:attr:`UIFILE` without overriding it resolve correctly.
'''
for klass in cls.__mro__:
if 'UIFILE' in klass.__dict__:
return Path(inspect.getfile(klass)).parent / klass.UIFILE
raise AttributeError(f'{cls.__name__} has no UIFILE defined')
def _identifyProperties(self) -> None:
'''Populate :attr:`_properties` and :attr:`_methods`.
Intersects the set of UI widget names with the device's
registered property and method names.
'''
uwidgets = {name for name, obj in self.__dict__.items()
if isinstance(obj, QtWidgets.QWidget)}
dproperties = set(self.device.properties)
dmethods = set(self.device.methods)
self._properties = list(uwidgets & dproperties)
self._methods = list(uwidgets & dmethods)
def _syncProperties(self) -> None:
'''Request current device values for all linked properties.
Calls :meth:`device.get` for each property, which emits
:attr:`propertyValue` and updates the corresponding widget
via :meth:`_onPropertyValue`. Works whether the device is
on the main thread (direct connection, synchronous) or a
worker thread (queued connection, asynchronous).
'''
for prop in self.properties:
self.device.get(prop)
@QtCore.Slot(str, object)
def _onPropertyValue(self, name: str, value: object) -> None:
'''Update the widget for *name* with *value*, blocking its signal.
Connected to :attr:`device.propertyValue` so that values
arriving from either synchronous or queued :meth:`device.get`
calls are applied without triggering a round-trip back to the
device.
Parameters
----------
name : str
Registered property name.
value : object
New value from the device.
'''
if name not in self._properties:
return
widget = self.__dict__.get(name)
if not isinstance(widget, QtWidgets.QWidget):
return
setter = self._wmethod(widget, self.wsetter)
if setter is None:
return
with QtCore.QSignalBlocker(widget):
try:
setter(value)
except Exception as ex:
logger.error(f'Could not set {name} to {value}: {ex}')
def _connectSignals(self) -> None:
'''Connect linked widget signals to the device and propertyChanged.
Connects :attr:`device.propertyValue` to :meth:`_onPropertyValue`
so that values emitted by the device (from either direct or queued
:meth:`device.get` calls) update the corresponding widget.
Properties with a ``debounce`` metadata value are connected
through a single-shot :class:`QTimer` so that rapid widget
changes (e.g. spinbox scrolling) are coalesced: only the final
value after the debounce interval elapses is sent to the device.
'''
self.device.propertyValue.connect(self._onPropertyValue)
for prop in self.properties:
widget = getattr(self, prop)
signal = self._wmethod(widget, self.wsignal)
if signal is None:
continue
debounce_ms = self.device.propertyMeta(prop).get('debounce', 0)
if debounce_ms:
self._connectDebounced(prop, signal, debounce_ms)
else:
signal.connect(self._setDeviceProperty)
for method in self.methods:
widget = getattr(self, method)
if isinstance(widget, QtWidgets.QPushButton):
widget.clicked.connect(
lambda m=method: self.device.execute(m))
def _connectDebounced(
self,
prop: str,
signal: QtCore.Signal,
debounce_ms: int
) -> None:
'''Connect *signal* to :meth:`_applyProperty` via a debounce timer.
Each call creates a single-shot :class:`QTimer`. Every signal
emission stores the latest value and restarts the timer; the
device is only updated when the timer fires (i.e. when the user
pauses for *debounce_ms* milliseconds).
Parameters
----------
prop : str
Property name passed to :meth:`_applyProperty`.
signal : QtCore.Signal
The widget signal to debounce.
debounce_ms : int
Quiet period in milliseconds before the device is updated.
'''
timer = QtCore.QTimer(self)
timer.setSingleShot(True)
timer.setInterval(debounce_ms)
pending = [None]
def on_change(value):
pending[0] = value
timer.start()
def on_timeout():
self._applyProperty(prop, pending[0])
signal.connect(on_change)
timer.timeout.connect(on_timeout)
def _applyProperty(
self, name: str, value: bool | int | float | str
) -> None:
'''Send *value* to the device for property *name*.
Called by both :meth:`_setDeviceProperty` (direct path) and the
debounce timer timeout (rate-limited path).
Parameters
----------
name : str
Registered property name.
value : bool | int | float | str
New value to send to the device.
'''
if name in self._properties:
logger.debug(f'Setting device: {name}: {value}')
self.device.set(name, value)
self.waitForDevice()
self.propertyChanged.emit(name, value)
@QtCore.Slot(bool)
@QtCore.Slot(int)
@QtCore.Slot(float)
@QtCore.Slot(str)
def _setDeviceProperty(self, value: bool | int | float | str) -> None:
'''Slot connected to non-debounced widget signals.
Reads the sender's object name and delegates to
:meth:`_applyProperty`.
'''
self._applyProperty(self.sender().objectName(), value)
[docs]
def waitForDevice(self) -> None:
'''Block until the device has completed the most recent change.
Called by :meth:`_setDeviceProperty` after every device write.
The base implementation is a no-op; subclasses should override
this if the instrument requires a settling delay.
'''
pass
[docs]
def showEvent(self, event) -> None:
'''Reconcile device settings on first show.
On the first time the widget is shown, schedules
:meth:`_firstShow` via a zero-delay timer so that reconciliation
runs after Qt has finished processing the show event. This
avoids a crash caused by opening a modal dialog (nested event
loop) from inside an event handler. Subsequent show events are
passed through without reconciling.
'''
device_open = self._device is not None and self._device.isOpen()
if not self._restored and device_open:
self._restored = True
QtCore.QTimer.singleShot(0, self._firstShow)
super().showEvent(event)
@QtCore.Slot()
def _firstShow(self) -> None:
'''Run first-show reconciliation after the event loop is idle.
Restores saved settings and syncs the UI while the device is still
on the main thread, then moves the device to a dedicated worker
thread. Polling is not started automatically; call
:meth:`startPolling` explicitly or connect a control to it when
continuous updates are needed.
'''
self._restoreSettings()
self._syncProperties()
self._startDeviceThread()
def _startDeviceThread(self) -> None:
'''Move the device into a dedicated worker thread.
Only applies to :class:`QSerialInstrument` instances — fake
instruments stay on the main thread. After this call, all
:meth:`device.get` and :meth:`device.set` invocations from the
GUI thread are delivered as queued slot calls and processed
sequentially by the worker thread's event loop, keeping serial
I/O off the main thread entirely.
'''
if not isinstance(self._device, QSerialInstrument):
return
self._thread = QtCore.QThread(self)
self._device.moveToThread(self._thread)
self._thread.start()
def _restoreSettings(self) -> None:
'''Reconcile hardware state with the saved configuration file.
Reads the current hardware state via :attr:`device.settings`
and the saved configuration via :meth:`Configure.read`.
- **No saved file**: writes hardware values to the config file
and returns without changing the hardware.
- **Files match**: no action.
- **Files differ**: shows a :class:`QReconcileDialog`. If the
user chooses "Keep Hardware" (or dismisses the dialog), the
config file is updated to reflect the hardware. If the user
chooses "Use Saved", the saved values are pushed to the device.
The default button in the dialog is controlled by
:attr:`HARDWARE_DOMINANT`.
'''
hw = self._device.settings
saved = self._configure.read(self._device)
if saved is None:
self._configure.save(self._device)
return
diff_keys = [
k for k in hw
if k in saved and values_differ(hw[k], saved[k])
]
if not diff_keys:
return
dialog = QReconcileDialog(
hw, saved, diff_keys,
hardware_dominant=self.HARDWARE_DOMINANT,
parent=self,
)
accepted = dialog.exec()
if not accepted or dialog.keep_hardware:
self._configure.save(self._device)
else:
self._device.settings = saved
[docs]
def closeEvent(self, event) -> None:
'''Stop the worker thread and save settings when the widget is closed.
Stops the device worker thread before saving so that no queued
slot calls arrive after the widget is gone. If the device has a
:meth:`stopPolling` slot, it is called before the thread is
stopped; it only sets a flag, so it is safe from any thread.
Saves using the current widget values (which reflect the last
known device state) rather than querying the device directly,
avoiding any cross-thread read. Only saves if the widget was
previously shown, so that test widgets closed during teardown do
not overwrite saved configuration.
'''
if self._thread is not None:
if hasattr(self._device, 'stopPolling'):
self._device.stopPolling()
self._thread.quit()
self._thread.wait()
if self._restored and self._device is not None:
self._configure.save(self._device, settings=self.settings)
super().closeEvent(event)
[docs]
@classmethod
def example(cls) -> None:
'''Display the widget.
Creates a ``QApplication``, instantiates the widget, shows it,
and runs the event loop.
Intended to be called from ``__main__`` in each widget module:
.. code-block:: python
if __name__ == '__main__':
QMyWidget.example()
'''
import sys
from qtpy.QtWidgets import QApplication
app = QApplication.instance() or QApplication(sys.argv)
widget = cls()
if widget.device is None or not widget.device.isOpen():
fake_cls = cls._fakeCls()
if fake_cls is None:
print(f'{cls.__name__}: instrument not found '
'or not connected.')
return
print(f'{cls.__name__}: instrument not found, '
f'using {fake_cls.__name__}.')
widget = cls(device=fake_cls())
widget.adjustSize()
widget.show()
sys.exit(app.exec())
@classmethod
def _fakeCls(cls) -> type | None:
'''Return the fake instrument class from the sibling ``fake`` module.
Delegates to :func:`~QInstrument.lib.lazy.find_fake_cls`.
'''
return find_fake_cls(cls)
__all__ = ['QInstrumentWidget']