mirror of
https://github.com/bec-project/bec_widgets.git
synced 2025-07-14 03:31:50 +02:00
refactor: BECConnector changed config structure
This commit is contained in:
@ -2,4 +2,5 @@ from .crosshair import Crosshair
|
||||
from .colors import Colors
|
||||
from .validator_delegate import DoubleValidationDelegate
|
||||
from .bec_table import BECTable
|
||||
from .bec_connector import BECConnector
|
||||
from .bec_connector import BECConnector, ConnectionConfig
|
||||
from .bec_dispatcher import BECDispatcher
|
||||
|
@ -1,57 +1,71 @@
|
||||
# pylint: disable = no-name-in-module,missing-module-docstring
|
||||
|
||||
import time
|
||||
from typing import Type, Optional
|
||||
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
from qtpy.QtCore import Slot as pyqtSlot
|
||||
|
||||
from bec_widgets.utils.bec_dispatcher import BECDispatcher
|
||||
|
||||
|
||||
class ConnectionConfig(BaseModel):
|
||||
"""Configuration for BECConnector mixin class"""
|
||||
|
||||
widget_class: str = Field(default="NonSpecifiedWidget", description="The class of the widget.")
|
||||
gui_id: Optional[str] = Field(
|
||||
default=None, validate_default=True, description="The GUI ID of the widget."
|
||||
)
|
||||
|
||||
@field_validator("gui_id")
|
||||
def generate_gui_id(cls, v, values):
|
||||
"""Generate a GUI ID if none is provided."""
|
||||
if v is None:
|
||||
widget_class = values.data["widget_class"]
|
||||
v = f"{widget_class}_{str(time.time())}"
|
||||
return v
|
||||
return v
|
||||
|
||||
|
||||
class BECConnector:
|
||||
"""Connection mixin class for all BEC widgets, to handle BEC client and device manager"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
client=None,
|
||||
gui_id=None,
|
||||
config: dict = None,
|
||||
):
|
||||
def __init__(self, client=None, config: ConnectionConfig = None, gui_id: str = None):
|
||||
# BEC related connections
|
||||
self.bec_dispatcher = BECDispatcher()
|
||||
self.client = self.bec_dispatcher.client if client is None else client
|
||||
self.dev = self.client.device_manager.devices
|
||||
|
||||
self.gui_id = gui_id
|
||||
|
||||
if self.gui_id is None:
|
||||
self.gui_id = self.__class__.__name__ + str(time.time())
|
||||
|
||||
# Current configuration
|
||||
self.config = config
|
||||
|
||||
# Init UI
|
||||
if self.config is None:
|
||||
print(f"No initial config found for {self.__class__.__name__}")
|
||||
if config:
|
||||
self.config = config
|
||||
else:
|
||||
self.on_config_update(self.config)
|
||||
print(
|
||||
f"No initial config found for {self.__class__.__name__}.\n"
|
||||
f"Initializing with default config."
|
||||
)
|
||||
self.config = ConnectionConfig(widget_class=self.__class__.__name__)
|
||||
|
||||
# TODO decide if worth to do on the level of base class
|
||||
if gui_id:
|
||||
self.config.gui_id = gui_id
|
||||
self.gui_id = gui_id
|
||||
else:
|
||||
self.gui_id = self.config.gui_id
|
||||
|
||||
@pyqtSlot(str)
|
||||
def set_gui_id(self, gui_id: str) -> None:
|
||||
"""
|
||||
Set the GUI ID for the widget.
|
||||
Args:
|
||||
gui_id(str): GUI ID
|
||||
"""
|
||||
self.gui_id = gui_id
|
||||
# TODO decide if to change here or put it also in dataclass like something like gui_id: str = None
|
||||
self.config["gui_id"] = gui_id
|
||||
self.config.gui_id = gui_id
|
||||
|
||||
def _init_config(self):
|
||||
"""Initialise the configuration"""
|
||||
raise NotImplementedError
|
||||
|
||||
def get_config(self):
|
||||
"""Return the current configuration settings."""
|
||||
return self.config
|
||||
def get_bec_shortcuts(self):
|
||||
"""Get BEC shortcuts for the widget."""
|
||||
self.dev = self.client.device_manager.devices
|
||||
self.scans = self.client.scans
|
||||
self.queue = self.client.queue
|
||||
self.scan_storage = self.queue.scan_storage
|
||||
self.dap = self.client.dap
|
||||
|
||||
def update_client(self, client) -> None:
|
||||
"""Update the client and device manager from BEC and create object for BEC shortcuts.
|
||||
@ -59,28 +73,24 @@ class BECConnector:
|
||||
client: BEC client
|
||||
"""
|
||||
self.client = client
|
||||
self.dev = self.client.device_manager.devices
|
||||
self.scans = self.client.scans
|
||||
self.queue = self.client.queue
|
||||
self.scan_storage = self.queue.scan_storage
|
||||
self.dap = self.client.dap
|
||||
self.get_bec_shortcuts()
|
||||
|
||||
# TODO will be pydantic model instead of python dict
|
||||
@pyqtSlot(dict)
|
||||
def on_config_update(self, config: dict) -> None: # TODO rpc this?
|
||||
@pyqtSlot(ConnectionConfig) # TODO can be also dict
|
||||
def on_config_update(self, config: ConnectionConfig | dict) -> None:
|
||||
"""
|
||||
Update the configuration for the widget.
|
||||
Args:
|
||||
config(dict): Configuration settings.
|
||||
config(ConnectionConfig): Configuration settings.
|
||||
"""
|
||||
self.config = config
|
||||
self._init_config()
|
||||
if isinstance(config, dict):
|
||||
config = ConnectionConfig(**config)
|
||||
# TODO add error handler
|
||||
|
||||
@pyqtSlot(dict)
|
||||
def on_instruction(self, msg_content: dict) -> None: # TODO decide this or rpc?
|
||||
"""
|
||||
Handle instructions sent to the GUI.
|
||||
Args:
|
||||
msg_content (dict): Message content with the instruction and parameters.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
self.config = config
|
||||
|
||||
def get_config(self):
|
||||
return self.config
|
||||
|
||||
|
||||
# connector = BECConnector()
|
||||
# print(connector.config)
|
||||
|
Reference in New Issue
Block a user