mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-04-14 04:30:54 +02:00
Compare commits
2 Commits
v0.49.0
...
feature/vs
| Author | SHA1 | Date | |
|---|---|---|---|
| de7eaf7826 | |||
| 1694215c06 |
28
CHANGELOG.md
28
CHANGELOG.md
@@ -2,34 +2,6 @@
|
||||
|
||||
<!--next-version-placeholder-->
|
||||
|
||||
## v0.49.0 (2024-04-24)
|
||||
|
||||
### Feature
|
||||
|
||||
* **rpc/client_utils:** Timeout for rpc response ([`6500a00`](https://gitlab.psi.ch/bec/bec-widgets/-/commit/6500a00682a2a7ca535a138bd9496ed8470856a8))
|
||||
|
||||
### Fix
|
||||
|
||||
* **rpc/client_utils:** Close clean up policy for BECFigure ([`9602085`](https://gitlab.psi.ch/bec/bec-widgets/-/commit/9602085f82cbc983f89b5bfe48bf35f08438fa87))
|
||||
|
||||
## v0.48.0 (2024-04-24)
|
||||
|
||||
### Feature
|
||||
|
||||
* **cli:** Added auto updates plugin support ([`6238693`](https://gitlab.psi.ch/bec/bec-widgets/-/commit/6238693ffb44b47a56b969bc4129f2af7a2c04fe))
|
||||
|
||||
## 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
|
||||
|
||||
@@ -1,2 +1 @@
|
||||
from .auto_updates import AutoUpdates, ScanInfo
|
||||
from .client import BECFigure
|
||||
|
||||
@@ -1,118 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .client import BECFigure
|
||||
|
||||
|
||||
class ScanInfo(BaseModel):
|
||||
scan_id: str
|
||||
scan_number: int
|
||||
scan_name: str
|
||||
scan_report_devices: list
|
||||
monitored_devices: list
|
||||
status: str
|
||||
|
||||
|
||||
class AutoUpdates:
|
||||
def __init__(self, figure: BECFigure, enabled: bool = True):
|
||||
self.enabled = enabled
|
||||
self.figure = figure
|
||||
|
||||
@staticmethod
|
||||
def get_scan_info(msg) -> ScanInfo:
|
||||
"""
|
||||
Update the script with the given data.
|
||||
"""
|
||||
info = msg.info
|
||||
status = msg.status
|
||||
scan_id = msg.scan_id
|
||||
scan_number = info.get("scan_number", 0)
|
||||
scan_name = info.get("scan_name", "Unknown")
|
||||
scan_report_devices = info.get("scan_report_devices", [])
|
||||
monitored_devices = info.get("readout_priority", {}).get("monitored", [])
|
||||
monitored_devices = [dev for dev in monitored_devices if dev not in scan_report_devices]
|
||||
return ScanInfo(
|
||||
scan_id=scan_id,
|
||||
scan_number=scan_number,
|
||||
scan_name=scan_name,
|
||||
scan_report_devices=scan_report_devices,
|
||||
monitored_devices=monitored_devices,
|
||||
status=status,
|
||||
)
|
||||
|
||||
def run(self, msg):
|
||||
"""
|
||||
Run the update function if enabled.
|
||||
"""
|
||||
if not self.enabled:
|
||||
return
|
||||
if msg.status != "open":
|
||||
return
|
||||
info = self.get_scan_info(msg)
|
||||
self.handler(info)
|
||||
|
||||
@staticmethod
|
||||
def get_selected_device(monitored_devices, selected_device):
|
||||
"""
|
||||
Get the selected device for the plot. If no device is selected, the first
|
||||
device in the monitored devices list is selected.
|
||||
"""
|
||||
if selected_device:
|
||||
return selected_device
|
||||
if len(monitored_devices) > 0:
|
||||
sel_device = monitored_devices[0]
|
||||
return sel_device
|
||||
return None
|
||||
|
||||
def handler(self, info: ScanInfo) -> None:
|
||||
"""
|
||||
Default update function.
|
||||
"""
|
||||
if info.scan_name == "line_scan" and info.scan_report_devices:
|
||||
self.simple_line_scan(info)
|
||||
return
|
||||
if info.scan_name == "grid_scan" and info.scan_report_devices:
|
||||
self.simple_grid_scan(info)
|
||||
return
|
||||
if info.scan_report_devices:
|
||||
self.best_effort(info)
|
||||
return
|
||||
|
||||
def simple_line_scan(self, info: ScanInfo) -> None:
|
||||
"""
|
||||
Simple line scan.
|
||||
"""
|
||||
dev_x = info.scan_report_devices[0]
|
||||
dev_y = self.get_selected_device(info.monitored_devices, self.figure.selected_device)
|
||||
if not dev_y:
|
||||
return
|
||||
self.figure.clear_all()
|
||||
plt = self.figure.plot(dev_x, dev_y)
|
||||
plt.set(title=f"Scan {info.scan_number}", x_label=dev_x, y_label=dev_y)
|
||||
|
||||
def simple_grid_scan(self, info: ScanInfo) -> None:
|
||||
"""
|
||||
Simple grid scan.
|
||||
"""
|
||||
dev_x = info.scan_report_devices[0]
|
||||
dev_y = info.scan_report_devices[1]
|
||||
dev_z = self.get_selected_device(info.monitored_devices, self.figure.selected_device)
|
||||
self.figure.clear_all()
|
||||
plt = self.figure.plot(dev_x, dev_y, dev_z, label=f"Scan {info.scan_number}")
|
||||
plt.set(title=f"Scan {info.scan_number}", x_label=dev_x, y_label=dev_y)
|
||||
|
||||
def best_effort(self, info: ScanInfo) -> None:
|
||||
"""
|
||||
Best effort scan.
|
||||
"""
|
||||
dev_x = info.scan_report_devices[0]
|
||||
dev_y = self.get_selected_device(info.monitored_devices, self.figure.selected_device)
|
||||
if not dev_y:
|
||||
return
|
||||
self.figure.clear_all()
|
||||
plt = self.figure.plot(dev_x, dev_y, label=f"Scan {info.scan_number}")
|
||||
plt.set(title=f"Scan {info.scan_number}", x_label=dev_x, y_label=dev_y)
|
||||
Binary file not shown.
|
Before Width: | Height: | Size: 5.6 KiB |
@@ -1,7 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import importlib.metadata as imd
|
||||
import os
|
||||
import select
|
||||
import subprocess
|
||||
@@ -18,7 +17,6 @@ from bec_lib.device import DeviceBase
|
||||
from qtpy.QtCore import QCoreApplication
|
||||
|
||||
import bec_widgets.cli.client as client
|
||||
from bec_widgets.cli.auto_updates import AutoUpdates
|
||||
from bec_widgets.utils.bec_dispatcher import BECDispatcher
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -56,22 +54,68 @@ def rpc_call(func):
|
||||
return wrapper
|
||||
|
||||
|
||||
def get_selected_device(monitored_devices, selected_device):
|
||||
"""
|
||||
Get the selected device for the plot. If no device is selected, the first
|
||||
device in the monitored devices list is selected.
|
||||
"""
|
||||
if selected_device:
|
||||
return selected_device
|
||||
if len(monitored_devices) > 0:
|
||||
sel_device = monitored_devices[0]
|
||||
return sel_device
|
||||
return None
|
||||
|
||||
|
||||
def update_script(figure: BECFigure, msg):
|
||||
"""
|
||||
Update the script with the given data.
|
||||
"""
|
||||
info = msg.info
|
||||
status = msg.status
|
||||
scan_id = msg.scan_id
|
||||
scan_number = info.get("scan_number", 0)
|
||||
scan_name = info.get("scan_name", "Unknown")
|
||||
scan_report_devices = info.get("scan_report_devices", [])
|
||||
monitored_devices = info.get("readout_priority", {}).get("monitored", [])
|
||||
monitored_devices = [dev for dev in monitored_devices if dev not in scan_report_devices]
|
||||
|
||||
if scan_name == "line_scan" and scan_report_devices:
|
||||
dev_x = scan_report_devices[0]
|
||||
dev_y = get_selected_device(monitored_devices, figure.selected_device)
|
||||
print(f"Selected device: {dev_y}")
|
||||
if not dev_y:
|
||||
return
|
||||
figure.clear_all()
|
||||
plt = figure.plot(dev_x, dev_y)
|
||||
plt.set(title=f"Scan {scan_number}", x_label=dev_x, y_label=dev_y)
|
||||
elif scan_name == "grid_scan" and scan_report_devices:
|
||||
print(f"Scan {scan_number} is running")
|
||||
dev_x = scan_report_devices[0]
|
||||
dev_y = scan_report_devices[1]
|
||||
dev_z = get_selected_device(monitored_devices, figure.selected_device)
|
||||
figure.clear_all()
|
||||
plt = figure.plot(dev_x, dev_y, dev_z, label=f"Scan {scan_number}")
|
||||
plt.set(title=f"Scan {scan_number}", x_label=dev_x, y_label=dev_y)
|
||||
elif scan_report_devices:
|
||||
dev_x = scan_report_devices[0]
|
||||
dev_y = get_selected_device(monitored_devices, figure.selected_device)
|
||||
if not dev_y:
|
||||
return
|
||||
figure.clear_all()
|
||||
plt = figure.plot(dev_x, dev_y, label=f"Scan {scan_number}")
|
||||
plt.set(title=f"Scan {scan_number}", x_label=dev_x, y_label=dev_y)
|
||||
|
||||
|
||||
class BECFigureClientMixin:
|
||||
def __init__(self, **kwargs) -> None:
|
||||
super().__init__(**kwargs)
|
||||
self._process = None
|
||||
self.update_script = self._get_update_script()
|
||||
self.update_script = update_script
|
||||
self._target_endpoint = MessageEndpoints.scan_status()
|
||||
self._selected_device = None
|
||||
self.stderr_output = []
|
||||
|
||||
def _get_update_script(self) -> AutoUpdates:
|
||||
eps = imd.entry_points(group="bec.widgets.auto_updates")
|
||||
for ep in eps:
|
||||
if ep.name == "plugin_widgets_update":
|
||||
return ep.load()(figure=self)
|
||||
return None
|
||||
|
||||
@property
|
||||
def selected_device(self):
|
||||
"""
|
||||
@@ -103,7 +147,8 @@ class BECFigureClientMixin:
|
||||
if isinstance(msg, messages.ScanStatusMessage):
|
||||
if not self.gui_is_alive():
|
||||
return
|
||||
self.update_script.run(msg)
|
||||
if msg.status == "open":
|
||||
self.update_script(self, msg)
|
||||
|
||||
def show(self) -> None:
|
||||
"""
|
||||
@@ -121,10 +166,7 @@ class BECFigureClientMixin:
|
||||
"""
|
||||
if self._process is None:
|
||||
return
|
||||
if self.gui_is_alive():
|
||||
self._run_rpc("close", (), wait_for_rpc_response=True)
|
||||
else:
|
||||
self._run_rpc("close", (), wait_for_rpc_response=False)
|
||||
self._run_rpc("close", (), wait_for_rpc_response=False)
|
||||
self._process.terminate()
|
||||
self._process_output_processing_thread.join()
|
||||
self._process = None
|
||||
@@ -170,15 +212,6 @@ class BECFigureClientMixin:
|
||||
self.stderr_output.append(self._process.stderr.read(1024))
|
||||
|
||||
|
||||
class RPCResponseTimeoutError(Exception):
|
||||
"""Exception raised when an RPC response is not received within the expected time."""
|
||||
|
||||
def __init__(self, request_id, timeout):
|
||||
super().__init__(
|
||||
f"RPC response not received within {timeout} seconds for request ID {request_id}"
|
||||
)
|
||||
|
||||
|
||||
class RPCBase:
|
||||
def __init__(self, gui_id: str = None, config: dict = None, parent=None) -> None:
|
||||
self._client = BECDispatcher().client
|
||||
@@ -224,7 +257,7 @@ class RPCBase:
|
||||
parameter={"args": args, "kwargs": kwargs, "gui_id": self._gui_id},
|
||||
metadata={"request_id": request_id},
|
||||
)
|
||||
|
||||
# print(f"RPCBase: {rpc_msg}")
|
||||
# pylint: disable=protected-access
|
||||
receiver = self._root._gui_id
|
||||
self._client.connector.set_and_publish(MessageEndpoints.gui_instructions(receiver), rpc_msg)
|
||||
@@ -259,28 +292,16 @@ class RPCBase:
|
||||
return cls(parent=self, **msg_result)
|
||||
return msg_result
|
||||
|
||||
def _wait_for_response(self, request_id: str, timeout: int = 5):
|
||||
def _wait_for_response(self, request_id):
|
||||
"""
|
||||
Wait for the response from the server.
|
||||
Args:
|
||||
request_id(str): The request ID.
|
||||
timeout(int): The timeout in seconds.
|
||||
|
||||
Returns:
|
||||
The response from the server.
|
||||
"""
|
||||
start_time = time.time()
|
||||
response = None
|
||||
|
||||
while response is None and self.gui_is_alive() and (time.time() - start_time) < timeout:
|
||||
while response is None and self.gui_is_alive():
|
||||
response = self._client.connector.get(
|
||||
MessageEndpoints.gui_instruction_response(request_id)
|
||||
)
|
||||
QCoreApplication.processEvents() # keep UI responsive (and execute signals/slots)
|
||||
time.sleep(0.1)
|
||||
if response is None and (time.time() - start_time) >= timeout:
|
||||
raise RPCResponseTimeoutError(request_id, timeout)
|
||||
|
||||
return response
|
||||
|
||||
def gui_is_alive(self):
|
||||
|
||||
@@ -1,6 +1,4 @@
|
||||
import inspect
|
||||
import threading
|
||||
import time
|
||||
|
||||
from bec_lib import MessageEndpoints, messages
|
||||
from qtpy.QtCore import QTimer
|
||||
@@ -111,25 +109,17 @@ class BECWidgetsCLIServer:
|
||||
def shutdown(self):
|
||||
self._shutdown_event = True
|
||||
self._heartbeat_timer.stop()
|
||||
self.client.shutdown()
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
import argparse
|
||||
import sys
|
||||
|
||||
from qtpy.QtCore import QSize
|
||||
from qtpy.QtGui import QIcon
|
||||
from qtpy.QtWidgets import QApplication, QMainWindow
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
app.setApplicationName("BEC Figure")
|
||||
icon = QIcon()
|
||||
icon.addFile("bec_widgets_icon.png", size=QSize(48, 48))
|
||||
app.setWindowIcon(icon)
|
||||
|
||||
win = QMainWindow()
|
||||
win.setWindowTitle("BEC Widgets")
|
||||
|
||||
parser = argparse.ArgumentParser(description="BEC Widgets CLI Server")
|
||||
parser.add_argument("--id", type=str, help="The id of the server")
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -127,7 +127,3 @@ class BECConnector:
|
||||
return self.config.model_dump()
|
||||
else:
|
||||
return self.config
|
||||
|
||||
def closeEvent(self, event):
|
||||
self.client.shutdown()
|
||||
super().closeEvent(event)
|
||||
|
||||
@@ -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.")
|
||||
@@ -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
|
||||
|
||||
@@ -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)}"
|
||||
)
|
||||
32
bec_widgets/widgets/editor/vscode.py
Normal file
32
bec_widgets/widgets/editor/vscode.py
Normal 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())
|
||||
59
bec_widgets/widgets/editor/vscode_server.py
Normal file
59
bec_widgets/widgets/editor/vscode_server.py
Normal 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()
|
||||
@@ -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.
|
||||
|
||||
@@ -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.
|
||||
|
||||
10
setup.py
10
setup.py
@@ -1,7 +1,7 @@
|
||||
# pylint: disable= missing-module-docstring
|
||||
from setuptools import find_packages, setup
|
||||
|
||||
__version__ = "0.49.0"
|
||||
__version__ = "0.46.6"
|
||||
|
||||
# Default to PyQt6 if no other Qt binding is installed
|
||||
QT_DEPENDENCY = "PyQt6>=6.0"
|
||||
@@ -48,11 +48,5 @@ if __name__ == "__main__":
|
||||
version=__version__,
|
||||
packages=find_packages(),
|
||||
include_package_data=True,
|
||||
package_data={
|
||||
"": [
|
||||
"*.ui",
|
||||
"*.yaml",
|
||||
"*.png",
|
||||
]
|
||||
},
|
||||
package_data={"": ["*.ui", "*.yaml"]},
|
||||
)
|
||||
|
||||
@@ -91,7 +91,6 @@ DEVICES = [
|
||||
FakeDevice("bpm4i"),
|
||||
FakeDevice("bpm3a"),
|
||||
FakeDevice("bpm3i"),
|
||||
FakeDevice("eiger"),
|
||||
]
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user