bec/bec_widgets
0
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-07-14 03:31:50 +02:00
Code Activity

wip

Browse Source
This commit is contained in:
wakonig_k
2025-03-28 11:11:59 +01:00
committed by wyzula-jan
parent 705e819352
commit 7719ac86b8
7 changed files with 104 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
bec_widgets/cli/server.py
View File

@ -14,7 +14,7 @@ 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.bec_qapp import BECApplication
from bec_widgets.utils.bec_dispatcher import BECDispatcher
logger = bec_logger.logger
@ -39,7 +39,7 @@ class SimpleFileLikeFromLogOutputFunc:
class GUIServer:
"""
Class that starts the GUI server.
This class is used to start the BEC GUI and is the main entry point for launching BEC Widgets in a subprocess.
"""
def __init__(self, args):
@ -48,8 +48,9 @@ class GUIServer:
self.gui_class = args.gui_class
self.gui_class_id = args.gui_class_id
self.hide = args.hide
self.app: BECApplication | None = None
self.app: QApplication | None = None
self.launcher_window: LaunchWindow | None = None
self.dispatcher: BECDispatcher | None = None
def start(self):
"""
@ -61,9 +62,9 @@ class GUIServer:
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()
# 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:
@ -82,30 +83,37 @@ class GUIServer:
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)
if self.app:
self.app.setQuitOnLastWindowClosed(True)
else:
self.launcher_window.hide()
self.app.setQuitOnLastWindowClosed(False)
if self.app:
self.app.setQuitOnLastWindowClosed(False)
def _run(self):
"""
Run the GUI server.
"""
self.app = QApplication(sys.argv)
self.app.setApplicationName("BEC")
service_config = self._get_service_config()
self.app = BECApplication(sys.argv, config=service_config, gui_id=self.gui_id)
self.app.setQuitOnLastWindowClosed(False)
self.dispatcher = BECDispatcher(config=service_config)
self.dispatcher.start_cli_server(gui_id=self.gui_id)
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)
self.app.aboutToQuit.connect(self.shutdown)
self.app.setQuitOnLastWindowClosed(False)
# 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.
@ -119,7 +127,8 @@ class GUIServer:
with RPCRegister.delayed_broadcast():
for widget in QApplication.instance().topLevelWidgets(): # type: ignore
widget.close()
self.app.quit()
if self.app:
self.app.quit()
# gui.bec.close()
# win.shutdown()
@ -128,6 +137,14 @@ class GUIServer:
sys.exit(self.app.exec())
def shutdown(self):
"""
Shutdown the GUI server.
"""
if self.dispatcher:
self.dispatcher.stop_cli_server()
self.dispatcher.disconnect_all()
def main():
"""
@ -157,7 +174,7 @@ def main():
if __name__ == "__main__":
import sys
# import sys
sys.argv = ["bec_widgets", "--gui_class", "MainWindow"]
# sys.argv = ["bec_widgets", "--gui_class", "MainWindow"]
main()

53
bec_widgets/utils/bec_dispatcher.py
View File

@ -1,6 +1,8 @@
from __future__ import annotations
import collections
import random
import string
from collections.abc import Callable
from typing import TYPE_CHECKING, Union
@ -17,6 +19,8 @@ logger = bec_logger.logger
if TYPE_CHECKING:
from bec_lib.endpoints import EndpointInfo
from bec_widgets.utils.cli_server import CLIServer
class QtThreadSafeCallback(QObject):
cb_signal = pyqtSignal(dict, dict)
@ -74,6 +78,7 @@ class BECDispatcher:
_instance = None
_initialized = False
client: BECClient
cli_server: CLIServer | None = None
def __new__(cls, client=None, config: str | ServiceConfig | None = None, *args, **kwargs):
if cls._instance is None:
@ -113,6 +118,9 @@ class BECDispatcher:
@classmethod
def reset_singleton(cls):
"""
Reset the singleton instance of the BECDispatcher.
"""
cls._instance = None
cls._initialized = False
@ -179,4 +187,49 @@ class BECDispatcher:
*args: Arbitrary positional arguments
**kwargs: Arbitrary keyword arguments
"""
# pylint: disable=protected-access
self.disconnect_topics(self.client.connector._topics_cb)
def start_cli_server(self, gui_id: str | None = None):
"""
Start the CLI server.
Args:
gui_id(str, optional): The GUI ID. Defaults to None. If None, a unique identifier will be generated.
"""
# pylint: disable=import-outside-toplevel
from bec_widgets.utils.cli_server import CLIServer
if gui_id is None:
gui_id = self.generate_unique_identifier()
if not self.client.started:
logger.error("Cannot start CLI server without a running client")
return
self.cli_server = CLIServer(gui_id, dispatcher=self, client=self.client)
logger.success("Started CLI server with gui_id: {gui_id}")
def stop_cli_server(self):
"""
Stop the CLI server.
"""
if self.cli_server is None:
logger.error("Cannot stop CLI server without starting it first")
return
self.cli_server.shutdown()
self.cli_server = None
logger.success("Stopped CLI server")
@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))

8
bec_widgets/utils/bec_qapp.py
View File

@ -72,14 +72,6 @@ class BECApplication:
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.app.aboutToQuit.connect(self.shutdown)

2
bec_widgets/utils/bec_widget.py
View File

@ -64,7 +64,7 @@ class BECWidget(BECConnector):
parent_dock=parent_dock,
parent_id=parent_id,
)
app = self._ensure_bec_app()
app = QApplication.instance()
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

2
bec_widgets/utils/cli_server.py
View File

@ -182,5 +182,5 @@ class CLIServer:
self.status = messages.BECStatus.IDLE
self._heartbeat_timer.stop()
self.emit_heartbeat()
logger.info("Succeded in shutting down gui")
logger.info("Succeded in shutting down CLI server")
self.client.shutdown()

18
bec_widgets/utils/ui_loader.py
View File

@ -1,11 +1,14 @@
import os
import inspect
from bec_lib.logger import bec_logger
from qtpy import PYQT6, PYSIDE6, QT_VERSION
from qtpy.QtCore import QFile, QIODevice
from bec_widgets.utils.generate_designer_plugin import DesignerPluginInfo
from bec_widgets.utils.plugin_utils import get_custom_classes
logger = bec_logger.logger
if PYSIDE6:
from PySide6.QtUiTools import QUiLoader
@ -18,9 +21,16 @@ if PYSIDE6:
def createWidget(self, class_name, parent=None, name=""):
if class_name in self.custom_widgets:
widget = self.custom_widgets[class_name](
self.baseinstance, parent_id=self.baseinstance.gui_id
)
# check if the custom widget has a parent_id argument
if "parent_id" in inspect.signature(self.custom_widgets[class_name]).parameters:
gui_id = getattr(self.baseinstance, "gui_id", None)
widget = self.custom_widgets[class_name](self.baseinstance, parent_id=gui_id)
else:
logger.warning(
f"Custom widget {class_name} does not have a parent_id argument. "
)
widget = self.custom_widgets[class_name](self.baseinstance)
widget.setObjectName(name)
return widget
return super().createWidget(class_name, self.baseinstance, name)

6
tests/unit_tests/conftest.py
View File

@ -29,11 +29,11 @@ def qapplication(qtbot, request, testable_qtimer_class): # pylint: disable=unus
print("Test failed, skipping cleanup checks")
return
qapp = BECApplication()
qapp.shutdown()
# qapp = BECApplication()
# qapp.shutdown()
testable_qtimer_class.check_all_stopped(qtbot)
qapp = QApplication.instance()
qapp.processEvents()
if hasattr(qapp, "os_listener") and qapp.os_listener:
qapp.removeEventFilter(qapp.os_listener)