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

feat: '._auto_updates_enabled' attribute can be used to activate auto updates installation in BECDockArea

Browse Source
This commit is contained in:
guijar_m
2024-11-25 18:31:21 +01:00
parent cffcdf2923
commit 31d87036c9
1 changed files with 30 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
bec_widgets/cli/client_utils.py
View File

@ -135,14 +135,27 @@ class RepeatTimer(threading.Timer):
class BECGuiClientMixin: class BECGuiClientMixin:
def __init__(self, **kwargs) -> None: def __init__(self, **kwargs) -> None:
super().__init__(**kwargs) super().__init__(**kwargs)
self._auto_updates_enabled = True
self._auto_updates = None
self._gui_started_timer = None self._gui_started_timer = None
self._gui_started_event = threading.Event() self._gui_started_event = threading.Event()
self._process = None self._process = None
self._process_output_processing_thread = None self._process_output_processing_thread = None
self.auto_updates = self._get_update_script()
self._target_endpoint = MessageEndpoints.scan_status() self._target_endpoint = MessageEndpoints.scan_status()
self._selected_device = None self._selected_device = None
@property
def auto_updates(self):
if self._auto_updates_enabled:
self._gui_started_event.wait()
return self._auto_updates
def shutdown_auto_updates(self):
if self._auto_updates_enabled:
if self._auto_updates is not None:
self._auto_updates.shutdown()
self._auto_updates = None
def _get_update_script(self) -> AutoUpdates | None: def _get_update_script(self) -> AutoUpdates | None:
eps = imd.entry_points(group="bec.widgets.auto_updates") eps = imd.entry_points(group="bec.widgets.auto_updates")
for ep in eps: for ep in eps:
@ -188,18 +201,23 @@ class BECGuiClientMixin:
if isinstance(msg, messages.ScanStatusMessage): if isinstance(msg, messages.ScanStatusMessage):
if not self.gui_is_alive(): if not self.gui_is_alive():
return return
self.auto_updates.msg_queue.put(msg) if self._auto_updates_enabled:
self.auto_updates.msg_queue.put(msg)
def _gui_post_startup(self): def _gui_post_startup(self):
if self.auto_updates is None: if self._auto_updates_enabled:
AutoUpdates.create_default_dock = True if self._auto_updates is None:
AutoUpdates.enabled = True auto_updates = self._get_update_script()
self.auto_updates = AutoUpdates(gui=self) if auto_updates is None:
if self.auto_updates.create_default_dock: AutoUpdates.create_default_dock = True
self.auto_updates.start_default_dock() AutoUpdates.enabled = True
fig = self.auto_updates.get_default_figure() auto_updates = AutoUpdates(gui=self)
self._gui_started_event.set() if auto_updates.create_default_dock:
self.show_all() auto_updates.start_default_dock()
# fig = auto_updates.get_default_figure()
self._auto_updates = auto_updates
self._gui_started_event.set()
self.show_all()
def start_server(self, wait=False) -> None: def start_server(self, wait=False) -> None:
""" """
@ -256,9 +274,7 @@ class BECGuiClientMixin:
self._process_output_processing_thread.join() self._process_output_processing_thread.join()
self._process.wait() self._process.wait()
self._process = None self._process = None
if self.auto_updates is not None: self.shutdown_auto_updates()
self.auto_updates.shutdown()
self.auto_updates = None
class RPCResponseTimeoutError(Exception): class RPCResponseTimeoutError(Exception):