mirror of
https://github.com/bec-project/bec_widgets.git
synced 2025-07-14 03:31:50 +02:00
feat: moved to bec qapp
This commit is contained in:
@ -1,191 +1,24 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import functools
|
||||
import argparse
|
||||
import json
|
||||
import signal
|
||||
import sys
|
||||
import types
|
||||
from contextlib import contextmanager, redirect_stderr, redirect_stdout
|
||||
from typing import Union
|
||||
from contextlib import redirect_stderr, redirect_stdout
|
||||
from typing import cast
|
||||
|
||||
from bec_lib.endpoints import MessageEndpoints
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_lib.service_config import ServiceConfig
|
||||
from bec_lib.utils.import_utils import lazy_import
|
||||
from qtpy.QtCore import Qt, QTimer
|
||||
from redis.exceptions import RedisError
|
||||
from qtpy.QtCore import Qt
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
from bec_widgets.applications.launch_window import LaunchWindow
|
||||
from bec_widgets.cli.rpc.rpc_register import RPCRegister
|
||||
from bec_widgets.utils import BECDispatcher
|
||||
from bec_widgets.utils.bec_connector import BECConnector
|
||||
from bec_widgets.utils.bec_qapp import BECApplication
|
||||
from bec_widgets.utils.error_popups import ErrorPopupUtility
|
||||
from bec_widgets.widgets.containers.dock import BECDockArea
|
||||
from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow, WindowWithUi
|
||||
|
||||
messages = lazy_import("bec_lib.messages")
|
||||
logger = bec_logger.logger
|
||||
|
||||
|
||||
@contextmanager
|
||||
def rpc_exception_hook(err_func):
|
||||
"""This context replaces the popup message box for error display with a specific hook"""
|
||||
# get error popup utility singleton
|
||||
popup = ErrorPopupUtility()
|
||||
# save current setting
|
||||
old_exception_hook = popup.custom_exception_hook
|
||||
|
||||
# install err_func, if it is a callable
|
||||
# IMPORTANT, Keep self here, because this method is overwriting the custom_exception_hook
|
||||
# of the ErrorPopupUtility (popup instance) class.
|
||||
def custom_exception_hook(self, exc_type, value, tb, **kwargs):
|
||||
err_func({"error": popup.get_error_message(exc_type, value, tb)})
|
||||
|
||||
popup.custom_exception_hook = types.MethodType(custom_exception_hook, popup)
|
||||
|
||||
try:
|
||||
yield popup
|
||||
finally:
|
||||
# restore state of error popup utility singleton
|
||||
popup.custom_exception_hook = old_exception_hook
|
||||
|
||||
|
||||
class BECWidgetsCLIServer:
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
gui_id: str,
|
||||
dispatcher: BECDispatcher = None,
|
||||
client=None,
|
||||
config=None,
|
||||
gui_class: type[BECDockArea, WindowWithUi] = BECDockArea,
|
||||
gui_class_id: str = "bec",
|
||||
) -> None:
|
||||
self.status = messages.BECStatus.BUSY
|
||||
self.dispatcher = BECDispatcher(config=config) if dispatcher is None else dispatcher
|
||||
self.client = self.dispatcher.client if client is None else client
|
||||
self.client.start()
|
||||
self.gui_id = gui_id
|
||||
# register broadcast callback
|
||||
self.rpc_register = RPCRegister()
|
||||
self.rpc_register.add_callback(self.broadcast_registry_update)
|
||||
|
||||
self.dispatcher.connect_slot(
|
||||
self.on_rpc_update, MessageEndpoints.gui_instructions(self.gui_id)
|
||||
)
|
||||
|
||||
# Setup QTimer for heartbeat
|
||||
self._heartbeat_timer = QTimer()
|
||||
self._heartbeat_timer.timeout.connect(self.emit_heartbeat)
|
||||
self._heartbeat_timer.start(200)
|
||||
|
||||
self.status = messages.BECStatus.RUNNING
|
||||
with RPCRegister.delayed_broadcast():
|
||||
self.gui = gui_class(parent=None, name=gui_class_id, gui_id=gui_class_id)
|
||||
logger.success(f"Server started with gui_id: {self.gui_id}")
|
||||
# Create initial object -> BECFigure or BECDockArea
|
||||
|
||||
def on_rpc_update(self, msg: dict, metadata: dict):
|
||||
request_id = metadata.get("request_id")
|
||||
logger.debug(f"Received RPC instruction: {msg}, metadata: {metadata}")
|
||||
with rpc_exception_hook(functools.partial(self.send_response, request_id, False)):
|
||||
try:
|
||||
obj = self.get_object_from_config(msg["parameter"])
|
||||
method = msg["action"]
|
||||
args = msg["parameter"].get("args", [])
|
||||
kwargs = msg["parameter"].get("kwargs", {})
|
||||
res = self.run_rpc(obj, method, args, kwargs)
|
||||
except Exception as e:
|
||||
logger.error(f"Error while executing RPC instruction: {e}")
|
||||
self.send_response(request_id, False, {"error": str(e)})
|
||||
else:
|
||||
logger.debug(f"RPC instruction executed successfully: {res}")
|
||||
self.send_response(request_id, True, {"result": res})
|
||||
|
||||
def send_response(self, request_id: str, accepted: bool, msg: dict):
|
||||
self.client.connector.set_and_publish(
|
||||
MessageEndpoints.gui_instruction_response(request_id),
|
||||
messages.RequestResponseMessage(accepted=accepted, message=msg),
|
||||
expire=60,
|
||||
)
|
||||
|
||||
def get_object_from_config(self, config: dict):
|
||||
gui_id = config.get("gui_id")
|
||||
obj = self.rpc_register.get_rpc_by_id(gui_id)
|
||||
if obj is None:
|
||||
raise ValueError(f"Object with gui_id {gui_id} not found")
|
||||
return obj
|
||||
|
||||
def run_rpc(self, obj, method, args, kwargs):
|
||||
# Run with rpc registry broadcast, but only once
|
||||
with RPCRegister.delayed_broadcast():
|
||||
logger.debug(f"Running RPC instruction: {method} with args: {args}, kwargs: {kwargs}")
|
||||
method_obj = getattr(obj, method)
|
||||
# check if the method accepts args and kwargs
|
||||
if not callable(method_obj):
|
||||
if not args:
|
||||
res = method_obj
|
||||
else:
|
||||
setattr(obj, method, args[0])
|
||||
res = None
|
||||
else:
|
||||
res = method_obj(*args, **kwargs)
|
||||
|
||||
if isinstance(res, list):
|
||||
res = [self.serialize_object(obj) for obj in res]
|
||||
elif isinstance(res, dict):
|
||||
res = {key: self.serialize_object(val) for key, val in res.items()}
|
||||
else:
|
||||
res = self.serialize_object(res)
|
||||
return res
|
||||
|
||||
def serialize_object(self, obj):
|
||||
if isinstance(obj, BECConnector):
|
||||
config = obj.config.model_dump()
|
||||
config["parent_id"] = obj.parent_id # add parent_id to config
|
||||
return {
|
||||
"gui_id": obj.gui_id,
|
||||
"name": (
|
||||
obj._name if hasattr(obj, "_name") else obj.__class__.__name__
|
||||
), # pylint: disable=protected-access
|
||||
"widget_class": obj.__class__.__name__,
|
||||
"config": config,
|
||||
"__rpc__": True,
|
||||
}
|
||||
return obj
|
||||
|
||||
def emit_heartbeat(self):
|
||||
logger.trace(f"Emitting heartbeat for {self.gui_id}")
|
||||
try:
|
||||
self.client.connector.set(
|
||||
MessageEndpoints.gui_heartbeat(self.gui_id),
|
||||
messages.StatusMessage(name=self.gui_id, status=self.status, info={}),
|
||||
expire=10,
|
||||
)
|
||||
except RedisError as exc:
|
||||
logger.error(f"Error while emitting heartbeat: {exc}")
|
||||
|
||||
def broadcast_registry_update(self, connections: dict):
|
||||
"""
|
||||
Broadcast the updated registry to all clients.
|
||||
"""
|
||||
|
||||
# We only need to broadcast the dock areas
|
||||
data = {key: self.serialize_object(val) for key, val in connections.items()}
|
||||
self.client.connector.xadd(
|
||||
MessageEndpoints.gui_registry_state(self.gui_id),
|
||||
msg_dict={"data": messages.GUIRegistryStateMessage(state=data)},
|
||||
max_size=1, # only single message in stream
|
||||
)
|
||||
|
||||
def shutdown(self): # TODO not sure if needed when cleanup is done at level of BECConnector
|
||||
self.status = messages.BECStatus.IDLE
|
||||
self._heartbeat_timer.stop()
|
||||
self.emit_heartbeat()
|
||||
logger.info("Succeded in shutting down gui")
|
||||
self.client.shutdown()
|
||||
|
||||
|
||||
class SimpleFileLikeFromLogOutputFunc:
|
||||
def __init__(self, log_func):
|
||||
self._log_func = log_func
|
||||
@ -204,37 +37,102 @@ class SimpleFileLikeFromLogOutputFunc:
|
||||
return
|
||||
|
||||
|
||||
def _start_server(
|
||||
gui_id: str,
|
||||
gui_class: BECDockArea | WindowWithUi,
|
||||
gui_class_id: str = "bec",
|
||||
config: str | None = None,
|
||||
):
|
||||
if config:
|
||||
try:
|
||||
config = json.loads(config)
|
||||
service_config = ServiceConfig(config=config)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
service_config = ServiceConfig(config_path=config)
|
||||
else:
|
||||
# if no config is provided, use the default config
|
||||
service_config = ServiceConfig()
|
||||
class GUIServer:
|
||||
"""
|
||||
Class that starts the GUI server.
|
||||
"""
|
||||
|
||||
# bec_logger.configure(
|
||||
# service_config.redis,
|
||||
# QtRedisConnector,
|
||||
# service_name="BECWidgetsCLIServer",
|
||||
# service_config=service_config.service_config,
|
||||
# )
|
||||
server = BECWidgetsCLIServer(
|
||||
gui_id=gui_id, config=service_config, gui_class=gui_class, gui_class_id=gui_class_id
|
||||
)
|
||||
return server
|
||||
def __init__(self, args):
|
||||
self.config = args.config
|
||||
self.gui_id = args.id
|
||||
self.gui_class = args.gui_class
|
||||
self.gui_class_id = args.gui_class_id
|
||||
self.hide = args.hide
|
||||
self.app: BECApplication | None = None
|
||||
self.launcher_window: LaunchWindow | None = None
|
||||
|
||||
def start(self):
|
||||
"""
|
||||
Start the GUI server.
|
||||
"""
|
||||
bec_logger.level = bec_logger.LOGLEVEL.INFO
|
||||
if self.hide:
|
||||
# pylint: disable=protected-access
|
||||
bec_logger._stderr_log_level = bec_logger.LOGLEVEL.ERROR
|
||||
bec_logger._update_sinks()
|
||||
|
||||
with redirect_stdout(SimpleFileLikeFromLogOutputFunc(logger.info)): # type: ignore
|
||||
with redirect_stderr(SimpleFileLikeFromLogOutputFunc(logger.error)): # type: ignore
|
||||
self._run()
|
||||
|
||||
def _get_service_config(self) -> ServiceConfig:
|
||||
if self.config:
|
||||
try:
|
||||
config = json.loads(self.config)
|
||||
service_config = ServiceConfig(config=config)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
service_config = ServiceConfig(config_path=config)
|
||||
else:
|
||||
# if no config is provided, use the default config
|
||||
service_config = ServiceConfig()
|
||||
return service_config
|
||||
|
||||
def _turn_off_the_lights(self, connections: dict):
|
||||
"""
|
||||
If there is only one connection remaining, it is the launcher, so we show it.
|
||||
Once the launcher is closed as the last window, we quit the application.
|
||||
"""
|
||||
self.app = cast(BECApplication, self.app)
|
||||
self.launcher_window = cast(LaunchWindow, self.launcher_window)
|
||||
|
||||
if len(connections) <= 1:
|
||||
self.launcher_window.show()
|
||||
self.launcher_window.activateWindow()
|
||||
self.launcher_window.raise_()
|
||||
self.app.setQuitOnLastWindowClosed(True)
|
||||
else:
|
||||
self.launcher_window.hide()
|
||||
self.app.setQuitOnLastWindowClosed(False)
|
||||
|
||||
def _run(self):
|
||||
"""
|
||||
Run the GUI server.
|
||||
"""
|
||||
service_config = self._get_service_config()
|
||||
self.app = BECApplication(sys.argv, config=service_config, gui_id=self.gui_id)
|
||||
self.app.setQuitOnLastWindowClosed(False)
|
||||
|
||||
self.launcher_window = LaunchWindow(gui_id=f"{self.gui_id}:launcher")
|
||||
self.launcher_window.setAttribute(Qt.WA_ShowWithoutActivating) # type: ignore
|
||||
|
||||
RPCRegister().callbacks.append(self._turn_off_the_lights)
|
||||
|
||||
if self.gui_class:
|
||||
# If the server is started with a specific gui class, we launch it.
|
||||
# This will automatically hide the launcher.
|
||||
self.launcher_window.launch(self.gui_class, name=self.gui_class_id)
|
||||
|
||||
def sigint_handler(*args):
|
||||
# display message, for people to let it terminate gracefully
|
||||
print("Caught SIGINT, exiting")
|
||||
# Widgets should be all closed.
|
||||
with RPCRegister.delayed_broadcast():
|
||||
for widget in QApplication.instance().topLevelWidgets(): # type: ignore
|
||||
widget.close()
|
||||
self.app.quit()
|
||||
|
||||
# gui.bec.close()
|
||||
# win.shutdown()
|
||||
signal.signal(signal.SIGINT, sigint_handler)
|
||||
signal.signal(signal.SIGTERM, sigint_handler)
|
||||
|
||||
sys.exit(self.app.exec())
|
||||
|
||||
|
||||
def main():
|
||||
import argparse
|
||||
from qtpy.QtWidgets import QApplication
|
||||
"""
|
||||
Main entry point for subprocesses that start a GUI server.
|
||||
"""
|
||||
|
||||
parser = argparse.ArgumentParser(description="BEC Widgets CLI Server")
|
||||
parser.add_argument("--id", type=str, default="test", help="The id of the server")
|
||||
@ -254,65 +152,12 @@ def main():
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
bec_logger.level = bec_logger.LOGLEVEL.INFO
|
||||
if args.hide:
|
||||
# pylint: disable=protected-access
|
||||
bec_logger._stderr_log_level = bec_logger.LOGLEVEL.ERROR
|
||||
bec_logger._update_sinks()
|
||||
|
||||
if args.gui_class == "BECDockArea":
|
||||
gui_class = BECDockArea
|
||||
elif args.gui_class == "MainWindow":
|
||||
gui_class = WindowWithUi
|
||||
else:
|
||||
print(
|
||||
"Please specify a valid gui_class to run. Use -h for help."
|
||||
"\n Starting with default gui_class BECFigure."
|
||||
)
|
||||
gui_class = BECDockArea
|
||||
|
||||
with redirect_stdout(SimpleFileLikeFromLogOutputFunc(logger.info)):
|
||||
with redirect_stderr(SimpleFileLikeFromLogOutputFunc(logger.error)):
|
||||
app = BECApplication(sys.argv)
|
||||
# set close on last window, only if not under control of client ;
|
||||
# indeed, Qt considers a hidden window a closed window, so if all windows
|
||||
# are hidden by default it exits
|
||||
app.setQuitOnLastWindowClosed(not args.hide)
|
||||
# store gui id within QApplication object, to make it available to all widgets #TODO not needed probably
|
||||
app.gui_id = args.id
|
||||
|
||||
# args.id = "abff6"
|
||||
# TODO to test other gui just change gui_class to something else such as WindowWithUi
|
||||
server = _start_server(args.id, WindowWithUi, args.gui_class_id, args.config)
|
||||
|
||||
win = BECMainWindow(gui_id=f"{server.gui_id}:window")
|
||||
win.setAttribute(Qt.WA_ShowWithoutActivating)
|
||||
# win.setWindowTitle("BEC")
|
||||
|
||||
RPCRegister().add_rpc(win)
|
||||
gui = server.gui
|
||||
win.setCentralWidget(gui)
|
||||
if not args.hide:
|
||||
win.show()
|
||||
|
||||
app.aboutToQuit.connect(server.shutdown)
|
||||
|
||||
def sigint_handler(*args):
|
||||
# display message, for people to let it terminate gracefully
|
||||
print("Caught SIGINT, exiting")
|
||||
# Widgets should be all closed.
|
||||
with RPCRegister.delayed_broadcast():
|
||||
for widget in QApplication.instance().topLevelWidgets():
|
||||
widget.close()
|
||||
app.quit()
|
||||
|
||||
# gui.bec.close()
|
||||
# win.shutdown()
|
||||
signal.signal(signal.SIGINT, sigint_handler)
|
||||
signal.signal(signal.SIGTERM, sigint_handler)
|
||||
|
||||
sys.exit(app.exec())
|
||||
server = GUIServer(args)
|
||||
server.start()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
|
||||
sys.argv = ["bec_widgets", "--gui_class", "MainWindow"]
|
||||
main()
|
||||
|
||||
@ -73,14 +73,15 @@ class BECDispatcher:
|
||||
|
||||
_instance = None
|
||||
_initialized = False
|
||||
client: BECClient
|
||||
|
||||
def __new__(cls, client=None, config: str = None, *args, **kwargs):
|
||||
def __new__(cls, client=None, config: str | ServiceConfig | None = None, *args, **kwargs):
|
||||
if cls._instance is None:
|
||||
cls._instance = super(BECDispatcher, cls).__new__(cls)
|
||||
cls._initialized = False
|
||||
return cls._instance
|
||||
|
||||
def __init__(self, client=None, config: str | ServiceConfig = None):
|
||||
def __init__(self, client=None, config: str | ServiceConfig | None = None):
|
||||
if self._initialized:
|
||||
return
|
||||
|
||||
|
||||
@ -3,61 +3,120 @@ from __future__ import annotations
|
||||
import os
|
||||
import random
|
||||
import string
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from bec_lib import bec_logger
|
||||
from bec_widgets.utils.bec_dispatcher import BECDispatcher
|
||||
from bec_widgets.utils.bec_widget import BECWidget
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_lib.service_config import ServiceConfig
|
||||
from qtpy.QtCore import QSize
|
||||
from qtpy.QtGui import QIcon
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
import collections
|
||||
from collections.abc import Callable
|
||||
from typing import TYPE_CHECKING, Union
|
||||
|
||||
import redis
|
||||
from bec_lib.client import BECClient
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_lib.redis_connector import MessageObject, RedisConnector
|
||||
from bec_lib.service_config import ServiceConfig
|
||||
from qtpy.QtCore import QObject
|
||||
from qtpy.QtCore import Signal
|
||||
import bec_widgets
|
||||
from bec_widgets.cli.rpc.rpc_register import RPCRegister
|
||||
from bec_widgets.utils.bec_dispatcher import BECDispatcher
|
||||
from bec_widgets.utils.cli_server import CLIServer
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
MODULE_PATH = os.path.dirname(bec_widgets.__file__)
|
||||
|
||||
|
||||
class BECApplication(QApplication):
|
||||
if TYPE_CHECKING: # pragma: no cover
|
||||
from bec_lib.client import BECClient
|
||||
|
||||
|
||||
class BECApplication:
|
||||
"""
|
||||
Custom QApplication class for BEC applications.
|
||||
"""
|
||||
|
||||
gui_id: str
|
||||
dispatcher: BECDispatcher
|
||||
rpc_register: RPCRegister
|
||||
client: BECClient
|
||||
is_bec_app: bool
|
||||
cli_server: CLIServer
|
||||
|
||||
_instance: BECApplication
|
||||
_initialized: bool
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*args,
|
||||
client=None,
|
||||
config: str | ServiceConfig = None,
|
||||
config: str | ServiceConfig | None = None,
|
||||
gui_id: str | None = None,
|
||||
**kwargs,
|
||||
):
|
||||
super().__init__(*args, **kwargs)
|
||||
if self._initialized:
|
||||
return
|
||||
self.app = QApplication.instance()
|
||||
if self.app is None:
|
||||
self.app = QApplication([])
|
||||
self._initialize_bec_app(client, config, gui_id)
|
||||
self._initialized = True
|
||||
|
||||
self.gui_id = gui_id or self.generate_unique_identifier()
|
||||
self.dispatcher = BECDispatcher(client=client, config=config)
|
||||
self.rpc_register = RPCRegister()
|
||||
self.client = self.dispatcher.client # fetch client from Dispatcher
|
||||
def _initialize_bec_app(
|
||||
self, client=None, config: str | ServiceConfig | None = None, gui_id: str | None = None
|
||||
):
|
||||
"""
|
||||
Initialize the BECApplication instance with the given client and configuration.
|
||||
|
||||
# Indicate this is a BEC application
|
||||
self.is_bec_app = True
|
||||
Args:
|
||||
app: The QApplication instance to initialize.
|
||||
client: The BECClient instance to use for communication.
|
||||
config: The ServiceConfig instance to use for configuration.
|
||||
gui_id: The unique identifier for this application.
|
||||
"""
|
||||
|
||||
self.app.gui_id = gui_id or BECApplication.generate_unique_identifier()
|
||||
self.app.dispatcher = BECDispatcher(client=client, config=config)
|
||||
self.app.rpc_register = RPCRegister()
|
||||
self.app.client = self.app.dispatcher.client # type: ignore
|
||||
|
||||
self.app.cli_server = CLIServer(
|
||||
gui_id=self.app.gui_id,
|
||||
dispatcher=self.app.dispatcher,
|
||||
client=self.app.client,
|
||||
config=config,
|
||||
)
|
||||
|
||||
self.app.is_bec_app = True
|
||||
|
||||
self.setup_bec_icon()
|
||||
self.register_all()
|
||||
|
||||
def __instancecheck__(self, instance: Any) -> bool:
|
||||
return isinstance(instance, (QApplication, BECApplication))
|
||||
|
||||
def __getattr__(self, name: str) -> Any:
|
||||
if hasattr(self.app, name):
|
||||
return getattr(self.app, name)
|
||||
return super().__getattr__(name)
|
||||
|
||||
def __new__(cls, *args, **kwargs) -> BECApplication:
|
||||
if not hasattr(cls, "_instance"):
|
||||
cls._instance = super().__new__(cls)
|
||||
cls._initialized = False
|
||||
return cls._instance
|
||||
|
||||
@classmethod
|
||||
def from_qapplication(
|
||||
cls, client=None, config: str | ServiceConfig | None = None, gui_id: str | None = None
|
||||
) -> BECApplication:
|
||||
"""
|
||||
Create a BECApplication instance from an existing QApplication instance.
|
||||
"""
|
||||
print("from_qapplication")
|
||||
app = QApplication.instance()
|
||||
if isinstance(app, BECApplication):
|
||||
return app
|
||||
|
||||
return cls(client=client, config=config, gui_id=gui_id)
|
||||
|
||||
def setup_bec_icon(self):
|
||||
"""
|
||||
Set the BEC icon for the application
|
||||
"""
|
||||
icon = QIcon()
|
||||
icon.addFile(
|
||||
os.path.join(MODULE_PATH, "assets", "app_icons", "bec_widgets_icon.png"),
|
||||
@ -67,90 +126,100 @@ class BECApplication(QApplication):
|
||||
|
||||
@staticmethod
|
||||
def generate_unique_identifier(length: int = 4) -> str:
|
||||
"""
|
||||
Generate a unique identifier for the application.
|
||||
|
||||
Args:
|
||||
length: The length of the identifier. Defaults to 4.
|
||||
|
||||
Returns:
|
||||
str: The unique identifier.
|
||||
"""
|
||||
allowed_chars = string.ascii_lowercase + string.digits
|
||||
return "".join(random.choices(allowed_chars, k=length))
|
||||
|
||||
# TODO not sure if needed
|
||||
def register_all(self):
|
||||
widgets = self.allWidgets()
|
||||
all_connections = self.rpc_register.list_all_connections()
|
||||
for widget in widgets:
|
||||
if not isinstance(widget, BECWidget):
|
||||
continue
|
||||
gui_id = getattr(widget, "gui_id", None)
|
||||
if gui_id and widget not in all_connections:
|
||||
self.rpc_register.add_rpc(widget)
|
||||
print(
|
||||
f"[BECQApplication]: Registered widget {widget.__class__} with GUI ID: {gui_id}"
|
||||
)
|
||||
# # TODO not sure if needed
|
||||
# def register_all(self):
|
||||
# widgets = self.allWidgets()
|
||||
# all_connections = self.rpc_register.list_all_connections()
|
||||
# for widget in widgets:
|
||||
# if not isinstance(widget, BECWidget):
|
||||
# continue
|
||||
# gui_id = getattr(widget, "gui_id", None)
|
||||
# if gui_id and widget not in all_connections:
|
||||
# self.rpc_register.add_rpc(widget)
|
||||
# print(
|
||||
# f"[BECQApplication]: Registered widget {widget.__class__} with GUI ID: {gui_id}"
|
||||
# )
|
||||
|
||||
# TODO not sure if needed
|
||||
def list_all_bec_widgets(self):
|
||||
widgets = self.allWidgets()
|
||||
bec_widgets = []
|
||||
for widget in widgets:
|
||||
if isinstance(widget, BECWidget):
|
||||
bec_widgets.append(widget)
|
||||
return bec_widgets
|
||||
# # TODO not sure if needed
|
||||
# def list_all_bec_widgets(self):
|
||||
# widgets = self.allWidgets()
|
||||
# bec_widgets = []
|
||||
# for widget in widgets:
|
||||
# if isinstance(widget, BECWidget):
|
||||
# bec_widgets.append(widget)
|
||||
# return bec_widgets
|
||||
|
||||
def list_hierarchy(self, only_bec_widgets: bool = True, show_parent: bool = True):
|
||||
"""
|
||||
List the hierarchy of all BECWidgets in this application.
|
||||
# def list_hierarchy(self, only_bec_widgets: bool = True, show_parent: bool = True):
|
||||
# """
|
||||
# List the hierarchy of all BECWidgets in this application.
|
||||
|
||||
Args:
|
||||
only_bec_widgets (bool): If True, prints only BECWidgets. Non-BECWidgets are skipped but their children are still traversed.
|
||||
show_parent (bool): If True, displays the immediate BECWidget ancestor for each item.
|
||||
"""
|
||||
bec_widgets = self.list_all_bec_widgets()
|
||||
# Identify top-level BECWidgets (whose parent is not another BECWidget)
|
||||
top_level = [
|
||||
w for w in bec_widgets if not isinstance(self._get_becwidget_ancestor(w), BECWidget)
|
||||
]
|
||||
# Args:
|
||||
# only_bec_widgets (bool): If True, prints only BECWidgets. Non-BECWidgets are skipped but their children are still traversed.
|
||||
# show_parent (bool): If True, displays the immediate BECWidget ancestor for each item.
|
||||
# """
|
||||
# bec_widgets = self.list_all_bec_widgets()
|
||||
# # Identify top-level BECWidgets (whose parent is not another BECWidget)
|
||||
# top_level = [
|
||||
# w for w in bec_widgets if not isinstance(self._get_becwidget_ancestor(w), BECWidget)
|
||||
# ]
|
||||
|
||||
print("[BECQApplication]: Listing BECWidget hierarchy:")
|
||||
for widget in top_level:
|
||||
self._print_becwidget_hierarchy(
|
||||
widget, indent=0, only_bec_widgets=only_bec_widgets, show_parent=show_parent
|
||||
)
|
||||
# print("[BECQApplication]: Listing BECWidget hierarchy:")
|
||||
# for widget in top_level:
|
||||
# self._print_becwidget_hierarchy(
|
||||
# widget, indent=0, only_bec_widgets=only_bec_widgets, show_parent=show_parent
|
||||
# )
|
||||
|
||||
def _print_becwidget_hierarchy(self, widget, indent=0, only_bec_widgets=True, show_parent=True):
|
||||
# Decide if this widget should be printed
|
||||
is_bec = isinstance(widget, BECWidget)
|
||||
print_this = (not only_bec_widgets) or is_bec
|
||||
# def _print_becwidget_hierarchy(self, widget, indent=0, only_bec_widgets=True, show_parent=True):
|
||||
# # Decide if this widget should be printed
|
||||
# is_bec = isinstance(widget, BECWidget)
|
||||
# print_this = (not only_bec_widgets) or is_bec
|
||||
|
||||
parent_info = ""
|
||||
if show_parent and is_bec:
|
||||
ancestor = self._get_becwidget_ancestor(widget)
|
||||
if ancestor is not None:
|
||||
parent_info = f" parent={ancestor.__class__.__name__}"
|
||||
else:
|
||||
parent_info = " parent=None"
|
||||
# parent_info = ""
|
||||
# if show_parent and is_bec:
|
||||
# ancestor = self._get_becwidget_ancestor(widget)
|
||||
# if ancestor is not None:
|
||||
# parent_info = f" parent={ancestor.__class__.__name__}"
|
||||
# else:
|
||||
# parent_info = " parent=None"
|
||||
|
||||
if print_this:
|
||||
prefix = " " * indent
|
||||
print(
|
||||
f"{prefix}- {widget.__class__.__name__} (objectName={widget.objectName()}){parent_info}"
|
||||
)
|
||||
# if print_this:
|
||||
# prefix = " " * indent
|
||||
# print(
|
||||
# f"{prefix}- {widget.__class__.__name__} (objectName={widget.objectName()}){parent_info}"
|
||||
# )
|
||||
|
||||
# Always recurse so deeper BECWidgets aren't missed
|
||||
for child in widget.children():
|
||||
# Skip known non-BECWidgets if only_bec_widgets is True, but keep recursion
|
||||
# We'll still call _print_becwidget_hierarchy to discover any BECWidget descendants.
|
||||
self._print_becwidget_hierarchy(
|
||||
child, indent + 2, only_bec_widgets=only_bec_widgets, show_parent=show_parent
|
||||
)
|
||||
# # Always recurse so deeper BECWidgets aren't missed
|
||||
# for child in widget.children():
|
||||
# # Skip known non-BECWidgets if only_bec_widgets is True, but keep recursion
|
||||
# # We'll still call _print_becwidget_hierarchy to discover any BECWidget descendants.
|
||||
# self._print_becwidget_hierarchy(
|
||||
# child, indent + 2, only_bec_widgets=only_bec_widgets, show_parent=show_parent
|
||||
# )
|
||||
|
||||
def _get_becwidget_ancestor(self, widget):
|
||||
"""
|
||||
Climb the .parent() chain until finding another BECWidget, or None.
|
||||
"""
|
||||
p = widget.parent()
|
||||
while p is not None:
|
||||
if isinstance(p, BECWidget):
|
||||
return p
|
||||
p = p.parent()
|
||||
return None
|
||||
# def _get_becwidget_ancestor(self, widget):
|
||||
# """
|
||||
# Climb the .parent() chain until finding another BECWidget, or None.
|
||||
# """
|
||||
# p = widget.parent()
|
||||
# while p is not None:
|
||||
# if isinstance(p, BECWidget):
|
||||
# return p
|
||||
# p = p.parent()
|
||||
# return None
|
||||
|
||||
def shutdown(self):
|
||||
self.dispatcher.disconnect_all()
|
||||
self.cli_server.shutdown()
|
||||
super().shutdown()
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import darkdetect
|
||||
from bec_lib.logger import bec_logger
|
||||
@ -56,7 +56,6 @@ class BECWidget(BECConnector):
|
||||
"""
|
||||
if not isinstance(self, QWidget):
|
||||
raise RuntimeError(f"{repr(self)} is not a subclass of QWidget")
|
||||
|
||||
super().__init__(
|
||||
client=client,
|
||||
config=config,
|
||||
@ -65,7 +64,7 @@ class BECWidget(BECConnector):
|
||||
parent_dock=parent_dock,
|
||||
parent_id=parent_id,
|
||||
)
|
||||
app = QApplication.instance()
|
||||
app = self._ensure_bec_app()
|
||||
if not hasattr(app, "theme"):
|
||||
# DO NOT SET THE THEME TO AUTO! Otherwise, the qwebengineview will segfault
|
||||
# Instead, we will set the theme to the system setting on startup
|
||||
@ -78,6 +77,13 @@ class BECWidget(BECConnector):
|
||||
logger.debug(f"Subscribing to theme updates for {self.__class__.__name__}")
|
||||
self._connect_to_theme_change()
|
||||
|
||||
def _ensure_bec_app(self):
|
||||
# pylint: disable=import-outside-toplevel
|
||||
from bec_widgets.utils.bec_qapp import BECApplication
|
||||
|
||||
app = BECApplication.from_qapplication()
|
||||
return app
|
||||
|
||||
def _connect_to_theme_change(self):
|
||||
"""Connect to the theme change signal."""
|
||||
qapp = QApplication.instance()
|
||||
|
||||
@ -1,11 +1,9 @@
|
||||
import os
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from qtpy.QtGui import QActionGroup, QAction
|
||||
from qtpy.QtWidgets import QStyle
|
||||
|
||||
from bec_lib.logger import bec_logger
|
||||
from qtpy.QtWidgets import QApplication, QMainWindow
|
||||
from qtpy.QtGui import QAction, QActionGroup
|
||||
from qtpy.QtWidgets import QApplication, QMainWindow, QStyle
|
||||
|
||||
from bec_widgets.cli.rpc.rpc_register import RPCRegister
|
||||
from bec_widgets.utils import UILoader
|
||||
@ -51,7 +49,7 @@ class BECMainWindow(BECWidget, QMainWindow):
|
||||
self.setCentralWidget(self.ui)
|
||||
|
||||
def _init_bec_specific_ui(self):
|
||||
if isinstance(self.app, BECApplication):
|
||||
if getattr(self.app, "is_bec_app", False):
|
||||
self.statusBar().showMessage(f"App ID: {self.app.gui_id}")
|
||||
else:
|
||||
logger.warning(
|
||||
@ -114,7 +112,7 @@ class BECMainWindow(BECWidget, QMainWindow):
|
||||
help_menu.addAction(widgets_docs)
|
||||
help_menu.addAction(bug_report)
|
||||
|
||||
debug_bar = menu_bar.addMenu("DEBUG")
|
||||
debug_bar = menu_bar.addMenu(f"DEBUG {self.__class__.__name__}")
|
||||
list_hierarchy = QAction("List App Hierarchy", self)
|
||||
list_hierarchy.triggered.connect(self.list_app_hierarchy)
|
||||
debug_bar.addAction(list_hierarchy)
|
||||
@ -169,7 +167,7 @@ class BECMainWindow(BECWidget, QMainWindow):
|
||||
else:
|
||||
name = "dock_area"
|
||||
name = WidgetContainerUtils.generate_unique_name(name, existing_dock_areas)
|
||||
dock_area = BECDockArea(name=name)
|
||||
dock_area = WindowWithUi() # BECDockArea(name=name)
|
||||
dock_area.resize(dock_area.minimumSizeHint())
|
||||
# TODO Should we simply use the specified name as title here?
|
||||
dock_area.window().setWindowTitle(f"BEC - {name}")
|
||||
@ -243,7 +241,10 @@ class WindowWithUi(BECMainWindow):
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
|
||||
app = BECApplication(sys.argv)
|
||||
app = QApplication(sys.argv)
|
||||
print(id(app))
|
||||
# app = BECApplication(sys.argv)
|
||||
# print(id(app))
|
||||
main_window = WindowWithUi()
|
||||
main_window.show()
|
||||
sys.exit(app.exec())
|
||||
|
||||
Reference in New Issue
Block a user