import logging from lib.helpers import timing from PyQt5.QtCore import QObject, QSemaphore, Qt, QTimer, pyqtSignal class Component(QObject): """emitted with data from the _get method""" out = pyqtSignal(list) _pause = pyqtSignal() _resume = pyqtSignal() _set_sources = pyqtSignal(dict) _set_period = pyqtSignal(dict) def __init__( self, config=None, name=None, period=None, lazy=True, paused=False, threaded=True, ): """ parameters: config: value for self.config, should be instance of lib.helpers.config_reader.ConfigReader name: value for self.name, should be used to retrive component configuration section (self.config[self.name]) period: period in seconds for periodic calls to _get, set to None to disable lazy: whether or not skip periodic _get calls if falling behind paused: whether or not periodic calls to_get are paused threaded: set this to tell the component if it should be thread synchronized or not(shoul be true if calls to methods are done from threads different from the component's one) """ super().__init__() self.config = config self.name = name if name is not None else str(id(self)) self._threaded = threaded self._period = period self._single_shot = lazy self._paused = paused self._started = False self._running = False self.sources = None if self._threaded: self._lock = QSemaphore(1) self._lock.acquire(max(self._lock.available(), 1)) self._timer = None self.log = logging.getLogger(f"{self.__class__.__name__} ({self.name})") if not self._threaded: self.start() def _config_changed(self): self.log.info("reconfigure") self.config_changed() self.log.debug(f"config: {self.config}") def config_changed(self): """ this method should be overridden when inheriting from the Component class and should contain all the initialization code that needs to access self.config so that the component will reinitialize if configuration changes this method will be called on start and when self.config (ConfigReader) emits the updated signal """ pass def start(self): """ this method is automatically called if threaded is set to False at object creation otherwise if the component is in a thread this method should be used like this: component = ComponentSubclass(threaded=True) thread = QThread() thread.setTerminationEnabled(True) component.moveToThread(thread) thread.started.connect(component.start) thread.start() component.wait_ready() # this is optional and will wait untill the component has finished started """ self._pause.connect(self._do_pause) self._resume.connect(self._do_resume) self._set_sources.connect(self._do_set_sources) self._set_period.connect(self._do_set_period) if self.config is not None: self.config.updated.connect(self._config_changed) self._config_changed() self._init_periodic() self._started = True if not self._paused: self._do_resume() elif self._threaded: self._lock.release() self.log.info("started") @property def started(self): """returns True if the component has been started""" if self._threaded: self._lock.acquire(max(self._lock.available(), 1)) started = self._started if self._threaded: self._lock.release() return started @property def running(self): """returns True if the periodic calls to _get are not paused""" if self._threaded: self._lock.acquire(max(self._lock.available(), 1)) running = self._running if self._threaded: self._lock.release() return running def wait_ready(self, timeout=5): """ waits untill the requested action has been completed by the component this will return immediately if threaded=False was passed at component initialization """ if self._threaded: timeout = round(timeout * 1000) if self._lock.tryAcquire(max(self._lock.available(), 1), timeout): self._lock.release() else: self._lock.release() raise RuntimeError(f"{self.name} was not ready before timeout of {timeout}ms") def pause(self): """will pause periodic calls to _get and sources trigghers""" if self._threaded: self._lock.acquire(max(self._lock.available(), 1)) if self._running is False: if self._threaded: self._lock.release() return if self._threaded: self._pause.emit() self.wait_ready() else: self._do_pause() def resume(self): """will resume periodic calls to _get and sources trigghers""" if self._threaded: self._lock.acquire(max(self._lock.available(), 1)) if self._running is True: if self._threaded: self._lock.release() return if self._threaded: self._resume.emit() self.wait_ready() else: self._do_resume() def set_sources(self, sources=None): # """ connect the given sources to trigger a call to _get, the sources parameter should be: a dict of signals ({"": , ...}) signals might contain one optional argument that will be passed as data to _get or None to disconnect all sources """ if sources is None: sources = {} if self._threaded: self._lock.acquire(max(self._lock.available(), 1)) self._set_sources.emit(sources) self.wait_ready() else: self._do_set_sources(sources) def add_sources(self, sources=None, overwrite_conflicting_sources=False): # sources should be {"source_name": signal_to_connect, ...} """ add the given sources to the current ones, the sources parameter should be: a dict of signals ({"": , ...}) signals might contain one optional argument that will be passed as data to _get or None if no sources are to be added this method calls set_sources, this is semplest but not the most efficient approach """ if self.sources is None: self_sources = {} else: self_sources = self.sources if sources is None: sources = {} if not overwrite_conflicting_sources: conflicting_sources = { n: [self_sources[n], sources[n]] for n in self_sources.keys() & sources.keys() if self_sources[n] is not sources[n] } if len(conflicting_sources): raise AssertionError("\n\t" + "\n\t".join([f"source named {n!r}: {s[0]!r} will not be replaced with {s[1]!r}" for n, s in conflicting_sources.items()])) self.set_sources({**self_sources, **sources}) def remove_sources(self, sources=None): """ remove the given sources to the current ones by name, the sources parameter should be: an iterable of source names (["", ...]) or None if no sources are to be removed this method calls set_sources, this is semplest but not the most efficient approach """ if sources is None or self.sources is None: return sources = set(sources) self.set_sources({n: s for n, s in self.sources.items() if n not in sources}) def _init_periodic(self): if self._period is not None: if self._timer is None: self._timer = QTimer() self._timer.setTimerType(Qt.PreciseTimer) self._timer.setSingleShot(self._single_shot) self._timer.setInterval(round(self._period * 1000)) self.log.debug(f"init periodic: period: {self._period}, single shot: {self._single_shot}") else: self.log.debug("no init periodic") def set_period(self, period=None, lazy=True): """will set the period for periodic calls to _get and whether or not those are lazy (see init parameters)""" if self._threaded: self._lock.acquire(max(self._lock.available(), 1)) self._set_period.emit({"period": period, "lazy": lazy}) self.wait_ready() else: self._do_set_period({"period": period, "lazy": lazy}) def _start_periodic(self): if self._timer is not None: self._timer.timeout.connect(self._get) self._timer.start() self.log.debug("started periodic") else: self.log.debug("no started periodic") def _stop_periodic(self): if self._timer is not None: self._timer.stop() try: self._timer.timeout.disconnect() except TypeError: pass self.log.debug("stopped periodic") else: self.log.debug("no stopped periodic") def _connect_sources(self): if self.sources is not None: for source in self.sources.values(): source.connect(self._get) self.log.debug(f"connected sources: {list(self.sources)}") else: self.log.debug("no connected sources") def _disconnect_sources(self): if self.sources is not None: for source in self.sources.values(): try: source.disconnect() except TypeError: pass self.log.debug(f"disconnected sources: {list(self.sources)}") else: self.log.debug("no disconnected sources") def _do_resume(self): self._start_periodic() self._connect_sources() self._running = True self.log.info("resumed") if self._threaded: self._lock.release() def _do_pause(self): self._stop_periodic() self._disconnect_sources() self._running = False self.log.info("paused") if self._threaded: self._lock.release() def _do_set_sources(self, sources): if self._running: self._disconnect_sources() if sources is not None and not len(sources): sources = None self.sources = sources if self._running: self._connect_sources() self.log.info("set sources") if self._threaded: self._lock.release() def _do_set_period(self, spec): self._period = spec.get("period", None) self._single_shot = spec.get("lazy", True) self._init_periodic() self.log.info("set period") if self._threaded: self._lock.release() def _get(self, data=None, emit=True): """ this method should be overridden when inheriting from the Component class the overriding method should retrive all the data and then call super()._get(data) this will emit the data in the proper format if the emit parameter is not set to False """ if emit: if data is None: data = [None] t = timing() got = [{ "time": t if type(d) is not dict or "time" not in d else d["time"], self.name: d, } for d in data] self.out.emit(got) self.log.debug(f"_get: {got}") else: self.log.debug("_get") if self._timer is not None and self._single_shot: self._timer.start() def set(self, val): """ this method should be overridden when inheriting from the Component class the overriding method should set the requested val and then call super()._set(set_value) this will log the value that has been set """ self.log.debug(f"set: {val}")