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: add possibility to provide service config

Browse Source
This commit is contained in:
usov_i
2023-08-07 10:00:31 +02:00
parent ffa5029535
commit 8c9a9c9353
1 changed files with 17 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
bec_widgets/bec_dispatcher.py
View File

@ -1,8 +1,10 @@
import argparse
import os
from dataclasses import dataclass from dataclasses import dataclass
from threading import RLock from threading import RLock
from bec_lib import BECClient from bec_lib import BECClient
from bec_lib.core import BECMessage, MessageEndpoints from bec_lib.core import BECMessage, MessageEndpoints, ServiceConfig
from bec_lib.core.redis_connector import RedisConsumerThreaded from bec_lib.core.redis_connector import RedisConsumerThreaded
from PyQt5.QtCore import QObject, pyqtSignal from PyQt5.QtCore import QObject, pyqtSignal
@ -23,9 +25,16 @@ class _BECDispatcher(QObject):
new_projection_id = pyqtSignal(dict) new_projection_id = pyqtSignal(dict)
new_projection_data = pyqtSignal(dict) new_projection_data = pyqtSignal(dict)
def __init__(self): def __init__(self, bec_config=None):
super().__init__() super().__init__()
self.client = BECClient() self.client = BECClient()
# TODO: this is a workaround for now to provide service config within qtdesigner, but is
# it possible to provide config via a cli arg?
if bec_config is None and os.path.isfile("bec_config.yaml"):
bec_config = "bec_config.yaml"
self.client.initialize(config=ServiceConfig(config_path=bec_config))
self.client.start() self.client.start()
self._slot_signal_map = { self._slot_signal_map = {
@ -150,4 +159,9 @@ class _BECDispatcher(QObject):
del self._daps[data_ep] del self._daps[data_ep]
bec_dispatcher = _BECDispatcher() parser = argparse.ArgumentParser()
parser.add_argument("--bec-config", default=None)
parser.add_argument("args", nargs=argparse.REMAINDER)
args = parser.parse_args()
bec_dispatcher = _BECDispatcher(args.bec_config)