1
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-04-08 09:47:52 +02:00

Compare commits

...

32 Commits

Author SHA1 Message Date
semantic-release
5c11fde0a9 0.72.0
Automatically generated by python-semantic-release
2024-06-24 11:47:52 +00:00
4ca1efeeb8 feat(connector): added threadpool wrapper 2024-06-24 13:41:10 +02:00
aa7ce2ea27 tests(status_box_test): temporary disabled tests for status_box due to high rate of failures 2024-06-24 13:15:59 +02:00
semantic-release
174f0cdcb6 0.71.1
Automatically generated by python-semantic-release
2024-06-23 15:30:25 +00:00
860517a321 fix: don't print exception if the auto-update module cannot be found in plugins 2024-06-23 17:23:39 +02:00
semantic-release
66daae6d9e 0.71.0
Automatically generated by python-semantic-release
2024-06-23 12:33:09 +00:00
83001a0d82 test(scan_control):e2e tests added 2024-06-23 12:53:15 +02:00
1b7921a7f2 doc(scan_control): docs added 2024-06-23 12:12:22 +02:00
8badb6adc1 fix(cleanup): cleanup added to device_input widgets and scan_control 2024-06-23 12:12:20 +02:00
37682e7b8a fix(scan_group_box): added row counter based on widgets 2024-06-23 12:11:30 +02:00
56e74a0e7d test(scan_control): tests added 2024-06-23 12:11:30 +02:00
ec4574ed5c fix(scan_control): added default min limit for args bundle if specified 2024-06-23 12:11:30 +02:00
21d20e0fc7 fix(device_line_edit):SizePolicy fixed for 100 horizontal 2024-06-23 12:11:30 +02:00
7ce3a83c58 fix(scan_control): argbox delete later added to prevent overlapping gui if scan changed 2024-06-23 12:11:30 +02:00
6dff1879c4 fix(scan_control): only scans with defined gui_config are allowed 2024-06-23 12:11:30 +02:00
c09644b29d tests WIP 2024-06-23 12:11:30 +02:00
d8cf44134c feat(scan_group_box): scan box for args and kwargs separated from ScanControlGUI code 2024-06-23 12:11:30 +02:00
ca856384f3 fix(WidgetIO): find handlers within base classes 2024-06-23 12:11:30 +02:00
4e2c9df6a4 refactor(device_line_edit): renamed default_device to default 2024-06-23 12:11:27 +02:00
8b822e0fa8 fix(scan_control): adapted widget to scan BEC gui config 2024-06-23 12:10:33 +02:00
67d398caf7 fix(scan_control): scan_control.py combatible with the newest BEC versions, test disabled 2024-06-23 12:10:33 +02:00
semantic-release
c2c27f8279 0.70.0
Automatically generated by python-semantic-release
2024-06-21 17:22:25 +00:00
50b3422528 fix(bec-desiger+plugins): imports fixed, PYSIDE6 check to not enable run plugins with pyqt6 2024-06-21 18:34:20 +02:00
4639eee0b9 feat(bec-designer): automatic plugin discovery 2024-06-21 18:34:20 +02:00
b4b27aea3d feat(device_line_edit): plugin added to bec-designer 2024-06-21 17:40:53 +02:00
e483b282db feat(device_combobox): plugin added to bec-designer 2024-06-21 17:30:02 +02:00
36391db607 feat: added entry point for bec-designer 2024-06-21 16:46:04 +02:00
5362334ff3 feat(utils/bec-designer): added startup script to launched QtDesigner compatible with conda environments 2024-06-21 16:46:04 +02:00
fdf11d8147 docs: fix typo in link 2024-06-21 14:44:50 +02:00
semantic-release
204f653b72 0.69.0
Automatically generated by python-semantic-release
2024-06-21 12:25:10 +00:00
48ae950d57 feat(widgets): added vscode widget 2024-06-21 13:47:05 +02:00
925c893f3f fix(generate_cli): fixed rpc generate for classes without user access; closes #226 2024-06-21 13:47:05 +02:00
42 changed files with 2031 additions and 808 deletions

View File

@@ -1,5 +1,95 @@
# CHANGELOG
## v0.72.0 (2024-06-24)
### Feature
* feat(connector): added threadpool wrapper ([`4ca1efe`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/4ca1efeeb8955604069f7b98374c7f82e1a8da67))
### Unknown
* tests(status_box_test): temporary disabled tests for status_box due to high rate of failures ([`aa7ce2e`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/aa7ce2ea27bb9564d4f5104bbff30725b8656453))
## v0.71.1 (2024-06-23)
### Fix
* fix: don't print exception if the auto-update module cannot be found in plugins ([`860517a`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/860517a3211075d1f6e2af7fa6a567b9e0cd77f3))
## v0.71.0 (2024-06-23)
### Feature
* feat(scan_group_box): scan box for args and kwargs separated from ScanControlGUI code ([`d8cf441`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/d8cf44134c30063e586771f9068947fef7a306d1))
### Fix
* fix(cleanup): cleanup added to device_input widgets and scan_control ([`8badb6a`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/8badb6adc1d003dbf0b2b1a800c34821f3fc9aa3))
* fix(scan_group_box): added row counter based on widgets ([`37682e7`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/37682e7b8a6ede38308880d285e41a948d6fe831))
* fix(scan_control): added default min limit for args bundle if specified ([`ec4574e`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/ec4574ed5c2c85ea6fbbe2b98f162a8e1220653b))
* fix(scan_control): argbox delete later added to prevent overlapping gui if scan changed ([`7ce3a83`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/7ce3a83c58cb69c2bf7cb7f4eaba7e6a2ca6c546))
* fix(scan_control): only scans with defined gui_config are allowed ([`6dff187`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/6dff1879c4178df0f8ebfd35101acdebb028d572))
* fix(WidgetIO): find handlers within base classes ([`ca85638`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/ca856384f380dabf28d43f1cd48511af784c035b))
* fix(scan_control): adapted widget to scan BEC gui config ([`8b822e0`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/8b822e0fa8e28f080b9a4bf81948a7280a4c07bf))
* fix(scan_control): scan_control.py combatible with the newest BEC versions, test disabled ([`67d398c`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/67d398caf74e08ab25a70cc5d85a5f0c2de8212d))
### Refactor
* refactor(device_line_edit): renamed default_device to default ([`4e2c9df`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/4e2c9df6a4979d935285fd7eba17fd7fd455a35c))
### Test
* test(scan_control): tests added ([`56e74a0`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/56e74a0e7da72d18e89bc30d1896dbf9ef97cd6b))
### Unknown
* test(scan_control):e2e tests added ([`83001a0`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/83001a0d8267e1320549b07032857dcf46ecd293))
* doc(scan_control): docs added ([`1b7921a`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/1b7921a7f2e3bcc846219a2a7aa0de0fd27bb8fe))
* fix(device_line_edit):SizePolicy fixed for 100 horizontal ([`21d20e0`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/21d20e0fc78e9a3853abe802733388cce119ce20))
* tests WIP ([`c09644b`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/c09644b29ddb291c91dc58bcd6ebf02ff45cab36))
## v0.70.0 (2024-06-21)
### Documentation
* docs: fix typo in link ([`fdf11d8`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/fdf11d8147750e379af9b17792761a267b49ae53))
### Feature
* feat(bec-designer): automatic plugin discovery ([`4639eee`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/4639eee0b975ebd7a946e0e290449f5b88c372eb))
* feat(device_line_edit): plugin added to bec-designer ([`b4b27ae`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/b4b27aea3d8c08fa3d5d5514c69dbde32721d1dc))
* feat(device_combobox): plugin added to bec-designer ([`e483b28`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/e483b282db20a81182b87938ea172654092419b5))
* feat: added entry point for bec-designer ([`36391db`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/36391db60735d57b371211791ddf8d3d00cebcf1))
* feat(utils/bec-designer): added startup script to launched QtDesigner compatible with conda environments ([`5362334`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/5362334ff3b07fc83653323a084a4b6946bade96))
### Fix
* fix(bec-desiger+plugins): imports fixed, PYSIDE6 check to not enable run plugins with pyqt6 ([`50b3422`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/50b3422528d46d74317e8c903b6286e868ab7fe0))
## v0.69.0 (2024-06-21)
### Feature
* feat(widgets): added vscode widget ([`48ae950`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/48ae950d57b454307ce409e2511f7b7adf3cfc6b))
### Fix
* fix(generate_cli): fixed rpc generate for classes without user access; closes #226 ([`925c893`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/925c893f3ff4337fc8b4d237c8ffc19a597b0996))
## v0.68.0 (2024-06-21)
### Feature
@@ -61,120 +151,3 @@ in their parent process ([`ce37416`](https://gitlab.psi.ch/bec/bec_widgets/-/com
* fix(pyqt): webengine must be imported before qcoreapplication ([`cbbd23a`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/cbbd23aa33095141e4c265719d176c4aa8c25996))
## v0.65.1 (2024-06-20)
### Fix
* fix: prevent segfault by closing the QCoreApplication, if any ([`fa344a5`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/fa344a5799b07a2d8ace63cc7010b69bc4ed6f1d))
## v0.65.0 (2024-06-20)
### Feature
* feat(device_input): DeviceLineEdit with QCompleter added ([`50e41ff`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/50e41ff26160ec26d77feb6d519e4dad902a9b9b))
* feat(device_combobox): DeviceInputBase and DeviceComboBox added ([`430b282`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/430b282039806e3fbc6cf98e958861a065760620))
### Fix
* fix(device_input_base): bug with setting config and overwriting default device and filter ([`d79f7e9`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/d79f7e9ccde03dc77819ca556c79736d30f7821a))
### Test
* test(device_input): tests added ([`1a0a98a`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/1a0a98a45367db414bed813bbd346b3e1ae8d550))
## v0.64.2 (2024-06-19)
### Fix
* fix(client_utils): added close rpc command to shutdown of gui from bec_ipython_client ([`e5a7d47`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/e5a7d47b21cbf066f740f1d11d7c9ea7c70f3080))
## v0.64.1 (2024-06-19)
### Fix
* fix(widgets): removed widget module import of sub widgets ([`216511b`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/216511b951ff0e15b6d7c70133095f3ac45c23f4))
### Refactor
* refactor(utils): moved get_rpc_widgets to plugin_utils ([`6dabbf8`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/6dabbf874fbbdde89c34a7885bf95aa9c895a28b))
### Test
* test: moved rpc_classes test ([`b3575eb`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/b3575eb06852b456cde915dfda281a3e778e3aeb))
## v0.64.0 (2024-06-19)
### Ci
* ci: add job optional dependency check ([`27426ce`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/27426ce7a52b4cbad7f3bef114d6efe6ad73bd7f))
### Documentation
* docs: fix links in developer section ([`9e16f2f`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/9e16f2faf9c59a5d36ae878512c5a910cca31e69))
* docs: refactor developer section, add widget tutorial ([`2a36d93`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/2a36d9364f242bf42e4cda4b50e6f46aa3833bbd))
### Feature
* feat: add option to change size of the fonts ([`ea805d1`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/ea805d1362fc084d3b703b6f81b0180072f0825d))
### Fix
* fix(plot_base): font size is set with setScale which is scaling the whole legend window ([`5d66720`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/5d6672069ea1cbceb62104f66c127e4e3c23e4a4))
### Test
* test: add tests ([`140ad83`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/140ad83380808928edf7953e23c762ab72a0a1e9))
## v0.63.2 (2024-06-14)
### Fix
* fix: do not import "server" in client, prevents from having trouble with QApplication creation order
Like with QtWebEngine ([`6f96498`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/6f96498de66358b89f3a2035627eed2e02dde5a1))
### Unknown
* Reapply "feat: implement non-polling, interruptible waiting of gui instruction response with timeout"
This reverts commit fe04dd80e59a0e74f7fdea603e0642707ecc7c2a. ([`836b6e6`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/836b6e64f694916d6b6f909dedf11a4a6d2c86a4))
## v0.63.1 (2024-06-13)
### Fix
* fix: just terminate the remote process in close() instead of communicating
The proper finalization sequence will be executed by the remote process
on SIGTERM ([`9263f8e`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/9263f8ef5c17ae7a007a1a564baf787b39061756))
## v0.63.0 (2024-06-13)
### Documentation
* docs: add documentation ([`bc709c4`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/bc709c4184c985d4e721f9ea7d1b3dad5e9153a7))
### Feature
* feat: add textbox widget ([`d9d4e3c`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/d9d4e3c9bf73ab2a5629c2867b50fc91e69489ec))
### Refactor
* refactor: add pydantic config, add change_theme ([`6b8432f`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/6b8432f5b20a71175a3537b5f6832b76e3b67d73))
### Test
* test: add test for text box ([`b49462a`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/b49462abeb186e56bac79d2ef0b0add1ef28a1a5))
### Unknown
* Revert "feat: implement non-polling, interruptible waiting of gui instruction response with timeout"
This reverts commit abc6caa2d0b6141dfbe1f3d025f78ae14deddcb3 ([`fe04dd8`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/fe04dd80e59a0e74f7fdea603e0642707ecc7c2a))
## v0.62.0 (2024-06-12)
### Unknown
* doc: add documentation about creating custom GUI applications embedding BEC Widgets ([`17a0068`](https://gitlab.psi.ch/bec/bec_widgets/-/commit/17a00687579f5efab1990cd83862ec0e78198633))

View File

@@ -17,8 +17,10 @@ class Widgets(str, enum.Enum):
BECDock = "BECDock"
BECDockArea = "BECDockArea"
BECFigure = "BECFigure"
ScanControl = "ScanControl"
SpiralProgressBar = "SpiralProgressBar"
TextBox = "TextBox"
VSCodeEditor = "VSCodeEditor"
WebsiteWidget = "WebsiteWidget"
@@ -1822,6 +1824,24 @@ class Ring(RPCBase):
"""
class ScanControl(RPCBase):
@property
@rpc_call
def config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call
def get_all_rpc(self) -> "dict":
"""
Get all registered RPC objects.
"""
class SpiralProgressBar(RPCBase):
@rpc_call
def get_all_rpc(self) -> "dict":
@@ -2049,6 +2069,9 @@ class TextBox(RPCBase):
"""
class VSCodeEditor(RPCBase): ...
class WebsiteWidget(RPCBase):
@rpc_call
def set_url(self, url: str) -> None:

View File

@@ -141,6 +141,10 @@ class BECGuiClientMixin:
for ep in eps:
if ep.name == "plugin_widgets_update":
try:
spec = importlib.util.find_spec(ep.module)
# if the module is not found, we skip it
if spec is None:
continue
return ep.load()(gui=self)
except Exception as e:
print(f"Error loading auto update script from plugin: {str(e)}")

View File

@@ -83,6 +83,9 @@ class {class_name}(RPCBase, BECGuiClientMixin):"""
else:
self.content += f"""
class {class_name}(RPCBase):"""
if not cls.USER_ACCESS:
self.content += """...
"""
for method in cls.USER_ACCESS:
obj = getattr(cls, method)
if isinstance(obj, property):

View File

@@ -0,0 +1,17 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
"""PySide6 port of the Qt Designer taskmenuextension example from Qt v6.x"""
import sys
from bec_ipython_client.main import BECIPythonClient
from qtpy.QtWidgets import QApplication
from tictactoe import TicTacToe
if __name__ == "__main__": # pragma: no cover
app = QApplication(sys.argv)
window = TicTacToe()
window.state = "-X-XO----"
window.show()
sys.exit(app.exec())

View File

@@ -0,0 +1,12 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from qtpy.QtDesigner import QPyDesignerCustomWidgetCollection
from tictactoe import TicTacToe
from tictactoeplugin import TicTacToePlugin
# Set PYSIDE_DESIGNER_PLUGINS to point to this directory and load the plugin
if __name__ == "__main__": # pragma: no cover
QPyDesignerCustomWidgetCollection.addCustomWidget(TicTacToePlugin())

View File

@@ -0,0 +1,4 @@
{
"files": ["tictactoe.py", "main.py", "registertictactoe.py", "tictactoeplugin.py",
"tictactoetaskmenu.py"]
}

View File

@@ -0,0 +1,135 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from qtpy.QtCore import Property, QPoint, QRect, QSize, Qt, Slot
from qtpy.QtGui import QPainter, QPen
from qtpy.QtWidgets import QWidget
EMPTY = "-"
CROSS = "X"
NOUGHT = "O"
DEFAULT_STATE = "---------"
class TicTacToe(QWidget): # pragma: no cover
def __init__(self, parent=None):
super().__init__(parent)
self._state = DEFAULT_STATE
self._turn_number = 0
def minimumSizeHint(self):
return QSize(200, 200)
def sizeHint(self):
return QSize(200, 200)
def setState(self, new_state):
self._turn_number = 0
self._state = DEFAULT_STATE
for position in range(min(9, len(new_state))):
mark = new_state[position]
if mark == CROSS or mark == NOUGHT:
self._turn_number += 1
self._change_state_at(position, mark)
position += 1
self.update()
def state(self):
return self._state
@Slot()
def clear_board(self):
self._state = DEFAULT_STATE
self._turn_number = 0
self.update()
def _change_state_at(self, pos, new_state):
self._state = self._state[:pos] + new_state + self._state[pos + 1 :]
def mousePressEvent(self, event):
if self._turn_number == 9:
self.clear_board()
return
for position in range(9):
cell = self._cell_rect(position)
if cell.contains(event.position().toPoint()):
if self._state[position] == EMPTY:
new_state = CROSS if self._turn_number % 2 == 0 else NOUGHT
self._change_state_at(position, new_state)
self._turn_number += 1
self.update()
def paintEvent(self, event):
with QPainter(self) as painter:
painter.setRenderHint(QPainter.Antialiasing)
painter.setPen(QPen(Qt.darkGreen, 1))
painter.drawLine(self._cell_width(), 0, self._cell_width(), self.height())
painter.drawLine(2 * self._cell_width(), 0, 2 * self._cell_width(), self.height())
painter.drawLine(0, self._cell_height(), self.width(), self._cell_height())
painter.drawLine(0, 2 * self._cell_height(), self.width(), 2 * self._cell_height())
painter.setPen(QPen(Qt.darkBlue, 2))
for position in range(9):
cell = self._cell_rect(position)
if self._state[position] == CROSS:
painter.drawLine(cell.topLeft(), cell.bottomRight())
painter.drawLine(cell.topRight(), cell.bottomLeft())
elif self._state[position] == NOUGHT:
painter.drawEllipse(cell)
painter.setPen(QPen(Qt.yellow, 3))
for position in range(0, 8, 3):
if (
self._state[position] != EMPTY
and self._state[position + 1] == self._state[position]
and self._state[position + 2] == self._state[position]
):
y = self._cell_rect(position).center().y()
painter.drawLine(0, y, self.width(), y)
self._turn_number = 9
for position in range(3):
if (
self._state[position] != EMPTY
and self._state[position + 3] == self._state[position]
and self._state[position + 6] == self._state[position]
):
x = self._cell_rect(position).center().x()
painter.drawLine(x, 0, x, self.height())
self._turn_number = 9
if (
self._state[0] != EMPTY
and self._state[4] == self._state[0]
and self._state[8] == self._state[0]
):
painter.drawLine(0, 0, self.width(), self.height())
self._turn_number = 9
if (
self._state[2] != EMPTY
and self._state[4] == self._state[2]
and self._state[6] == self._state[2]
):
painter.drawLine(0, self.height(), self.width(), 0)
self._turn_number = 9
def _cell_rect(self, position):
h_margin = self.width() / 30
v_margin = self.height() / 30
row = int(position / 3)
column = position - 3 * row
pos = QPoint(column * self._cell_width() + h_margin, row * self._cell_height() + v_margin)
size = QSize(self._cell_width() - 2 * h_margin, self._cell_height() - 2 * v_margin)
return QRect(pos, size)
def _cell_width(self):
return self.width() / 3
def _cell_height(self):
return self.height() / 3
state = Property(str, state, setState)

View File

@@ -0,0 +1,68 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtGui import QIcon
from tictactoe import TicTacToe
from tictactoetaskmenu import TicTacToeTaskMenuFactory
DOM_XML = """
<ui language='c++'>
<widget class='TicTacToe' name='ticTacToe'>
<property name='geometry'>
<rect>
<x>0</x>
<y>0</y>
<width>200</width>
<height>200</height>
</rect>
</property>
<property name='state'>
<string>-X-XO----</string>
</property>
</widget>
</ui>
"""
class TicTacToePlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
t = TicTacToe(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return ""
def icon(self):
return QIcon()
def includeFile(self):
return "tictactoe"
def initialize(self, form_editor):
self._form_editor = form_editor
manager = form_editor.extensionManager()
iid = TicTacToeTaskMenuFactory.task_menu_iid()
manager.registerExtensions(TicTacToeTaskMenuFactory(manager), iid)
def isContainer(self):
return False
def isInitialized(self):
return self._form_editor is not None
def name(self):
return "TicTacToe"
def toolTip(self):
return "Tic Tac Toe Example, demonstrating class QDesignerTaskMenuExtension (Python)"
def whatsThis(self):
return self.toolTip()

View File

@@ -0,0 +1,67 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from qtpy.QtCore import Slot
from qtpy.QtDesigner import QExtensionFactory, QPyDesignerTaskMenuExtension
from qtpy.QtGui import QAction
from qtpy.QtWidgets import QDialog, QDialogButtonBox, QVBoxLayout
from tictactoe import TicTacToe
class TicTacToeDialog(QDialog): # pragma: no cover
def __init__(self, parent):
super().__init__(parent)
layout = QVBoxLayout(self)
self._ticTacToe = TicTacToe(self)
layout.addWidget(self._ticTacToe)
button_box = QDialogButtonBox(
QDialogButtonBox.Ok | QDialogButtonBox.Cancel | QDialogButtonBox.Reset
)
button_box.accepted.connect(self.accept)
button_box.rejected.connect(self.reject)
reset_button = button_box.button(QDialogButtonBox.Reset)
reset_button.clicked.connect(self._ticTacToe.clear_board)
layout.addWidget(button_box)
def set_state(self, new_state):
self._ticTacToe.setState(new_state)
def state(self):
return self._ticTacToe.state
class TicTacToeTaskMenu(QPyDesignerTaskMenuExtension):
def __init__(self, ticTacToe, parent):
super().__init__(parent)
self._ticTacToe = ticTacToe
self._edit_state_action = QAction("Edit State...", None)
self._edit_state_action.triggered.connect(self._edit_state)
def taskActions(self):
return [self._edit_state_action]
def preferredEditAction(self):
return self._edit_state_action
@Slot()
def _edit_state(self):
dialog = TicTacToeDialog(self._ticTacToe)
dialog.set_state(self._ticTacToe.state)
if dialog.exec() == QDialog.Accepted:
self._ticTacToe.state = dialog.state()
class TicTacToeTaskMenuFactory(QExtensionFactory):
def __init__(self, extension_manager):
super().__init__(extension_manager)
@staticmethod
def task_menu_iid():
return "org.qt-project.Qt.Designer.TaskMenu"
def createExtension(self, object, iid, parent):
if iid != TicTacToeTaskMenuFactory.task_menu_iid():
return None
if object.__class__.__name__ != "TicTacToe":
return None
return TicTacToeTaskMenu(object, parent)

View File

@@ -2,10 +2,11 @@
from __future__ import annotations
import time
from typing import Optional, Type
from typing import Optional
from bec_lib.utils.import_utils import lazy_import, lazy_import_from
from bec_lib.utils.import_utils import lazy_import_from
from pydantic import BaseModel, Field, field_validator
from qtpy.QtCore import QObject, QRunnable, QThreadPool, Signal
from qtpy.QtCore import Slot as pyqtSlot
from bec_widgets.cli.rpc_register import RPCRegister
@@ -33,6 +34,31 @@ class ConnectionConfig(BaseModel):
return v
class WorkerSignals(QObject):
progress = Signal(dict)
completed = Signal()
class Worker(QRunnable):
"""
Worker class to run a function in a separate thread.
"""
def __init__(self, func, *args, **kwargs):
super().__init__()
self.signals = WorkerSignals()
self.func = func
self.args = args
self.kwargs = kwargs
def run(self):
"""
Run the specified function in the thread.
"""
self.func(*self.args, **self.kwargs)
self.signals.completed.emit()
class BECConnector:
"""Connection mixin class for all BEC widgets, to handle BEC client and device manager"""
@@ -63,6 +89,43 @@ class BECConnector:
self.rpc_register = RPCRegister()
self.rpc_register.add_rpc(self)
self._thread_pool = QThreadPool.globalInstance()
def submit_task(self, fn, *args, on_complete: pyqtSlot = None, **kwargs) -> Worker:
"""
Submit a task to run in a separate thread. The task will run the specified
function with the provided arguments and emit the completed signal when done.
Use this method if you want to wait for a task to complete without blocking the
main thread.
Args:
fn: Function to run in a separate thread.
*args: Arguments for the function.
on_complete: Slot to run when the task is complete.
**kwargs: Keyword arguments for the function.
Returns:
worker: The worker object that will run the task.
Examples:
>>> def my_function(a, b):
>>> print(a + b)
>>> self.submit_task(my_function, 1, 2)
>>> def my_function(a, b):
>>> print(a + b)
>>> def on_complete():
>>> print("Task complete")
>>> self.submit_task(my_function, 1, 2, on_complete=on_complete)
"""
worker = Worker(fn, *args, **kwargs)
if on_complete:
worker.signals.completed.connect(on_complete)
self._thread_pool.start(worker)
return worker
def get_all_rpc(self) -> dict:
"""Get all registered RPC objects."""
all_connections = self.rpc_register.list_all_connections()

View File

@@ -0,0 +1,87 @@
import os
import sys
import sysconfig
from pathlib import Path
from qtpy import PYSIDE6
if PYSIDE6:
from PySide6.scripts.pyside_tool import (
_extend_path_var,
init_virtual_env,
is_pyenv_python,
is_virtual_env,
qt_tool_wrapper,
ui_tool_binary,
)
import bec_widgets
def patch_designer(): # pragma: no cover
if not PYSIDE6:
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
return
init_virtual_env()
major_version = sys.version_info[0]
minor_version = sys.version_info[1]
os.environ["PY_MAJOR_VERSION"] = str(major_version)
os.environ["PY_MINOR_VERSION"] = str(minor_version)
if sys.platform == "linux":
version = f"{major_version}.{minor_version}"
library_name = f"libpython{version}{sys.abiflags}.so"
if is_pyenv_python():
library_name = str(Path(sysconfig.get_config_var("LIBDIR")) / library_name)
os.environ["LD_PRELOAD"] = library_name
elif sys.platform == "darwin":
library_name = f"libpython{major_version}.{minor_version}.dylib"
lib_path = str(Path(sysconfig.get_config_var("LIBDIR")) / library_name)
os.environ["DYLD_INSERT_LIBRARIES"] = lib_path
elif sys.platform == "win32":
if is_virtual_env():
_extend_path_var("PATH", os.fspath(Path(sys._base_executable).parent), True)
qt_tool_wrapper(ui_tool_binary("designer"), sys.argv[1:])
def find_plugin_paths(base_path: Path):
"""
Recursively find all directories containing a .pyproject file.
"""
plugin_paths = []
for path in base_path.rglob("*.pyproject"):
plugin_paths.append(str(path.parent))
return plugin_paths
def set_plugin_environment_variable(plugin_paths):
"""
Set the PYSIDE_DESIGNER_PLUGINS environment variable with the given plugin paths.
"""
current_paths = os.environ.get("PYSIDE_DESIGNER_PLUGINS", "")
if current_paths:
current_paths = current_paths.split(os.pathsep)
else:
current_paths = []
current_paths.extend(plugin_paths)
os.environ["PYSIDE_DESIGNER_PLUGINS"] = os.pathsep.join(current_paths)
# Patch the designer function
def main(): # pragma: no cover
if not PYSIDE6:
print("PYSIDE6 is not available in the environment. Exiting...")
return
base_dir = Path(os.path.dirname(bec_widgets.__file__)).resolve()
plugin_paths = find_plugin_paths(base_dir)
set_plugin_environment_variable(plugin_paths)
patch_designer()
if __name__ == "__main__": # pragma: no cover
main()

View File

@@ -119,7 +119,7 @@ class WidgetIO:
widget: Widget instance.
ignore_errors(bool, optional): Whether to ignore if no handler is found.
"""
handler_class = WidgetIO._handlers.get(type(widget))
handler_class = WidgetIO._find_handler(widget)
if handler_class:
return handler_class().get_value(widget) # Instantiate the handler
if not ignore_errors:
@@ -136,12 +136,28 @@ class WidgetIO:
value: Value to set.
ignore_errors(bool, optional): Whether to ignore if no handler is found.
"""
handler_class = WidgetIO._handlers.get(type(widget))
handler_class = WidgetIO._find_handler(widget)
if handler_class:
handler_class().set_value(widget, value) # Instantiate the handler
elif not ignore_errors:
raise ValueError(f"No handler for widget type: {type(widget)}")
@staticmethod
def _find_handler(widget):
"""
Find the appropriate handler for the widget by checking its base classes.
Args:
widget: Widget instance.
Returns:
handler_class: The handler class if found, otherwise None.
"""
for base in type(widget).__mro__:
if base in WidgetIO._handlers:
return WidgetIO._handlers[base]
return None
################## for exporting and importing widget hierarchies ##################

View File

@@ -18,7 +18,7 @@ class DeviceComboBox(DeviceInputBase, QComboBox):
config: Device input configuration.
gui_id: GUI ID.
device_filter: Device filter, name of the device class.
default_device: Default device name.
default: Default device name.
arg_name: Argument name, can be used for the other widgets which has to call some other function in bec using correct argument names.
"""
@@ -29,7 +29,7 @@ class DeviceComboBox(DeviceInputBase, QComboBox):
config: DeviceInputConfig = None,
gui_id: str | None = None,
device_filter: str | None = None,
default_device: str | None = None,
default: str | None = None,
arg_name: str | None = None,
):
super().__init__(client=client, config=config, gui_id=gui_id)
@@ -41,8 +41,8 @@ class DeviceComboBox(DeviceInputBase, QComboBox):
self.config.arg_name = arg_name
if device_filter is not None:
self.set_device_filter(device_filter)
if default_device is not None:
self.set_default_device(default_device)
if default is not None:
self.set_default_device(default)
def set_device_filter(self, device_filter: str):
"""
@@ -83,13 +83,10 @@ class DeviceComboBox(DeviceInputBase, QComboBox):
raise ValueError(f"Device {device_name} is not found.")
return device_obj
def cleanup(self):
"""Cleanup the widget."""
super().cleanup()
if __name__ == "__main__": # pragma: no cover
import sys
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
w = DeviceComboBox(default_device="samx")
w.show()
sys.exit(app.exec_())
def closeEvent(self, event):
super().cleanup()
QComboBox().closeEvent(event)

View File

@@ -0,0 +1,4 @@
{
"files": ["device_combobox.py", "launch_device_combobox.py",
]
}

View File

@@ -0,0 +1,54 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtGui import QIcon
from bec_widgets.widgets.device_inputs import DeviceComboBox
DOM_XML = """
<ui language='c++'>
<widget class='DeviceComboBox' name='device_combobox'>
</widget>
</ui>
"""
class DeviceComboBoxPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
t = DeviceComboBox(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return ""
def icon(self):
return QIcon()
def includeFile(self):
return "device_combobox"
def initialize(self, form_editor):
self._form_editor = form_editor
def isContainer(self):
return False
def isInitialized(self):
return self._form_editor is not None
def name(self):
return "DeviceComboBox"
def toolTip(self):
return "Device ComboBox Example for BEC Widgets"
def whatsThis(self):
return self.toolTip()

View File

@@ -0,0 +1,11 @@
from bec_widgets.widgets.device_inputs import DeviceComboBox
if __name__ == "__main__": # pragma: no cover
import sys
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
w = DeviceComboBox()
w.show()
sys.exit(app.exec_())

View File

@@ -0,0 +1,17 @@
def main(): # pragma: no cover
from qtpy import PYSIDE6
if not PYSIDE6:
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from bec_widgets.widgets.device_inputs.device_combobox.device_combobox_plugin import (
DeviceComboBoxPlugin,
)
QPyDesignerCustomWidgetCollection.addCustomWidget(DeviceComboBoxPlugin())
if __name__ == "__main__": # pragma: no cover
main()

View File

@@ -5,7 +5,7 @@ from bec_widgets.utils import BECConnector, ConnectionConfig
class DeviceInputConfig(ConnectionConfig):
device_filter: str | list[str] | None = None
default_device: str | None = None
default: str | None = None
arg_name: str | None = None
@@ -65,7 +65,7 @@ class DeviceInputBase(BECConnector):
default_device(str): Default device name.
"""
self.validate_device(default_device)
self.config.default_device = default_device
self.config.default = default_device
def get_device_list(self, filter: str | list[str] | None = None) -> list[str]:
"""
@@ -118,3 +118,6 @@ class DeviceInputBase(BECConnector):
"""
if device not in self.get_device_list(self.config.device_filter):
raise ValueError(f"Device {device} is not valid.")
def cleanup(self):
super().cleanup()

View File

@@ -1,6 +1,7 @@
from typing import TYPE_CHECKING
from qtpy.QtWidgets import QCompleter, QLineEdit
from qtpy.QtCore import QSize
from qtpy.QtWidgets import QCompleter, QLineEdit, QSizePolicy
from bec_widgets.widgets.device_inputs.device_input_base import DeviceInputBase, DeviceInputConfig
@@ -18,7 +19,7 @@ class DeviceLineEdit(DeviceInputBase, QLineEdit):
config: Device input configuration.
gui_id: GUI ID.
device_filter: Device filter, name of the device class.
default_device: Default device name.
default: Default device name.
arg_name: Argument name, can be used for the other widgets which has to call some other function in bec using correct argument names.
"""
@@ -29,7 +30,7 @@ class DeviceLineEdit(DeviceInputBase, QLineEdit):
config: DeviceInputConfig = None,
gui_id: str | None = None,
device_filter: str | list[str] | None = None,
default_device: str | None = None,
default: str | None = None,
arg_name: str | None = None,
):
QLineEdit.__init__(self, parent=parent)
@@ -41,10 +42,14 @@ class DeviceLineEdit(DeviceInputBase, QLineEdit):
if arg_name is not None:
self.config.arg_name = arg_name
self.arg_name = arg_name
if device_filter is not None:
self.set_device_filter(device_filter)
if default_device is not None:
self.set_default_device(default_device)
if default is not None:
self.set_default_device(default)
self.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed)
self.setMinimumSize(QSize(100, 0))
def set_device_filter(self, device_filter: str | list[str]):
"""
@@ -90,13 +95,10 @@ class DeviceLineEdit(DeviceInputBase, QLineEdit):
raise ValueError(f"Device {device_name} is not found.")
return device_obj
def cleanup(self):
"""Cleanup the widget."""
super().cleanup()
if __name__ == "__main__": # pragma: no cover
import sys
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
w = DeviceLineEdit(default_device="samx")
w.show()
sys.exit(app.exec_())
def closeEvent(self, event):
super().cleanup()
QLineEdit().closeEvent(event)

View File

@@ -0,0 +1,4 @@
{
"files": ["device_line_edit.py", "launch_device_line_edit.py",
]
}

View File

@@ -0,0 +1,54 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtGui import QIcon
from bec_widgets.widgets.device_inputs import DeviceLineEdit
DOM_XML = """
<ui language='c++'>
<widget class='DeviceLineEdit' name='device_line_edit'>
</widget>
</ui>
"""
class DeviceLineEditPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
t = DeviceLineEdit(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return ""
def icon(self):
return QIcon()
def includeFile(self):
return "device_line_edit"
def initialize(self, form_editor):
self._form_editor = form_editor
def isContainer(self):
return False
def isInitialized(self):
return self._form_editor is not None
def name(self):
return "DeviceLineEdit"
def toolTip(self):
return "Device LineEdit Example for BEC Widgets with autocomplete."
def whatsThis(self):
return self.toolTip()

View File

@@ -0,0 +1,11 @@
from bec_widgets.widgets.device_inputs import DeviceLineEdit
if __name__ == "__main__": # pragma: no cover
import sys
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
w = DeviceLineEdit()
w.show()
sys.exit(app.exec_())

View File

@@ -0,0 +1,17 @@
def main(): # pragma: no cover
from qtpy import PYSIDE6
if not PYSIDE6:
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from bec_widgets.widgets.device_inputs.device_line_edit.device_line_edit_plugin import (
DeviceLineEditPlugin,
)
QPyDesignerCustomWidgetCollection.addCustomWidget(DeviceLineEditPlugin())
if __name__ == "__main__": # pragma: no cover
main()

View File

@@ -1,53 +1,37 @@
import qdarktheme
from bec_lib.endpoints import MessageEndpoints
from qtpy.QtWidgets import (
QApplication,
QCheckBox,
QComboBox,
QDoubleSpinBox,
QFrame,
QGridLayout,
QGroupBox,
QHBoxLayout,
QHeaderView,
QLabel,
QLayout,
QLineEdit,
QPushButton,
QSpinBox,
QTableWidget,
QTableWidgetItem,
QSizePolicy,
QVBoxLayout,
QWidget,
)
from bec_widgets.utils.bec_dispatcher import BECDispatcher
from bec_widgets.utils.widget_io import WidgetIO
from bec_widgets.utils import BECConnector
from bec_widgets.widgets.buttons.stop_button.stop_button import StopButton
from bec_widgets.widgets.scan_control.scan_group_box import ScanGroupBox
class ScanArgType:
DEVICE = "device"
FLOAT = "float"
INT = "int"
BOOL = "bool"
STR = "str"
class ScanControl(BECConnector, QWidget):
class ScanControl(QWidget):
WIDGET_HANDLER = {
ScanArgType.DEVICE: QLineEdit,
ScanArgType.FLOAT: QDoubleSpinBox,
ScanArgType.INT: QSpinBox,
ScanArgType.BOOL: QCheckBox,
ScanArgType.STR: QLineEdit,
}
def __init__(self, parent=None, client=None, allowed_scans=None):
super().__init__(parent)
def __init__(
self, parent=None, client=None, gui_id: str | None = None, allowed_scans: list | None = None
):
super().__init__(client=client, gui_id=gui_id)
QWidget.__init__(self, parent=parent)
# Client from BEC + shortcuts to device manager and scans
self.client = BECDispatcher().client if client is None else client
self.dev = self.client.device_manager.devices
self.scans = self.client.scans
self.get_bec_shortcuts()
# Main layout
self.layout = QVBoxLayout(self)
self.arg_box = None
self.kwarg_boxes = []
self.expert_mode = False # TODO implement in the future versions
# Scan list - allowed scans for the GUI
self.allowed_scans = allowed_scans
@@ -56,389 +40,173 @@ class ScanControl(QWidget):
self._init_UI()
def _init_UI(self):
self.verticalLayout = QVBoxLayout(self)
"""
Initializes the UI of the scan control widget. Create the top box for scan selection and populate scans to main combobox.
"""
# Scan selection group box
self.scan_selection_group = QGroupBox("Scan Selection", self)
self.scan_selection_layout = QVBoxLayout(self.scan_selection_group)
self.comboBox_scan_selection = QComboBox(self.scan_selection_group)
self.button_run_scan = QPushButton("Run Scan", self.scan_selection_group)
self.scan_selection_layout.addWidget(self.comboBox_scan_selection)
self.scan_selection_layout.addWidget(self.button_run_scan)
self.verticalLayout.addWidget(self.scan_selection_group)
# Scan control group box
self.scan_control_group = QGroupBox("Scan Control", self)
self.scan_control_layout = QVBoxLayout(self.scan_control_group)
self.verticalLayout.addWidget(self.scan_control_group)
# Kwargs layout - just placeholder
self.kwargs_layout = QGridLayout()
self.scan_control_layout.addLayout(self.kwargs_layout)
# 1st Separator
self.add_horizontal_separator(self.scan_control_layout)
# Buttons
self.button_layout = QHBoxLayout()
self.pushButton_add_bundle = QPushButton("Add Bundle", self.scan_control_group)
self.pushButton_add_bundle.clicked.connect(self.add_bundle)
self.pushButton_remove_bundle = QPushButton("Remove Bundle", self.scan_control_group)
self.pushButton_remove_bundle.clicked.connect(self.remove_bundle)
self.button_layout.addWidget(self.pushButton_add_bundle)
self.button_layout.addWidget(self.pushButton_remove_bundle)
self.scan_control_layout.addLayout(self.button_layout)
# 2nd Separator
self.add_horizontal_separator(self.scan_control_layout)
# Initialize the QTableWidget for args
self.args_table = QTableWidget()
self.args_table.verticalHeader().setSectionResizeMode(QHeaderView.Fixed)
self.scan_control_layout.addWidget(self.args_table)
self.scan_selection_group = self.create_scan_selection_group()
self.scan_selection_group.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed)
self.layout.addWidget(self.scan_selection_group)
# Connect signals
self.comboBox_scan_selection.currentIndexChanged.connect(self.on_scan_selected)
self.button_run_scan.clicked.connect(self.run_scan)
self.button_add_bundle.clicked.connect(self.add_arg_bundle)
self.button_remove_bundle.clicked.connect(self.remove_arg_bundle)
# Initialize scan selection
self.populate_scans()
def add_horizontal_separator(self, layout) -> None:
def create_scan_selection_group(self) -> QGroupBox:
"""
Adds a horizontal separator to the given layout
Creates the scan selection group box with combobox to select the scan and start/stop button.
Args:
layout: Layout to add the separator to
Returns:
QGroupBox: Group box containing the scan selection widgets.
"""
separator = QFrame(self.scan_control_group)
separator.setFrameShape(QFrame.HLine)
separator.setFrameShadow(QFrame.Sunken)
layout.addWidget(separator)
scan_selection_group = QGroupBox("Scan Selection", self)
self.scan_selection_layout = QGridLayout(scan_selection_group)
self.comboBox_scan_selection = QComboBox(scan_selection_group)
# Run button
self.button_run_scan = QPushButton("Start", scan_selection_group)
self.button_run_scan.setStyleSheet("background-color: #559900; color: white")
# Stop button
self.button_stop_scan = StopButton(parent=scan_selection_group)
# Add bundle button
self.button_add_bundle = QPushButton("Add Bundle", scan_selection_group)
# Remove bundle button
self.button_remove_bundle = QPushButton("Remove Bundle", scan_selection_group)
self.scan_selection_layout.addWidget(self.comboBox_scan_selection, 0, 0, 1, 2)
self.scan_selection_layout.addWidget(self.button_run_scan, 1, 0)
self.scan_selection_layout.addWidget(self.button_stop_scan, 1, 1)
self.scan_selection_layout.addWidget(self.button_add_bundle, 2, 0)
self.scan_selection_layout.addWidget(self.button_remove_bundle, 2, 1)
return scan_selection_group
def populate_scans(self):
"""Populates the scan selection combo box with available scans"""
self.available_scans = self.client.producer.get(MessageEndpoints.available_scans()).resource
"""Populates the scan selection combo box with available scans from BEC session."""
self.available_scans = self.client.connector.get(
MessageEndpoints.available_scans()
).resource
if self.allowed_scans is None:
allowed_scans = self.available_scans.keys()
supported_scans = ["ScanBase", "SyncFlyScanBase", "AsyncFlyScanBase"]
allowed_scans = [
scan_name
for scan_name, scan_info in self.available_scans.items()
if scan_info["base_class"] in supported_scans and len(scan_info["gui_config"]) > 0
]
else:
allowed_scans = self.allowed_scans
# TODO check parent class is ScanBase -> filter out the scans not relevant for GUI
self.comboBox_scan_selection.addItems(allowed_scans)
def on_scan_selected(self):
"""Callback for scan selection combo box"""
self.reset_layout()
selected_scan_name = self.comboBox_scan_selection.currentText()
selected_scan_info = self.available_scans.get(selected_scan_name, {})
print(selected_scan_info) # TODO remove when widget will be more mature
# Generate kwargs input
self.generate_kwargs_input_fields(selected_scan_info)
gui_config = selected_scan_info.get("gui_config", {})
self.arg_group = gui_config.get("arg_group", None)
self.kwarg_groups = gui_config.get("kwarg_groups", None)
# Args section
self.generate_args_input_fields(selected_scan_info)
if self.arg_box is None:
self.button_add_bundle.setEnabled(False)
self.button_remove_bundle.setEnabled(False)
def add_labels_to_layout(self, labels: list, grid_layout: QGridLayout) -> None:
if len(self.arg_group["arg_inputs"]) > 0:
self.button_add_bundle.setEnabled(True)
self.button_remove_bundle.setEnabled(True)
self.add_arg_group(self.arg_group)
if len(self.kwarg_groups) > 0:
self.add_kwargs_boxes(self.kwarg_groups)
self.update()
self.adjustSize()
def add_kwargs_boxes(self, groups: list):
"""
Adds labels to the given grid layout as a separate row.
Adds the given gui_groups to the scan control layout.
Args:
labels (list): List of label names to add.
grid_layout (QGridLayout): The grid layout to which labels will be added.
groups(list): List of dictionaries containing the gui_group information.
"""
row_index = grid_layout.rowCount() # Get the next available row
for column_index, label_name in enumerate(labels):
label = QLabel(label_name.capitalize(), self.scan_control_group)
# Add the label to the grid layout at the calculated row and current column
grid_layout.addWidget(label, row_index, column_index)
for group in groups:
box = ScanGroupBox(box_type="kwargs", config=group)
box.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed)
self.layout.addWidget(box)
self.kwarg_boxes.append(box)
def add_labels_to_table(
self, labels: list, table: QTableWidget
) -> None: # TODO could be moved to BECTable
def add_arg_group(self, group: dict):
"""
Adds labels to the given table widget as a header row.
Adds the given gui_groups to the scan control layout.
Args:
labels(list): List of label names to add.
table(QTableWidget): The table widget to which labels will be added.
"""
table.setColumnCount(len(labels))
table.setHorizontalHeaderLabels(labels)
self.arg_box = ScanGroupBox(box_type="args", config=group)
self.arg_box.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed)
self.layout.addWidget(self.arg_box)
def generate_args_input_fields(self, scan_info: dict) -> None:
"""
Generates input fields for args.
def add_arg_bundle(self):
self.arg_box.add_widget_bundle()
Args:
scan_info(dict): Scan signature dictionary from BEC.
"""
def remove_arg_bundle(self):
self.arg_box.remove_widget_bundle()
# Setup args table limits
self.set_args_table_limits(self.args_table, scan_info)
def reset_layout(self):
"""Clears the scan control layout from GuiGroups and ArgGroups boxes."""
if self.arg_box is not None:
self.layout.removeWidget(self.arg_box)
self.arg_box.deleteLater()
self.arg_box = None
if self.kwarg_boxes != []:
self.remove_kwarg_boxes()
# Get arg_input from selected scan
self.arg_input = scan_info.get("arg_input", {})
# Generate labels for table
self.add_labels_to_table(list(self.arg_input.keys()), self.args_table)
# add minimum number of args rows
if self.arg_size_min is not None:
for i in range(self.arg_size_min):
self.add_bundle()
def generate_kwargs_input_fields(self, scan_info: dict) -> None:
"""
Generates input fields for kwargs
Args:
scan_info(dict): Scan signature dictionary from BEC.
"""
# Create a new kwarg layout to replace the old one - this is necessary because otherwise row count is not reseted
self.clear_and_delete_layout(self.kwargs_layout)
self.kwargs_layout = self.create_new_grid_layout() # Create new grid layout
self.scan_control_layout.insertLayout(0, self.kwargs_layout)
# Get signature
signature = scan_info.get("signature", [])
# Extract kwargs from the converted signature
kwargs = [param["name"] for param in signature if param["kind"] == "KEYWORD_ONLY"]
# Add labels
self.add_labels_to_layout(kwargs, self.kwargs_layout)
# Add widgets
widgets = self.generate_widgets_from_signature(kwargs, signature)
self.add_widgets_row_to_layout(self.kwargs_layout, widgets)
def generate_widgets_from_signature(self, items: list, signature: dict = None) -> list:
"""
Generates widgets from the given list of items.
Args:
items(list): List of items to create widgets for.
signature(dict, optional): Scan signature dictionary from BEC.
Returns:
list: List of widgets created from the given items.
"""
widgets = [] # Initialize an empty list to hold the widgets
for item in items:
if signature:
# If a signature is provided, extract type and name from it
kwarg_info = next((info for info in signature if info["name"] == item), None)
if kwarg_info:
item_type = kwarg_info.get("annotation", "_empty")
item_name = item
else:
# If no signature is provided, assume the item is a tuple of (name, type)
item_name, item_type = item
widget_class = self.WIDGET_HANDLER.get(item_type, None)
if widget_class is None:
print(f"Unsupported annotation '{item_type}' for parameter '{item_name}'")
continue
# Instantiate the widget and set some properties if necessary
widget = widget_class()
# set high default range for spin boxes #TODO can be linked to motor/device limits from BEC
if isinstance(widget, (QSpinBox, QDoubleSpinBox)):
widget.setRange(-9999, 9999)
widget.setValue(0)
# Add the widget to the list
widgets.append(widget)
return widgets
def set_args_table_limits(self, table: QTableWidget, scan_info: dict) -> None:
# Get bundle info
arg_bundle_size = scan_info.get("arg_bundle_size", {})
self.arg_size_min = arg_bundle_size.get("min", 1)
self.arg_size_max = arg_bundle_size.get("max", None)
# Clear the previous input fields
table.setRowCount(0) # Wipe table
def add_widgets_row_to_layout(
self, grid_layout: QGridLayout, widgets: list, row_index: int = None
) -> None:
"""
Adds a row of widgets to the given grid layout.
Args:
grid_layout (QGridLayout): The grid layout to which widgets will be added.
items (list): List of parameter names to create widgets for.
row_index (int): The row index where the widgets should be added.
"""
# If row_index is not specified, add to the next available row
if row_index is None:
row_index = grid_layout.rowCount()
for column_index, widget in enumerate(widgets):
# Add the widget to the grid layout at the specified row and column
grid_layout.addWidget(widget, row_index, column_index)
def add_widgets_row_to_table(
self, table_widget: QTableWidget, widgets: list, row_index: int = None
) -> None:
"""
Adds a row of widgets to the given QTableWidget.
Args:
table_widget (QTableWidget): The table widget to which widgets will be added.
widgets (list): List of widgets to add to the table.
row_index (int): The row index where the widgets should be added. If None, add to the end.
"""
# If row_index is not specified, add to the end of the table
if row_index is None or row_index > table_widget.rowCount():
row_index = table_widget.rowCount()
if self.arg_size_max is not None: # ensure the max args size is not exceeded
if row_index >= self.arg_size_max:
return
table_widget.insertRow(row_index)
for column_index, widget in enumerate(widgets):
# If the widget is a subclass of QWidget, use setCellWidget
if issubclass(type(widget), QWidget):
table_widget.setCellWidget(row_index, column_index, widget)
else:
# Otherwise, assume it's a string or some other value that should be displayed as text
item = QTableWidgetItem(str(widget))
table_widget.setItem(row_index, column_index, item)
# Optionally, adjust the row height based on the content #TODO decide if needed
table_widget.setRowHeight(
row_index,
max(widget.sizeHint().height() for widget in widgets if isinstance(widget, QWidget)),
)
def remove_last_row_from_table(self, table_widget: QTableWidget) -> None:
"""
Removes the last row from the given QTableWidget until only one row is left.
Args:
table_widget (QTableWidget): The table widget from which the last row will be removed.
"""
row_count = table_widget.rowCount()
if (
row_count > self.arg_size_min
): # Check to ensure there is a minimum number of rows remaining
table_widget.removeRow(row_count - 1)
def create_new_grid_layout(self):
new_layout = QGridLayout()
# TODO maybe setup other layouts properties here?
return new_layout
def clear_and_delete_layout(self, layout: QLayout):
"""
Clears and deletes the given layout and all its child widgets.
Args:
layout(QLayout): Layout to clear and delete
"""
if layout is not None:
while layout.count():
item = layout.takeAt(0)
widget = item.widget()
if widget:
widget.deleteLater()
else:
sub_layout = item.layout()
if sub_layout:
self.clear_and_delete_layout(sub_layout)
layout.deleteLater()
def add_bundle(self) -> None:
"""Adds a new bundle to the scan control layout"""
# Get widgets used for particular scan and save them to be able to use for adding bundles
args_widgets = self.generate_widgets_from_signature(
self.arg_input.items()
) # TODO decide if make sense to put widget list into method parameters
# Add first widgets row to the table
self.add_widgets_row_to_table(self.args_table, args_widgets)
def remove_bundle(self) -> None:
"""Removes the last bundle from the scan control layout"""
self.remove_last_row_from_table(self.args_table)
def extract_kwargs_from_grid_row(self, grid_layout: QGridLayout, row: int) -> dict:
kwargs = {}
for column in range(grid_layout.columnCount()):
label_item = grid_layout.itemAtPosition(row, column)
if label_item is not None:
label_widget = label_item.widget()
if isinstance(label_widget, QLabel):
key = label_widget.text()
# The corresponding value widget is in the next row
value_item = grid_layout.itemAtPosition(row + 1, column)
if value_item is not None:
value_widget = value_item.widget()
# Use WidgetIO.get_value to extract the value
value = WidgetIO.get_value(value_widget)
kwargs[key] = value
return kwargs
def extract_args_from_table(self, table: QTableWidget) -> list:
"""
Extracts the arguments from the given table widget.
Args:
table(QTableWidget): Table widget from which to extract the arguments
"""
args = []
for row in range(table.rowCount()):
row_args = []
for column in range(table.columnCount()):
widget = table.cellWidget(row, column)
if widget:
if isinstance(widget, QLineEdit): # special case for QLineEdit for Devices
value = widget.text().lower()
if value in self.dev:
value = getattr(self.dev, value)
else:
raise ValueError(f"The device '{value}' is not recognized.")
else:
value = WidgetIO.get_value(widget)
row_args.append(value)
args.extend(row_args)
return args
def remove_kwarg_boxes(self):
for box in self.kwarg_boxes:
self.layout.removeWidget(box)
box.deleteLater()
self.kwarg_boxes = []
def run_scan(self):
# Extract kwargs for the scan
kwargs = {
k.lower(): v
for k, v in self.extract_kwargs_from_grid_row(self.kwargs_layout, 1).items()
}
# Extract args from the table
args = self.extract_args_from_table(self.args_table)
# Convert args to lowercase if they are strings
args = [arg.lower() if isinstance(arg, str) else arg for arg in args]
# Execute the scan
args = []
kwargs = {}
if self.arg_box is not None:
args = self.arg_box.get_parameters()
for box in self.kwarg_boxes:
box_kwargs = box.get_parameters()
kwargs.update(box_kwargs)
scan_function = getattr(self.scans, self.comboBox_scan_selection.currentText())
if callable(scan_function):
scan_function(*args, **kwargs)
def cleanup(self):
self.button_stop_scan.cleanup()
if self.arg_box:
for widget in self.arg_box.widgets:
if hasattr(widget, "cleanup"):
widget.cleanup()
for kwarg_box in self.kwarg_boxes:
for widget in kwarg_box.widgets:
if hasattr(widget, "cleanup"):
widget.cleanup()
super().cleanup()
def closeEvent(self, event):
self.cleanup()
QWidget().closeEvent(event)
# Application example
if __name__ == "__main__": # pragma: no cover
# BECclient global variables
client = BECDispatcher().client
client.start()
app = QApplication([])
scan_control = ScanControl(client=client) # allowed_scans=["line_scan", "grid_scan"])
scan_control = ScanControl()
qdarktheme.setup_theme("auto")
window = scan_control
window.show()
app.exec()

View File

@@ -0,0 +1,223 @@
from typing import Literal
from qtpy.QtWidgets import (
QCheckBox,
QComboBox,
QDoubleSpinBox,
QGridLayout,
QGroupBox,
QLabel,
QLineEdit,
QSpinBox,
)
from bec_widgets.utils.widget_io import WidgetIO
from bec_widgets.widgets.device_inputs import DeviceLineEdit
class ScanArgType:
DEVICE = "device"
FLOAT = "float"
INT = "int"
BOOL = "bool"
STR = "str"
DEVICEBASE = "DeviceBase"
LITERALS = "dict"
class ScanSpinBox(QSpinBox):
def __init__(
self, parent=None, arg_name: str = None, default: int | None = None, *args, **kwargs
):
super().__init__(parent=parent, *args, **kwargs)
self.arg_name = arg_name
self.setRange(-9999, 9999)
if default is not None:
self.setValue(default)
class ScanDoubleSpinBox(QDoubleSpinBox):
def __init__(
self, parent=None, arg_name: str = None, default: float | None = None, *args, **kwargs
):
super().__init__(parent=parent, *args, **kwargs)
self.arg_name = arg_name
self.setRange(-9999, 9999)
if default is not None:
self.setValue(default)
class ScanLineEdit(QLineEdit):
def __init__(
self, parent=None, arg_name: str = None, default: str | None = None, *args, **kwargs
):
super().__init__(parent=parent, *args, **kwargs)
self.arg_name = arg_name
if default is not None:
self.setText(default)
class ScanCheckBox(QCheckBox):
def __init__(
self, parent=None, arg_name: str = None, default: bool | None = None, *args, **kwargs
):
super().__init__(parent=parent, *args, **kwargs)
self.arg_name = arg_name
if default is not None:
self.setChecked(default)
class ScanGroupBox(QGroupBox):
WIDGET_HANDLER = {
ScanArgType.DEVICE: DeviceLineEdit,
ScanArgType.DEVICEBASE: DeviceLineEdit,
ScanArgType.FLOAT: ScanDoubleSpinBox,
ScanArgType.INT: ScanSpinBox,
ScanArgType.BOOL: ScanCheckBox,
ScanArgType.STR: ScanLineEdit,
ScanArgType.LITERALS: QComboBox, # TODO figure out combobox logic
}
def __init__(
self,
parent=None,
box_type=Literal["args", "kwargs"],
config: dict | None = None,
*args,
**kwargs,
):
super().__init__(parent=parent, *args, **kwargs)
self.config = config
self.box_type = box_type
self.layout = QGridLayout(self)
self.labels = []
self.widgets = []
self.init_box(self.config)
def init_box(self, config: dict):
box_name = config.get("name", "ScanGroupBox")
self.inputs = config.get("inputs", {})
self.setTitle(box_name)
# Labels
self.add_input_labels(self.inputs, 0)
# Widgets
if self.box_type == "args":
min_bundle = self.config.get("min", 1)
for i in range(1, min_bundle + 1):
self.add_input_widgets(self.inputs, i)
else:
self.add_input_widgets(self.inputs, 1)
def add_input_labels(self, group_inputs: dict, row: int) -> None:
"""
Adds the given arg_group from arg_bundle to the scan control layout. The input labels are always added to the first row.
Args:
group(dict): Dictionary containing the arg_group information.
"""
for column_index, item in enumerate(group_inputs):
arg_name = item.get("name", None)
display_name = item.get("display_name", arg_name)
label = QLabel(text=display_name)
self.layout.addWidget(label, row, column_index)
self.labels.append(label)
def add_input_widgets(self, group_inputs: dict, row) -> None:
"""
Adds the given arg_group from arg_bundle to the scan control layout.
Args:
group_inputs(dict): Dictionary containing the arg_group information.
row(int): The row to add the widgets to.
"""
for column_index, item in enumerate(group_inputs):
arg_name = item.get("name", None)
default = item.get("default", None)
widget = self.WIDGET_HANDLER.get(item["type"], None)
if widget is None:
print(f"Unsupported annotation '{item['type']}' for parameter '{item['name']}'")
continue
if default == "_empty":
default = None
widget_to_add = widget(arg_name=arg_name, default=default)
tooltip = item.get("tooltip", None)
if tooltip is not None:
widget_to_add.setToolTip(item["tooltip"])
self.layout.addWidget(widget_to_add, row, column_index)
self.widgets.append(widget_to_add)
def add_widget_bundle(self):
"""
Adds a new row of widgets to the scan control layout. Only usable for arg_groups.
"""
if self.box_type != "args":
return
arg_max = self.config.get("max", None)
row = self.layout.rowCount()
if arg_max is not None and row >= arg_max:
return
self.add_input_widgets(self.inputs, row)
def remove_widget_bundle(self):
"""
Removes the last row of widgets from the scan control layout. Only usable for arg_groups.
"""
if self.box_type != "args":
return
arg_min = self.config.get("min", None)
row = self.count_arg_rows()
if arg_min is not None and row <= arg_min:
return
for widget in self.widgets[-len(self.inputs) :]:
widget.deleteLater()
self.widgets = self.widgets[: -len(self.inputs)]
def get_parameters(self):
"""
Returns the parameters from the widgets in the scan control layout formated to run scan from BEC.
"""
if self.box_type == "args":
return self._get_arg_parameterts()
elif self.box_type == "kwargs":
return self._get_kwarg_parameters()
def _get_arg_parameterts(self):
args = []
for i in range(1, self.layout.rowCount()):
for j in range(self.layout.columnCount()):
widget = self.layout.itemAtPosition(i, j).widget()
if isinstance(widget, DeviceLineEdit):
value = widget.get_device()
else:
value = WidgetIO.get_value(widget)
args.append(value)
return args
def _get_kwarg_parameters(self):
kwargs = {}
for i in range(self.layout.columnCount()):
widget = self.layout.itemAtPosition(1, i).widget()
if isinstance(widget, DeviceLineEdit):
value = widget.get_device()
else:
value = WidgetIO.get_value(widget)
kwargs[widget.arg_name] = value
return kwargs
def count_arg_rows(self):
widget_rows = 0
for row in range(self.layout.rowCount()):
for col in range(self.layout.columnCount()):
item = self.layout.itemAtPosition(row, col)
if item is not None:
widget = item.widget()
if widget is not None:
if isinstance(widget, DeviceLineEdit):
widget_rows += 1
return widget_rows

View File

View File

@@ -0,0 +1,86 @@
import os
import select
import shlex
import signal
import subprocess
import sys
from bec_widgets.widgets.website.website import WebsiteWidget
class VSCodeEditor(WebsiteWidget):
"""
A widget to display the VSCode editor.
"""
token = "bec"
host = "127.0.0.1"
port = 7000
USER_ACCESS = []
def __init__(self, parent=None, config=None, client=None, gui_id=None):
self.process = None
self._url = f"http://{self.host}:{self.port}?tkn={self.token}"
super().__init__(parent=parent, config=config, client=client, gui_id=gui_id)
self.start_server()
def start_server(self):
"""
Start the server.
This method starts the server for the VSCode editor in a subprocess.
"""
cmd = shlex.split(
f"code serve-web --port {self.port} --connection-token={self.token} --accept-server-license-terms"
)
self.process = subprocess.Popen(
cmd, text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, preexec_fn=os.setsid
)
os.set_blocking(self.process.stdout.fileno(), False)
while self.process.poll() is None:
readylist, _, _ = select.select([self.process.stdout], [], [], 1)
if self.process.stdout in readylist:
output = self.process.stdout.read(1024)
if output and f"available at {self._url}" in output:
break
self.set_url(self._url)
def closeEvent(self, event):
"""
Hook for the close event to terminate the server.
"""
self.cleanup_vscode()
super().closeEvent(event)
def cleanup_vscode(self):
"""
Cleanup the VSCode editor.
"""
if not self.process:
return
os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
self.process.wait()
def cleanup(self):
"""
Cleanup the widget. This method is called from the dock area when the widget is removed.
"""
self.cleanup_vscode()
return super().cleanup()
if __name__ == "__main__": # pragma: no cover
import sys
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
widget = VSCodeEditor()
widget.show()
app.exec_()
widget.bec_dispatcher.disconnect_all()
widget.client.shutdown()

View File

@@ -1,10 +1,19 @@
from qtpy.QtCore import QUrl
from qtpy.QtCore import QUrl, qInstallMessageHandler
from qtpy.QtWebEngineWidgets import QWebEngineView
from qtpy.QtWidgets import QApplication
from bec_widgets.utils import BECConnector
def suppress_qt_messages(type_, context, msg):
if context.category in ["js", "default"]:
return
print(msg)
qInstallMessageHandler(suppress_qt_messages)
class WebsiteWidget(BECConnector, QWebEngineView):
"""
A simple widget to display a website

View File

@@ -2,7 +2,7 @@
# BEC Status Box
**Purpose:**
The [BECStatusBox]](/api_reference/_autosummary/bec_widgets.cli.client.BECStatusBox) Widget is a widget that allows you to monitor the status/health of the all running BEC processes. The widget generates the view automatically and updates the status of the processes in real-time. The top level indicates the overall state of the BEC core services (DeviceServer, ScanServer, SciHub, ScanBundler and FileWriter), but you can also see the status of each individual process by opening the collapsed view. In the collapsed view, you can double click on each process to get a popup window with live updates of the metrics for each process in real-time.
The [BECStatusBox](/api_reference/_autosummary/bec_widgets.cli.client.BECStatusBox) is a widget that allows you to monitor the status/health of the all running BEC processes. The widget generates the view automatically and updates the status of the processes in real-time. The top level indicates the overall state of the BEC core services (DeviceServer, ScanServer, SciHub, ScanBundler and FileWriter), but you can also see the status of each individual process by opening the collapsed view. In the collapsed view, you can double click on each process to get a popup window with live updates of the metrics for each process in real-time.
**Key Features:**

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.4 MiB

View File

@@ -0,0 +1,35 @@
(user.widgets.scan_control)=
# Scan Control
**Purpose:**
The `ScanControl` widget is designed to generate a graphical user interface (GUI) to control various scan operations
based on the scan's signature and `gui_config`. The widget is used to control the scan operations, such as starting,
stopping, and pausing the scan. The widget also provides a graphical representation of the scan progress and the scan
status. The widget is designed to be used in conjunction with the `ScanServer` and `ScanBundler` services from the BEC
core services.
By default the widget supports only the scans which have defined `gui_config` and are inhereted from these scan classes:
- [ScanBase](https://beamline-experiment-control.readthedocs.io/en/latest/api_reference/_autosummary/bec_server.scan_server.scans.ScanBase.html)
- [SyncFlyScanBase](https://beamline-experiment-control.readthedocs.io/en/latest/api_reference/_autosummary/bec_server.scan_server.scans.SyncFlyScanBase.html)
- [AsyncFlyScanBase](https://beamline-experiment-control.readthedocs.io/en/latest/api_reference/_autosummary/bec_server.scan_server.scans.AsyncFlyScanBase.html)
**Key Features:**
- Automatically generates a control interface based on scan signatures and `gui_config`.
- Supports adding and removing argument bundles dynamically.
- Provides a visual representation of scan parameters grouped by functionality.
- Integrates start and stop controls for executing and halting scans.
**Example of Use:**
**Code example:**
The following code snipped demonstrates how to create a `ScanControl` widget using BEC Widgets within `BECIPythonClient`
![ScanControl](./scan_control.gif)
```python
scan_control = gui.add_dock().add_widget("ScanControl")
```

View File

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "bec_widgets"
version = "0.68.0"
version = "0.72.0"
description = "BEC Widgets"
requires-python = ">=3.10"
classifiers = [
@@ -49,6 +49,7 @@ Homepage = "https://gitlab.psi.ch/bec/bec_widgets"
[project.scripts]
bw-generate-cli = "bec_widgets.cli.generate_cli:main"
bec-gui-server = "bec_widgets.cli.server:main"
bec-designer = "bec_widgets.utils.bec_designer:main"
[tool.hatch.build.targets.wheel]
include = ["*"]

View File

@@ -0,0 +1,71 @@
import time
import pytest
from bec_widgets.utils.widget_io import WidgetIO
from bec_widgets.widgets.scan_control import ScanControl
@pytest.fixture(scope="function")
def scan_control(qtbot, bec_client_lib): # , mock_dev):
widget = ScanControl(client=bec_client_lib)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
yield widget
def test_scan_control_populate_scans_e2e(scan_control):
expected_scans = [
"grid_scan",
"fermat_scan",
"round_scan",
"cont_line_scan",
"cont_line_fly_scan",
"round_scan_fly",
"round_roi_scan",
"time_scan",
"monitor_scan",
"acquire",
"line_scan",
]
items = [
scan_control.comboBox_scan_selection.itemText(i)
for i in range(scan_control.comboBox_scan_selection.count())
]
assert scan_control.comboBox_scan_selection.count() == len(expected_scans)
assert sorted(items) == sorted(expected_scans)
def test_run_line_scan_with_parameters_e2e(scan_control, bec_client_lib, qtbot):
client = bec_client_lib
queue = client.queue
scan_name = "line_scan"
kwargs = {"exp_time": 0.01, "steps": 10, "relative": True, "burst_at_each_point": 1}
args = {"device": "samx", "start": -5, "stop": 5}
scan_control.comboBox_scan_selection.setCurrentText(scan_name)
# Set kwargs in the UI
for kwarg_box in scan_control.kwarg_boxes:
for widget in kwarg_box.widgets:
for key, value in kwargs.items():
if widget.arg_name == key:
WidgetIO.set_value(widget, value)
break
# Set args in the UI
for widget in scan_control.arg_box.widgets:
for key, value in args.items():
if widget.arg_name == key:
WidgetIO.set_value(widget, value)
break
# Run the scan
scan_control.button_run_scan.click()
time.sleep(2)
last_scan = queue.scan_storage.storage[-1]
assert last_scan.status_message.info["scan_name"] == scan_name
assert last_scan.status_message.info["exp_time"] == kwargs["exp_time"]
assert last_scan.status_message.info["scan_motors"] == [args["device"]]
assert last_scan.status_message.info["num_points"] == kwargs["steps"]

View File

@@ -1,5 +1,9 @@
# pylint: disable = no-name-in-module,missing-class-docstring, missing-module-docstring
import time
import pytest
from qtpy.QtCore import Slot
from qtpy.QtWidgets import QApplication
from bec_widgets.utils import BECConnector, ConnectionConfig
@@ -55,3 +59,22 @@ def test_bec_connector_update_client(bec_connector, mocked_client):
def test_bec_connector_get_config(bec_connector):
assert bec_connector.get_config(dict_output=False) == bec_connector.config
assert bec_connector.get_config() == bec_connector.config.model_dump()
def test_bec_connector_submit_task(bec_connector):
def test_func():
time.sleep(2)
print("done")
completed = False
@Slot()
def complete_func():
nonlocal completed
completed = True
bec_connector.submit_task(test_func, on_complete=complete_func)
assert not completed
while not completed:
QApplication.processEvents()
time.sleep(0.1)

View File

@@ -1,152 +1,152 @@
import re
from unittest import mock
import pytest
from bec_lib.messages import BECStatus, ServiceMetricMessage, StatusMessage
from qtpy.QtCore import QMetaMethod
from bec_widgets.widgets.bec_status_box.bec_status_box import BECServiceInfoContainer, BECStatusBox
from .client_mocks import mocked_client
@pytest.fixture
def status_box(qtbot, mocked_client):
with mock.patch(
"bec_widgets.widgets.bec_status_box.bec_status_box.BECServiceStatusMixin"
) as mock_service_status_mixin:
widget = BECStatusBox(client=mocked_client)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
yield widget
def test_status_box_init(qtbot, mocked_client):
with mock.patch(
"bec_widgets.widgets.bec_status_box.bec_status_box.BECServiceStatusMixin"
) as mock_service_status_mixin:
name = "my test"
widget = BECStatusBox(parent=None, service_name=name, client=mocked_client)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
assert widget.headerItem().DontShowIndicator.value == 1
assert widget.children()[0].children()[0].config.service_name == name
def test_update_top_item(qtbot, mocked_client):
with (
mock.patch(
"bec_widgets.widgets.bec_status_box.bec_status_box.BECServiceStatusMixin"
) as mock_service_status_mixin,
mock.patch(
"bec_widgets.widgets.bec_status_box.status_item.StatusItem.update_config"
) as mock_update,
):
name = "my test"
widget = BECStatusBox(parent=None, service_name=name, client=mocked_client)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
widget.update_top_item_status(status="RUNNING")
assert widget.bec_service_info_container[name].status == "RUNNING"
assert mock_update.call_args == mock.call(widget.bec_service_info_container[name].dict())
def test_create_status_widget(status_box):
name = "test_service"
status = BECStatus.IDLE
info = {"test": "test"}
metrics = {"metric": "test_metric"}
item = status_box._create_status_widget(name, status, info, metrics)
assert item.config.service_name == name
assert item.config.status == status.name
assert item.config.info == info
assert item.config.metrics == metrics
def test_bec_service_container(status_box):
name = "test_service"
status = BECStatus.IDLE
info = {"test": "test"}
metrics = {"metric": "test_metric"}
expected_return = BECServiceInfoContainer(
service_name=name, status=status, info=info, metrics=metrics
)
assert status_box.service_name in status_box.bec_service_info_container
assert len(status_box.bec_service_info_container) == 1
status_box._update_bec_service_container(name, status, info, metrics)
assert len(status_box.bec_service_info_container) == 2
assert status_box.bec_service_info_container[name] == expected_return
def test_add_tree_item(status_box):
name = "test_service"
status = BECStatus.IDLE
info = {"test": "test"}
metrics = {"metric": "test_metric"}
assert len(status_box.children()[0].children()) == 1
status_box.add_tree_item(name, status, info, metrics)
assert len(status_box.children()[0].children()) == 2
assert name in status_box.tree_items
def test_update_service_status(status_box):
"""Also checks check redundant tree items"""
name = "test_service"
status = BECStatus.IDLE
info = {"test": "test"}
metrics = {"metric": "test_metric"}
status_box.add_tree_item(name, status, info, {})
not_connected_name = "invalid_service"
status_box.add_tree_item(not_connected_name, status, info, metrics)
services_status = {name: StatusMessage(name=name, status=status, info=info)}
services_metrics = {name: ServiceMetricMessage(name=name, metrics=metrics)}
with mock.patch.object(status_box, "update_core_services", return_value=services_status):
assert not_connected_name in status_box.tree_items
status_box.update_service_status(services_status, services_metrics)
assert status_box.tree_items[name][1].config.metrics == metrics
assert not_connected_name not in status_box.tree_items
def test_update_core_services(qtbot, mocked_client):
with (
mock.patch(
"bec_widgets.widgets.bec_status_box.bec_status_box.BECServiceStatusMixin"
) as mock_service_status_mixin,
mock.patch(
"bec_widgets.widgets.bec_status_box.bec_status_box.BECStatusBox.update_top_item_status"
) as mock_update,
):
name = "my test"
status_box = BECStatusBox(parent=None, service_name=name, client=mocked_client)
qtbot.addWidget(status_box)
qtbot.waitExposed(status_box)
status_box.CORE_SERVICES = ["test_service"]
name = "test_service"
status = BECStatus.RUNNING
info = {"test": "test"}
metrics = {"metric": "test_metric"}
services_status = {name: StatusMessage(name=name, status=status, info=info)}
services_metrics = {name: ServiceMetricMessage(name=name, metrics=metrics)}
status_box.update_core_services(services_status, services_metrics)
assert mock_update.call_args == mock.call(status.name)
status = BECStatus.IDLE
services_status = {name: StatusMessage(name=name, status=status, info=info)}
services_metrics = {name: ServiceMetricMessage(name=name, metrics=metrics)}
status_box.update_core_services(services_status, services_metrics)
assert mock_update.call_args == mock.call("ERROR")
def test_double_click_item(status_box):
name = "test_service"
status = BECStatus.IDLE
info = {"test": "test"}
metrics = {"MyData": "This should be shown nicely"}
status_box.add_tree_item(name, status, info, metrics)
item, status_item = status_box.tree_items[name]
with mock.patch.object(status_item, "show_popup") as mock_show_popup:
status_box.itemDoubleClicked.emit(item, 0)
assert mock_show_popup.call_count == 1
# import re
# from unittest import mock
#
# import pytest
# from bec_lib.messages import BECStatus, ServiceMetricMessage, StatusMessage
# from qtpy.QtCore import QMetaMethod
#
# from bec_widgets.widgets.bec_status_box.bec_status_box import BECServiceInfoContainer, BECStatusBox
#
# from .client_mocks import mocked_client
#
#
# @pytest.fixture
# def status_box(qtbot, mocked_client):
# with mock.patch(
# "bec_widgets.widgets.bec_status_box.bec_status_box.BECServiceStatusMixin"
# ) as mock_service_status_mixin:
# widget = BECStatusBox(client=mocked_client)
# qtbot.addWidget(widget)
# qtbot.waitExposed(widget)
# yield widget
#
#
# def test_status_box_init(qtbot, mocked_client):
# with mock.patch(
# "bec_widgets.widgets.bec_status_box.bec_status_box.BECServiceStatusMixin"
# ) as mock_service_status_mixin:
# name = "my test"
# widget = BECStatusBox(parent=None, service_name=name, client=mocked_client)
# qtbot.addWidget(widget)
# qtbot.waitExposed(widget)
# assert widget.headerItem().DontShowIndicator.value == 1
# assert widget.children()[0].children()[0].config.service_name == name
#
#
# def test_update_top_item(qtbot, mocked_client):
# with (
# mock.patch(
# "bec_widgets.widgets.bec_status_box.bec_status_box.BECServiceStatusMixin"
# ) as mock_service_status_mixin,
# mock.patch(
# "bec_widgets.widgets.bec_status_box.status_item.StatusItem.update_config"
# ) as mock_update,
# ):
# name = "my test"
# widget = BECStatusBox(parent=None, service_name=name, client=mocked_client)
# qtbot.addWidget(widget)
# qtbot.waitExposed(widget)
# widget.update_top_item_status(status="RUNNING")
# assert widget.bec_service_info_container[name].status == "RUNNING"
# assert mock_update.call_args == mock.call(widget.bec_service_info_container[name].dict())
#
#
# def test_create_status_widget(status_box):
# name = "test_service"
# status = BECStatus.IDLE
# info = {"test": "test"}
# metrics = {"metric": "test_metric"}
# item = status_box._create_status_widget(name, status, info, metrics)
# assert item.config.service_name == name
# assert item.config.status == status.name
# assert item.config.info == info
# assert item.config.metrics == metrics
#
#
# def test_bec_service_container(status_box):
# name = "test_service"
# status = BECStatus.IDLE
# info = {"test": "test"}
# metrics = {"metric": "test_metric"}
# expected_return = BECServiceInfoContainer(
# service_name=name, status=status, info=info, metrics=metrics
# )
# assert status_box.service_name in status_box.bec_service_info_container
# assert len(status_box.bec_service_info_container) == 1
# status_box._update_bec_service_container(name, status, info, metrics)
# assert len(status_box.bec_service_info_container) == 2
# assert status_box.bec_service_info_container[name] == expected_return
#
#
# def test_add_tree_item(status_box):
# name = "test_service"
# status = BECStatus.IDLE
# info = {"test": "test"}
# metrics = {"metric": "test_metric"}
# assert len(status_box.children()[0].children()) == 1
# status_box.add_tree_item(name, status, info, metrics)
# assert len(status_box.children()[0].children()) == 2
# assert name in status_box.tree_items
#
#
# def test_update_service_status(status_box):
# """Also checks check redundant tree items"""
# name = "test_service"
# status = BECStatus.IDLE
# info = {"test": "test"}
# metrics = {"metric": "test_metric"}
# status_box.add_tree_item(name, status, info, {})
# not_connected_name = "invalid_service"
# status_box.add_tree_item(not_connected_name, status, info, metrics)
#
# services_status = {name: StatusMessage(name=name, status=status, info=info)}
# services_metrics = {name: ServiceMetricMessage(name=name, metrics=metrics)}
#
# with mock.patch.object(status_box, "update_core_services", return_value=services_status):
# assert not_connected_name in status_box.tree_items
# status_box.update_service_status(services_status, services_metrics)
# assert status_box.tree_items[name][1].config.metrics == metrics
# assert not_connected_name not in status_box.tree_items
#
#
# def test_update_core_services(qtbot, mocked_client):
# with (
# mock.patch(
# "bec_widgets.widgets.bec_status_box.bec_status_box.BECServiceStatusMixin"
# ) as mock_service_status_mixin,
# mock.patch(
# "bec_widgets.widgets.bec_status_box.bec_status_box.BECStatusBox.update_top_item_status"
# ) as mock_update,
# ):
# name = "my test"
# status_box = BECStatusBox(parent=None, service_name=name, client=mocked_client)
# qtbot.addWidget(status_box)
# qtbot.waitExposed(status_box)
# status_box.CORE_SERVICES = ["test_service"]
# name = "test_service"
# status = BECStatus.RUNNING
# info = {"test": "test"}
# metrics = {"metric": "test_metric"}
# services_status = {name: StatusMessage(name=name, status=status, info=info)}
# services_metrics = {name: ServiceMetricMessage(name=name, metrics=metrics)}
#
# status_box.update_core_services(services_status, services_metrics)
# assert mock_update.call_args == mock.call(status.name)
#
# status = BECStatus.IDLE
# services_status = {name: StatusMessage(name=name, status=status, info=info)}
# services_metrics = {name: ServiceMetricMessage(name=name, metrics=metrics)}
# status_box.update_core_services(services_status, services_metrics)
# assert mock_update.call_args == mock.call("ERROR")
#
#
# def test_double_click_item(status_box):
# name = "test_service"
# status = BECStatus.IDLE
# info = {"test": "test"}
# metrics = {"MyData": "This should be shown nicely"}
# status_box.add_tree_item(name, status, info, metrics)
# item, status_item = status_box.tree_items[name]
# with mock.patch.object(status_item, "show_popup") as mock_show_popup:
# status_box.itemDoubleClicked.emit(item, 0)
# assert mock_show_popup.call_count == 1

View File

@@ -17,7 +17,7 @@ def test_device_input_base_init(device_input_base):
assert isinstance(device_input_base, DeviceInputBase)
assert device_input_base.config.widget_class == "DeviceInputBase"
assert device_input_base.config.device_filter is None
assert device_input_base.config.default_device is None
assert device_input_base.config.default is None
assert device_input_base.devices == []
@@ -26,12 +26,12 @@ def test_device_input_base_init_with_config(mocked_client):
"widget_class": "DeviceInputBase",
"gui_id": "test_gui_id",
"device_filter": "FakePositioner",
"default_device": "samx",
"default": "samx",
}
widget = DeviceInputBase(client=mocked_client, config=config)
assert widget.config.gui_id == "test_gui_id"
assert widget.config.device_filter == "FakePositioner"
assert widget.config.default_device == "samx"
assert widget.config.default == "samx"
def test_device_input_base_set_device_filter(device_input_base):
@@ -47,7 +47,7 @@ def test_device_input_base_set_device_filter_error(device_input_base):
def test_device_input_base_set_default_device(device_input_base):
device_input_base.set_default_device("samx")
assert device_input_base.config.default_device == "samx"
assert device_input_base.config.default == "samx"
def test_device_input_base_set_default_device_error(device_input_base):

View File

@@ -21,7 +21,7 @@ def device_input_combobox_with_config(qtbot, mocked_client):
"widget_class": "DeviceComboBox",
"gui_id": "test_gui_id",
"device_filter": "FakePositioner",
"default_device": "samx",
"default": "samx",
"arg_name": "test_arg_name",
}
widget = DeviceComboBox(client=mocked_client, config=config)
@@ -37,7 +37,7 @@ def device_input_combobox_with_kwargs(qtbot, mocked_client):
client=mocked_client,
gui_id="test_gui_id",
device_filter="FakePositioner",
default_device="samx",
default="samx",
arg_name="test_arg_name",
)
qtbot.addWidget(widget)
@@ -52,7 +52,7 @@ def test_device_input_combobox_init(device_input_combobox):
assert isinstance(device_input_combobox, DeviceComboBox)
assert device_input_combobox.config.widget_class == "DeviceComboBox"
assert device_input_combobox.config.device_filter is None
assert device_input_combobox.config.default_device is None
assert device_input_combobox.config.default is None
assert device_input_combobox.devices == [
"samx",
"samy",
@@ -72,14 +72,14 @@ def test_device_input_combobox_init(device_input_combobox):
def test_device_input_combobox_init_with_config(device_input_combobox_with_config):
assert device_input_combobox_with_config.config.gui_id == "test_gui_id"
assert device_input_combobox_with_config.config.device_filter == "FakePositioner"
assert device_input_combobox_with_config.config.default_device == "samx"
assert device_input_combobox_with_config.config.default == "samx"
assert device_input_combobox_with_config.config.arg_name == "test_arg_name"
def test_device_input_combobox_init_with_kwargs(device_input_combobox_with_kwargs):
assert device_input_combobox_with_kwargs.config.gui_id == "test_gui_id"
assert device_input_combobox_with_kwargs.config.device_filter == "FakePositioner"
assert device_input_combobox_with_kwargs.config.default_device == "samx"
assert device_input_combobox_with_kwargs.config.default == "samx"
assert device_input_combobox_with_kwargs.config.arg_name == "test_arg_name"
@@ -106,7 +106,7 @@ def device_input_line_edit_with_config(qtbot, mocked_client):
"widget_class": "DeviceLineEdit",
"gui_id": "test_gui_id",
"device_filter": "FakePositioner",
"default_device": "samx",
"default": "samx",
"arg_name": "test_arg_name",
}
widget = DeviceLineEdit(client=mocked_client, config=config)
@@ -122,7 +122,7 @@ def device_input_line_edit_with_kwargs(qtbot, mocked_client):
client=mocked_client,
gui_id="test_gui_id",
device_filter="FakePositioner",
default_device="samx",
default="samx",
arg_name="test_arg_name",
)
qtbot.addWidget(widget)
@@ -137,7 +137,7 @@ def test_device_input_line_edit_init(device_input_line_edit):
assert isinstance(device_input_line_edit, DeviceLineEdit)
assert device_input_line_edit.config.widget_class == "DeviceLineEdit"
assert device_input_line_edit.config.device_filter is None
assert device_input_line_edit.config.default_device is None
assert device_input_line_edit.config.default is None
assert device_input_line_edit.devices == [
"samx",
"samy",
@@ -157,14 +157,14 @@ def test_device_input_line_edit_init(device_input_line_edit):
def test_device_input_line_edit_init_with_config(device_input_line_edit_with_config):
assert device_input_line_edit_with_config.config.gui_id == "test_gui_id"
assert device_input_line_edit_with_config.config.device_filter == "FakePositioner"
assert device_input_line_edit_with_config.config.default_device == "samx"
assert device_input_line_edit_with_config.config.default == "samx"
assert device_input_line_edit_with_config.config.arg_name == "test_arg_name"
def test_device_input_line_edit_init_with_kwargs(device_input_line_edit_with_kwargs):
assert device_input_line_edit_with_kwargs.config.gui_id == "test_gui_id"
assert device_input_line_edit_with_kwargs.config.device_filter == "FakePositioner"
assert device_input_line_edit_with_kwargs.config.default_device == "samx"
assert device_input_line_edit_with_kwargs.config.default == "samx"
assert device_input_line_edit_with_kwargs.config.arg_name == "test_arg_name"

View File

@@ -2,131 +2,288 @@
from unittest.mock import MagicMock
import pytest
from qtpy.QtWidgets import QLineEdit
from bec_lib.messages import AvailableResourceMessage
from bec_widgets.utils.widget_io import WidgetIO
from bec_widgets.widgets.scan_control import ScanControl
from tests.unit_tests.test_msgs.available_scans_message import available_scans_message
from .client_mocks import mocked_client
class FakePositioner:
"""Fake minimal positioner class for testing."""
def __init__(self, name, enabled=True):
self.name = name
self.enabled = enabled
def __contains__(self, item):
return item == self.name
def get_mocked_device(device_name):
"""Helper function to mock the devices"""
if device_name == "samx":
return FakePositioner(name="samx", enabled=True)
@pytest.fixture(scope="function")
def mocked_client():
# Create a MagicMock object
client = MagicMock()
# Mock the producer.get method to return the packed message
client.producer.get.return_value = available_scans_message
# # Mock the device_manager.devices attribute to return a mock object for samx
client.device_manager.devices = MagicMock()
client.device_manager.devices.__contains__.side_effect = lambda x: x == "samx"
client.device_manager.devices.samx = get_mocked_device("samx")
return client
available_scans_message = AvailableResourceMessage(
resource={
"line_scan": {
"class": "LineScan",
"base_class": "ScanBase",
"arg_input": {"device": "device", "start": "float", "stop": "float"},
"gui_config": {
"scan_class_name": "LineScan",
"arg_group": {
"name": "Scan Arguments",
"bundle": 3,
"arg_inputs": {"device": "device", "start": "float", "stop": "float"},
"inputs": [
{
"arg": True,
"name": "device",
"type": "device",
"display_name": "Device",
"tooltip": None,
"default": None,
"expert": False,
},
{
"arg": True,
"name": "start",
"type": "float",
"display_name": "Start",
"tooltip": None,
"default": None,
"expert": False,
},
{
"arg": True,
"name": "stop",
"type": "float",
"display_name": "Stop",
"tooltip": None,
"default": None,
"expert": False,
},
],
"min": 1,
"max": None,
},
"kwarg_groups": [
{
"name": "Movement Parameters",
"inputs": [
{
"arg": False,
"name": "steps",
"type": "int",
"display_name": "Steps",
"tooltip": "Number of steps",
"default": None,
"expert": False,
},
{
"arg": False,
"name": "relative",
"type": "bool",
"display_name": "Relative",
"tooltip": "If True, the start and end positions are relative to the current position",
"default": False,
"expert": False,
},
],
},
{
"name": "Acquisition Parameters",
"inputs": [
{
"arg": False,
"name": "exp_time",
"type": "float",
"display_name": "Exp Time",
"tooltip": "Exposure time in s",
"default": 0,
"expert": False,
},
{
"arg": False,
"name": "burst_at_each_point",
"type": "int",
"display_name": "Burst At Each Point",
"tooltip": "Number of acquisition per point",
"default": 1,
"expert": False,
},
],
},
],
},
"required_kwargs": ["steps", "relative"],
"arg_bundle_size": {"bundle": 3, "min": 1, "max": None},
},
"grid_scan": {
"class": "Scan",
"base_class": "ScanBase",
"arg_input": {"device": "device", "start": "float", "stop": "float", "steps": "int"},
"gui_config": {
"scan_class_name": "Scan",
"arg_group": {
"name": "Scan Arguments",
"bundle": 4,
"arg_inputs": {
"device": "device",
"start": "float",
"stop": "float",
"steps": "int",
},
"inputs": [
{
"arg": True,
"name": "device",
"type": "device",
"display_name": "Device",
"tooltip": None,
"default": None,
"expert": False,
},
{
"arg": True,
"name": "start",
"type": "float",
"display_name": "Start",
"tooltip": None,
"default": None,
"expert": False,
},
{
"arg": True,
"name": "stop",
"type": "float",
"display_name": "Stop",
"tooltip": None,
"default": None,
"expert": False,
},
{
"arg": True,
"name": "steps",
"type": "int",
"display_name": "Steps",
"tooltip": None,
"default": None,
"expert": False,
},
],
"min": 2,
"max": None,
},
"kwarg_groups": [
{
"name": "Scan Parameters",
"inputs": [
{
"arg": False,
"name": "exp_time",
"type": "float",
"display_name": "Exp Time",
"tooltip": "Exposure time in seconds",
"default": 0,
"expert": False,
},
{
"arg": False,
"name": "settling_time",
"type": "float",
"display_name": "Settling Time",
"tooltip": "Settling time in seconds",
"default": 0,
"expert": False,
},
{
"arg": False,
"name": "burst_at_each_point",
"type": "int",
"display_name": "Burst At Each Point",
"tooltip": "Number of exposures at each point",
"default": 1,
"expert": False,
},
{
"arg": False,
"name": "relative",
"type": "bool",
"display_name": "Relative",
"tooltip": "If True, the motors will be moved relative to their current position",
"default": False,
"expert": False,
},
],
}
],
},
"required_kwargs": ["relative"],
"arg_bundle_size": {"bundle": 4, "min": 2, "max": None},
},
"not_supported_scan_class": {"base_class": "NotSupportedScanClass"},
}
)
@pytest.fixture(scope="function")
def scan_control(qtbot, mocked_client): # , mock_dev):
mocked_client.connector.set("scans/available_scans", available_scans_message)
widget = ScanControl(client=mocked_client)
# widget.dev.samx = MagicMock()
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
yield widget
def test_populate_scans(scan_control, mocked_client):
# The comboBox should be populated with all scan from the message right after initialization
expected_scans = available_scans_message.resource.keys()
assert scan_control.comboBox_scan_selection.count() == len(expected_scans)
for scan in expected_scans: # Each scan should be in the comboBox
assert scan_control.comboBox_scan_selection.findText(scan) != -1
@pytest.mark.parametrize(
"scan_name", ["line_scan", "grid_scan"]
) # TODO now only for line_scan and grid_scan, later for all loaded scans
def test_on_scan_selected(scan_control, scan_name):
# Expected scan info from the message signature
expected_scan_info = available_scans_message.resource[scan_name]
# Select a scan from the comboBox
scan_control.comboBox_scan_selection.setCurrentText(scan_name)
# Check labels and widgets in args table
for index, (arg_key, arg_value) in enumerate(expected_scan_info["arg_input"].items()):
label = scan_control.args_table.horizontalHeaderItem(index)
assert label.text().lower() == arg_key # labes
for row in range(expected_scan_info["arg_bundle_size"]["min"]):
widget = scan_control.args_table.cellWidget(row, index)
assert widget is not None # Confirm that a widget exists
expected_widget_type = scan_control.WIDGET_HANDLER.get(arg_value, None)
assert isinstance(widget, expected_widget_type) # Confirm the widget type matches
# kwargs
kwargs_from_signature = [
param for param in expected_scan_info["signature"] if param["kind"] == "KEYWORD_ONLY"
expected_scans = ["line_scan", "grid_scan"]
items = [
scan_control.comboBox_scan_selection.itemText(i)
for i in range(scan_control.comboBox_scan_selection.count())
]
# Check labels and widgets in kwargs grid layout
for index, kwarg_info in enumerate(kwargs_from_signature):
label_widget = scan_control.kwargs_layout.itemAtPosition(1, index).widget()
assert label_widget.text() == kwarg_info["name"].capitalize()
widget = scan_control.kwargs_layout.itemAtPosition(2, index).widget()
expected_widget_type = scan_control.WIDGET_HANDLER.get(kwarg_info["annotation"], QLineEdit)
assert isinstance(widget, expected_widget_type)
assert scan_control.comboBox_scan_selection.count() == 2
assert sorted(items) == sorted(expected_scans)
@pytest.mark.parametrize("scan_name", ["line_scan", "grid_scan"])
def test_add_remove_bundle(scan_control, scan_name):
# Expected scan info from the message signature
def test_on_scan_selected(scan_control, scan_name):
expected_scan_info = available_scans_message.resource[scan_name]
scan_control.comboBox_scan_selection.setCurrentText(scan_name)
# Select a scan from the comboBox
# Check arg_box labels and widgets
for index, (arg_key, arg_value) in enumerate(expected_scan_info["arg_input"].items()):
label = scan_control.arg_box.layout.itemAtPosition(0, index).widget()
assert label.text().lower() == arg_key
for row in range(1, expected_scan_info["arg_bundle_size"]["min"] + 1):
widget = scan_control.arg_box.layout.itemAtPosition(row, index).widget()
assert widget is not None # Confirm that a widget exists
expected_widget_type = scan_control.arg_box.WIDGET_HANDLER.get(arg_value, None)
assert isinstance(widget, expected_widget_type) # Confirm the widget type matches
# Check kwargs boxes
kwargs_group = [param for param in expected_scan_info["gui_config"]["kwarg_groups"]]
print(kwargs_group)
for kwarg_box, kwarg_group in zip(scan_control.kwarg_boxes, kwargs_group):
assert kwarg_box.title() == kwarg_group["name"]
for index, kwarg_info in enumerate(kwarg_group["inputs"]):
label = kwarg_box.layout.itemAtPosition(0, index).widget()
assert label.text() == kwarg_info["display_name"]
widget = kwarg_box.layout.itemAtPosition(1, index).widget()
expected_widget_type = kwarg_box.WIDGET_HANDLER.get(kwarg_info["type"], None)
assert isinstance(widget, expected_widget_type)
@pytest.mark.parametrize("scan_name", ["line_scan", "grid_scan"])
def test_add_remove_bundle(scan_control, scan_name, qtbot):
expected_scan_info = available_scans_message.resource[scan_name]
scan_control.comboBox_scan_selection.setCurrentText(scan_name)
# Initial number of args row
initial_num_of_rows = scan_control.args_table.rowCount()
initial_num_of_rows = scan_control.arg_box.count_arg_rows()
# Check initial row count of args table
assert scan_control.args_table.rowCount() == expected_scan_info["arg_bundle_size"]["min"]
assert initial_num_of_rows == expected_scan_info["arg_bundle_size"]["min"]
# Try to remove default number of args row
scan_control.pushButton_remove_bundle.click()
assert scan_control.args_table.rowCount() == expected_scan_info["arg_bundle_size"]["min"]
scan_control.button_add_bundle.click()
scan_control.button_add_bundle.click()
# Try to add two bundles
scan_control.pushButton_add_bundle.click()
scan_control.pushButton_add_bundle.click()
# check the case where no max number of args are defined
# TODO do check also for the case where max number of args are defined
if expected_scan_info["arg_bundle_size"]["max"] is None:
assert scan_control.args_table.rowCount() == initial_num_of_rows + 2
assert scan_control.arg_box.count_arg_rows() == initial_num_of_rows + 2
# Remove one bundle
scan_control.pushButton_remove_bundle.click()
scan_control.button_remove_bundle.click()
qtbot.wait(200)
# check the case where no max number of args are defined
if expected_scan_info["arg_bundle_size"]["max"] is None:
assert scan_control.args_table.rowCount() == initial_num_of_rows + 1
assert scan_control.arg_box.count_arg_rows() == initial_num_of_rows + 1
def test_run_line_scan_with_parameters(scan_control, mocked_client):
@@ -134,32 +291,21 @@ def test_run_line_scan_with_parameters(scan_control, mocked_client):
kwargs = {"exp_time": 0.1, "steps": 10, "relative": True, "burst_at_each_point": 1}
args = {"device": "samx", "start": -5, "stop": 5}
# Select a scan from the comboBox
scan_control.comboBox_scan_selection.setCurrentText(scan_name)
# Set kwargs in the UI
for label_index in range(
scan_control.kwargs_layout.rowCount() + 1
): # from some reason rowCount() returns 1 less than the actual number of rows
label_item = scan_control.kwargs_layout.itemAtPosition(1, label_index)
if label_item:
label_widget = label_item.widget()
kwarg_key = WidgetIO.get_value(label_widget).lower()
if kwarg_key in kwargs:
widget_item = scan_control.kwargs_layout.itemAtPosition(2, label_index)
if widget_item:
widget = widget_item.widget()
WidgetIO.set_value(widget, kwargs[kwarg_key])
for kwarg_box in scan_control.kwarg_boxes:
for widget in kwarg_box.widgets:
for key, value in kwargs.items():
if widget.arg_name == key:
WidgetIO.set_value(widget, value)
break
# Set args in the UI
for col_index in range(scan_control.args_table.columnCount()):
header_item = scan_control.args_table.horizontalHeaderItem(col_index)
if header_item:
arg_key = header_item.text().lower()
if arg_key in args:
for row_index in range(scan_control.args_table.rowCount()):
widget = scan_control.args_table.cellWidget(row_index, col_index)
WidgetIO.set_value(widget, args[arg_key])
for widget in scan_control.arg_box.widgets:
for key, value in args.items():
if widget.arg_name == key:
WidgetIO.set_value(widget, value)
break
# Mock the scan function
mocked_scan_function = MagicMock()
@@ -172,13 +318,7 @@ def test_run_line_scan_with_parameters(scan_control, mocked_client):
called_args, called_kwargs = mocked_scan_function.call_args
# Check if the scan function was called correctly
expected_device = (
mocked_client.device_manager.devices.samx
) # This is the FakePositioner instance
expected_device = mocked_client.device_manager.devices.samx
expected_args_list = [expected_device, args["start"], args["stop"]]
assert called_args == tuple(
expected_args_list
), "The positional arguments passed to the scan function do not match expected values."
assert (
called_kwargs == kwargs
), "The keyword arguments passed to the scan function do not match expected values."
assert called_args == tuple(expected_args_list)
assert called_kwargs == kwargs

View File

@@ -0,0 +1,160 @@
# pylint: disable = no-name-in-module,missing-class-docstring, missing-module-docstring
import pytest
from bec_widgets.utils.widget_io import WidgetIO
from bec_widgets.widgets.scan_control.scan_group_box import ScanGroupBox
def test_kwarg_box(qtbot):
group_input = {
"name": "Kwarg Test",
"inputs": [
# Test float
{
"arg": False,
"name": "exp_time",
"type": "float",
"display_name": "Exp Time",
"tooltip": "Exposure time in seconds",
"default": 0,
"expert": False,
},
# Test int
{
"arg": False,
"name": "num_points",
"type": "int",
"display_name": "Num Points",
"tooltip": "Number of points",
"default": 1,
"expert": False,
},
# Test bool
{
"arg": False,
"name": "relative",
"type": "bool",
"display_name": "Relative",
"tooltip": "If True, the motors will be moved relative to their current position",
"default": False,
"expert": False,
},
# Test str
{
"arg": False,
"name": "scan_type",
"type": "str",
"display_name": "Scan Type",
"tooltip": "Type of scan",
"default": "line",
"expert": False,
},
],
}
kwarg_box = ScanGroupBox(box_type="kwargs", config=group_input)
assert kwarg_box is not None
assert kwarg_box.box_type == "kwargs"
assert kwarg_box.config == group_input
assert kwarg_box.title() == "Kwarg Test"
# Labels
assert kwarg_box.layout.itemAtPosition(0, 0).widget().text() == "Exp Time"
assert kwarg_box.layout.itemAtPosition(0, 1).widget().text() == "Num Points"
assert kwarg_box.layout.itemAtPosition(0, 2).widget().text() == "Relative"
assert kwarg_box.layout.itemAtPosition(0, 3).widget().text() == "Scan Type"
# Widget 0
assert kwarg_box.widgets[0].__class__.__name__ == "ScanDoubleSpinBox"
assert kwarg_box.widgets[0].arg_name == "exp_time"
assert WidgetIO.get_value(kwarg_box.widgets[0]) == 0
assert kwarg_box.widgets[0].toolTip() == "Exposure time in seconds"
# Widget 1
assert kwarg_box.widgets[1].__class__.__name__ == "ScanSpinBox"
assert kwarg_box.widgets[1].arg_name == "num_points"
assert WidgetIO.get_value(kwarg_box.widgets[1]) == 1
assert kwarg_box.widgets[1].toolTip() == "Number of points"
# Widget 2
assert kwarg_box.widgets[2].__class__.__name__ == "ScanCheckBox"
assert kwarg_box.widgets[2].arg_name == "relative"
assert WidgetIO.get_value(kwarg_box.widgets[2]) == False
assert (
kwarg_box.widgets[2].toolTip()
== "If True, the motors will be moved relative to their current position"
)
# Widget 3
assert kwarg_box.widgets[3].__class__.__name__ == "ScanLineEdit"
assert kwarg_box.widgets[3].arg_name == "scan_type"
assert WidgetIO.get_value(kwarg_box.widgets[3]) == "line"
assert kwarg_box.widgets[3].toolTip() == "Type of scan"
parameters = kwarg_box.get_parameters()
assert parameters == {"exp_time": 0, "num_points": 1, "relative": False, "scan_type": "line"}
def test_arg_box(qtbot):
group_input = {
"name": "Arg Test",
"inputs": [
# Test device
{
"arg": True,
"name": "device",
"type": "str",
"display_name": "Device",
"tooltip": "Device to scan",
"default": "samx",
"expert": False,
},
# Test float
{
"arg": True,
"name": "start",
"type": "float",
"display_name": "Start",
"tooltip": "Start position",
"default": 0,
"expert": False,
},
# Test int
{
"arg": True,
"name": "stop",
"type": "int",
"display_name": "Stop",
"tooltip": "Stop position",
"default": 1,
"expert": False,
},
],
}
arg_box = ScanGroupBox(box_type="args", config=group_input)
assert arg_box is not None
assert arg_box.box_type == "args"
assert arg_box.config == group_input
assert arg_box.title() == "Arg Test"
# Labels
assert arg_box.layout.itemAtPosition(0, 0).widget().text() == "Device"
assert arg_box.layout.itemAtPosition(0, 1).widget().text() == "Start"
assert arg_box.layout.itemAtPosition(0, 2).widget().text() == "Stop"
# Widget 0
assert arg_box.widgets[0].__class__.__name__ == "ScanLineEdit"
assert arg_box.widgets[0].arg_name == "device"
assert WidgetIO.get_value(arg_box.widgets[0]) == "samx"
assert arg_box.widgets[0].toolTip() == "Device to scan"
# Widget 1
assert arg_box.widgets[1].__class__.__name__ == "ScanDoubleSpinBox"
assert arg_box.widgets[1].arg_name == "start"
assert WidgetIO.get_value(arg_box.widgets[1]) == 0
assert arg_box.widgets[1].toolTip() == "Start position"
# Widget 2
assert arg_box.widgets[2].__class__.__name__ == "ScanSpinBox"
assert arg_box.widgets[2].arg_name

View File

@@ -0,0 +1,61 @@
import os
import shlex
import subprocess
from unittest import mock
import pytest
from bec_widgets.widgets.vscode.vscode import VSCodeEditor
from .client_mocks import mocked_client
@pytest.fixture
def vscode_widget(qtbot, mocked_client):
with mock.patch("bec_widgets.widgets.vscode.vscode.subprocess.Popen") as mock_popen:
widget = VSCodeEditor(client=mocked_client)
yield widget
def test_vscode_widget(qtbot, vscode_widget):
assert vscode_widget.process is not None
assert vscode_widget._url == "http://127.0.0.1:7000?tkn=bec"
def test_start_server(qtbot, mocked_client):
with mock.patch("bec_widgets.widgets.vscode.vscode.subprocess.Popen") as mock_popen:
mock_process = mock.Mock()
mock_process.stdout.fileno.return_value = 1
mock_process.poll.return_value = None
mock_process.stdout.read.return_value = (
f"available at http://{VSCodeEditor.host}:{VSCodeEditor.port}?tkn={VSCodeEditor.token}"
)
mock_popen.return_value = mock_process
widget = VSCodeEditor(client=mocked_client)
mock_popen.assert_called_once_with(
shlex.split(
f"code serve-web --port {widget.port} --connection-token={widget.token} --accept-server-license-terms"
),
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
preexec_fn=os.setsid,
)
def test_close_event(qtbot, vscode_widget):
with mock.patch("bec_widgets.widgets.vscode.vscode.os.killpg") as mock_killpg:
with mock.patch("bec_widgets.widgets.vscode.vscode.os.getpgid") as mock_getpgid:
with mock.patch(
"bec_widgets.widgets.website.website.WebsiteWidget.closeEvent"
) as mock_close_event:
mock_getpgid.return_value = 123
vscode_widget.process = mock.Mock()
vscode_widget.process.pid = 123
vscode_widget.closeEvent(None)
mock_killpg.assert_called_once_with(123, 15)
vscode_widget.process.wait.assert_called_once()
mock_close_event.assert_called_once()