diff --git a/src/dynalab_core/derive.py b/src/dynalab_core/derive.py index 874a0c5..3391618 100644 --- a/src/dynalab_core/derive.py +++ b/src/dynalab_core/derive.py @@ -35,12 +35,11 @@ class DeriveUnit: return_signal: SignalDescriptor, ) -> None: self._id: UUID = unit_uuid - self._worker_thread: Thread | None = None self._process_function: Callable = processing_function self._return_signal: SignalDescriptor = return_signal self._stop_event: threading.Event = threading.Event() - self._input_queue: Queue[list[ValueDescriptor]] = [] - self._output_queue: Queue[ValueDescriptor] = [] + self._input_queue: Queue[list[ValueDescriptor]] = Queue() + self._output_queue: Queue[ValueDescriptor] = Queue() self._num_input_args: int = 0 self._parser_state: int = ParserState.INIT @@ -66,10 +65,12 @@ class DeriveUnit: if sig.return_annotation is not ValueDescriptor: raise DeriveUnitInvalidSignatureError - print(f"Signature validated with {self._num_input_args} input values") - # TODO: Define and start worker thread + self._worker_thread: Thread = Thread( + target=self._worker_function, name=f"{self._id}_worker_thread", daemon=True + ) + self._worker_thread.start() - def worker_function(self) -> None: + def _worker_function(self) -> None: while not self._stop_event.is_set(): try: input_args = self._input_queue.get(timeout=0.1) diff --git a/test/manual/derive_unit.py b/test/manual/derive_unit.py index 325a9f9..0698d54 100644 --- a/test/manual/derive_unit.py +++ b/test/manual/derive_unit.py @@ -38,4 +38,13 @@ unit = DeriveUnit(uuid4(), process, [sig_a, sig_b], sig_r) val_a = ValueDescriptor(signal_id=sig_a.id, value=1, timestamp=0) val_b = ValueDescriptor(signal_id=sig_a.id, value=2, timestamp=10) -print(unit.process_offline([val_a, val_b])) +unit.put_data([val_a, val_b]) + +try: + while True: + output = unit.get_output() + if output is not None: + print(output) + +except KeyboardInterrupt: + exit(0)