1
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-04-17 05:55:36 +02:00

Compare commits

..

2 Commits

Author SHA1 Message Date
de7eaf7826 feat: added websitewidget 2024-04-23 09:23:17 +02:00
1694215c06 feat: added simple vscode widget 2024-04-21 10:08:44 +02:00
11 changed files with 129 additions and 156 deletions

View File

@@ -2,18 +2,6 @@
<!--next-version-placeholder-->
## v0.47.0 (2024-04-23)
### Feature
* **utils/thread_checker:** Util class to check the thread leakage for closeEvent in qt ([`71cb80d`](https://gitlab.psi.ch/bec/bec-widgets/-/commit/71cb80d544c5f4ef499379a431ce0c17907c7ce8))
## v0.46.7 (2024-04-21)
### Fix
* **plot/image:** Monitors are now validated with current bec session ([`67a99a1`](https://gitlab.psi.ch/bec/bec-widgets/-/commit/67a99a1a19c261f9a1f09635f274cd9fbfe53639))
## v0.46.6 (2024-04-19)
### Fix

View File

@@ -2,7 +2,6 @@ from .bec_connector import BECConnector, ConnectionConfig
from .bec_dispatcher import BECDispatcher
from .bec_table import BECTable
from .colors import Colors
from .container_utils import WidgetContainerUtils
from .crosshair import Crosshair
from .entry_validator import EntryValidator
from .rpc_decorator import register_rpc_methods, rpc_public

View File

@@ -1,45 +0,0 @@
import itertools
from typing import Type
from qtpy.QtWidgets import QWidget
class WidgetContainerUtils:
@staticmethod
def generate_unique_widget_id(container: dict, prefix: str = "widget") -> str:
"""
Generate a unique widget ID.
Args:
container(dict): The container of widgets.
prefix(str): The prefix of the widget ID.
Returns:
widget_id(str): The unique widget ID.
"""
existing_ids = set(container.keys())
for i in itertools.count(1):
widget_id = f"{prefix}_{i}"
if widget_id not in existing_ids:
return widget_id
@staticmethod
def find_first_widget_by_class(
container: dict, widget_class: Type[QWidget], can_fail: bool = True
) -> QWidget | None:
"""
Find the first widget of a given class in the figure.
Args:
container(dict): The container of widgets.
widget_class(Type): The class of the widget to find.
can_fail(bool): If True, the method will return None if no widget is found. If False, it will raise an error.
Returns:
widget: The widget of the given class.
"""
for widget_id, widget in container.items():
if isinstance(widget, widget_class):
return widget
if can_fail:
return None
else:
raise ValueError(f"No widget of class {widget_class} found.")

View File

@@ -3,15 +3,6 @@ class EntryValidator:
self.devices = devices
def validate_signal(self, name: str, entry: str = None) -> str:
"""
Validate a signal entry for a given device. If the entry is not provided, the first signal entry will be used from the device hints.
Args:
name(str): Device name
entry(str): Signal entry
Returns:
str: Signal entry
"""
if name not in self.devices:
raise ValueError(f"Device '{name}' not found in current BEC session")
@@ -24,17 +15,3 @@ class EntryValidator:
raise ValueError(f"Entry '{entry}' not found in device '{name}' signals")
return entry
def validate_monitor(self, monitor: str) -> str:
"""
Validate a monitor entry for a given device.
Args:
monitor(str): Monitor entry
Returns:
str: Monitor entry
"""
if monitor not in self.devices:
raise ValueError(f"Device '{monitor}' not found in current BEC session")
return monitor

View File

@@ -1,37 +0,0 @@
import threading
class ThreadTracker:
def __init__(self, exclude_names=None):
self.exclude_names = exclude_names if exclude_names else []
self.initial_threads = self._capture_threads()
def _capture_threads(self):
return set(
th
for th in threading.enumerate()
if not any(ex_name in th.name for ex_name in self.exclude_names)
and th is not threading.main_thread()
)
def _thread_info(self, threads):
return ", \n".join(f"{th.name}(ID: {th.ident})" for th in threads)
def check_unfinished_threads(self):
current_threads = self._capture_threads()
additional_threads = current_threads - self.initial_threads
closed_threads = self.initial_threads - current_threads
if additional_threads:
raise Exception(
f"###### Initial threads ######:\n {self._thread_info(self.initial_threads)}\n"
f"###### Current threads ######:\n {self._thread_info(current_threads)}\n"
f"###### Closed threads ######:\n {self._thread_info(closed_threads)}\n"
f"###### Unfinished threads ######:\n {self._thread_info(additional_threads)}"
)
else:
print(
"All threads properly closed.\n"
f"###### Initial threads ######:\n {self._thread_info(self.initial_threads)}\n"
f"###### Current threads ######:\n {self._thread_info(current_threads)}\n"
f"###### Closed threads ######:\n {self._thread_info(closed_threads)}"
)

View File

@@ -0,0 +1,32 @@
from qtpy.QtCore import QUrl
from qtpy.QtWebEngineWidgets import QWebEngineView
from qtpy.QtWidgets import QApplication, QVBoxLayout, QWidget
class WebsiteWidget(QWidget):
def __init__(self, url):
super().__init__()
self.editor = QWebEngineView(self)
layout = QVBoxLayout()
layout.addWidget(self.editor)
self.setLayout(layout)
self.editor.setUrl(QUrl(url))
class VSCodeEditor(WebsiteWidget):
token = "bec"
host = "localhost"
port = 7000
def __init__(self):
super().__init__(f"http://{self.host}:{self.port}?tkn={self.token}")
if __name__ == "__main__":
import sys
app = QApplication(sys.argv)
mainWin = WebsiteWidget("https://scilog.psi.ch")
mainWin.show()
sys.exit(app.exec())

View File

@@ -0,0 +1,59 @@
"""
Module to handle the vscode server
"""
import subprocess
class VSCodeServer:
"""
Class to handle the vscode server
"""
_instance = None
def __init__(self, port=7000, token="bec"):
self.started = False
self._server = None
self.port = port
self.token = token
def __new__(cls, *args, forced=False, **kwargs):
if cls._instance is None or forced:
cls._instance = super(VSCodeServer, cls).__new__(cls)
return cls._instance
def start_server(self):
"""
Start the vscode server in a subprocess
"""
if self.started:
return
self._server = subprocess.Popen(
f"code serve-web --port {self.port} --connection-token={self.token} --accept-server-license-terms",
shell=True,
)
self.started = True
def wait(self):
"""
Wait for the server to finish
"""
if not self.started:
return
if not self._server:
return
self._server.wait()
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="Start the vscode server")
parser.add_argument("--port", type=int, default=7000, help="Port to start the server")
parser.add_argument("--token", type=str, default="bec", help="Token to start the server")
args = parser.parse_args()
server = VSCodeServer(port=args.port, token=args.token)
server.start_server()
server.wait()

View File

@@ -14,7 +14,7 @@ from pyqtgraph.Qt import uic
from qtpy.QtCore import Signal as pyqtSignal
from qtpy.QtWidgets import QApplication, QMainWindow, QVBoxLayout, QWidget
from bec_widgets.utils import BECConnector, BECDispatcher, ConnectionConfig, WidgetContainerUtils
from bec_widgets.utils import BECConnector, BECDispatcher, ConnectionConfig
from bec_widgets.widgets.plots import (
BECImageShow,
BECMotorMap,
@@ -188,7 +188,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
config(dict): Additional configuration for the widget.
**axis_kwargs(dict): Additional axis properties to set on the widget after creation.
"""
widget_id = WidgetContainerUtils.generate_unique_widget_id(self._widgets)
widget_id = self._generate_unique_widget_id()
waveform = self.add_widget(
widget_type="Waveform1D",
widget_id=widget_id,
@@ -278,9 +278,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
Returns:
BECWaveform: The waveform plot widget.
"""
waveform = WidgetContainerUtils.find_first_widget_by_class(
self._widgets, BECWaveform, can_fail=True
)
waveform = self._find_first_widget_by_class(BECWaveform, can_fail=True)
if waveform is not None:
if axis_kwargs:
waveform.set(**axis_kwargs)
@@ -357,9 +355,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
Returns:
BECImageShow: The image widget.
"""
image = WidgetContainerUtils.find_first_widget_by_class(
self._widgets, BECImageShow, can_fail=True
)
image = self._find_first_widget_by_class(BECImageShow, can_fail=True)
if image is not None:
if axis_kwargs:
image.set(**axis_kwargs)
@@ -414,7 +410,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
BECImageShow: The image widget.
"""
widget_id = WidgetContainerUtils.generate_unique_widget_id(self._widgets)
widget_id = self._generate_unique_widget_id()
if config is None:
config = ImageConfig(
widget_class="BECImageShow",
@@ -461,9 +457,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
Returns:
BECMotorMap: The motor map widget.
"""
motor_map = WidgetContainerUtils.find_first_widget_by_class(
self._widgets, BECMotorMap, can_fail=True
)
motor_map = self._find_first_widget_by_class(BECMotorMap, can_fail=True)
if motor_map is not None:
if axis_kwargs:
motor_map.set(**axis_kwargs)
@@ -497,7 +491,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
Returns:
BECMotorMap: The motor map widget.
"""
widget_id = WidgetContainerUtils.generate_unique_widget_id(self._widgets)
widget_id = self._generate_unique_widget_id()
if config is None:
config = MotorMapConfig(
widget_class="BECMotorMap",
@@ -538,7 +532,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
**axis_kwargs(dict): Additional axis properties to set on the widget after creation.
"""
if not widget_id:
widget_id = WidgetContainerUtils.generate_unique_widget_id(self._widgets)
widget_id = self._generate_unique_widget_id()
if widget_id in self._widgets:
raise ValueError(f"Widget with ID '{widget_id}' already exists.")
@@ -616,6 +610,25 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
self.setBackground("k" if theme == "dark" else "w")
self.config.theme = theme
def _find_first_widget_by_class(
self, widget_class: Type[BECPlotBase], can_fail: bool = True
) -> BECPlotBase | None:
"""
Find the first widget of a given class in the figure.
Args:
widget_class(Type[BECPlotBase]): The class of the widget to find.
can_fail(bool): If True, the method will return None if no widget is found. If False, it will raise an error.
Returns:
BECPlotBase: The widget of the given class.
"""
for widget_id, widget in self._widgets.items():
if isinstance(widget, widget_class):
return widget
if can_fail:
return None
else:
raise ValueError(f"No widget of class {widget_class} found.")
def _remove_by_coordinates(self, row: int, col: int) -> None:
"""
Remove a widget from the figure by its coordinates.
@@ -682,6 +695,14 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
row += 1
return row, col
def _generate_unique_widget_id(self):
"""Generate a unique widget ID."""
existing_ids = set(self._widgets.keys())
for i in itertools.count(1):
widget_id = f"widget_{i}"
if widget_id not in existing_ids:
return widget_id
def _change_grid(self, widget_id: str, row: int, col: int):
"""
Change the grid to reflect the new position of the widget.

View File

@@ -12,7 +12,7 @@ from qtpy.QtCore import Signal as pyqtSignal
from qtpy.QtCore import Slot as pyqtSlot
from qtpy.QtWidgets import QWidget
from bec_widgets.utils import BECConnector, ConnectionConfig, EntryValidator
from bec_widgets.utils import BECConnector, ConnectionConfig
from .plot_base import BECPlotBase, WidgetConfig
@@ -335,9 +335,7 @@ class BECImageShow(BECPlotBase):
super().__init__(
parent=parent, parent_figure=parent_figure, config=config, client=client, gui_id=gui_id
)
# Get bec shortcuts dev, scans, queue, scan_storage, dap
self.get_bec_shortcuts()
self.entry_validator = EntryValidator(self.dev)
self._images = defaultdict(dict)
self.apply_config(self.config)
self.processor = ImageProcessor()
@@ -509,8 +507,6 @@ class BECImageShow(BECPlotBase):
f"Monitor with ID '{monitor}' already exists in widget '{self.gui_id}'."
)
monitor = self.entry_validator.validate_monitor(monitor)
image_config = ImageItemConfig(
widget_class="BECImageItem",
parent_id=self.gui_id,
@@ -789,22 +785,6 @@ class BECImageShow(BECPlotBase):
return True
return False
def _validate_monitor(self, monitor: str, validate_bec: bool = True):
"""
Validate the monitor name.
Args:
monitor(str): The name of the monitor.
validate_bec(bool): Whether to validate the monitor name with BEC.
Returns:
bool: True if the monitor name is valid, False otherwise.
"""
if not monitor or monitor == "":
return False
if validate_bec:
return monitor in self.dev
return True
def cleanup(self):
"""
Clean up the widget.

View File

@@ -1,7 +1,7 @@
# pylint: disable= missing-module-docstring
from setuptools import find_packages, setup
__version__ = "0.47.0"
__version__ = "0.46.6"
# Default to PyQt6 if no other Qt binding is installed
QT_DEPENDENCY = "PyQt6>=6.0"

View File

@@ -91,7 +91,6 @@ DEVICES = [
FakeDevice("bpm4i"),
FakeDevice("bpm3a"),
FakeDevice("bpm3i"),
FakeDevice("eiger"),
]