From 268964bbda6d003bf9906db84e9604cd08091035 Mon Sep 17 00:00:00 2001 From: Antonio Bevilacqua Date: Tue, 9 Jun 2026 17:51:19 +0200 Subject: [PATCH 1/7] feat: nodes now receive state externally from their pipeline BREAKING CHANGE: old update signature does no longer work --- juturna/components/__init__.py | 2 + juturna/components/_node.py | 13 ++++- juturna/components/_pipeline.py | 5 ++ juturna/components/_state.py | 36 ++++++++++++ .../passthrough_identity.py | 3 +- tests/test_node_threads.py | 4 +- .../nodes/proc/_aggregator/aggregator.py | 5 +- .../nodes/proc/_amplifier/amplifier.py | 3 +- .../nodes/sink/_crasher/crasher.py | 5 +- .../test_plugins/nodes/sink/_dumper/dumper.py | 3 +- .../source/_data_streamer/data_streamer.py | 3 +- .../nodes/source/_sequencer/sequencer.py | 2 +- tests/test_state.py | 57 +++++++++++++++++++ 13 files changed, 126 insertions(+), 15 deletions(-) create mode 100644 juturna/components/_state.py create mode 100644 tests/test_state.py diff --git a/juturna/components/__init__.py b/juturna/components/__init__.py index 839882c3..3974e75b 100644 --- a/juturna/components/__init__.py +++ b/juturna/components/__init__.py @@ -3,6 +3,7 @@ from juturna.components._node import Node from juturna.components._pipeline import Pipeline from juturna.components._buffer import Buffer +from juturna.components._state import State __all__ = [ @@ -10,4 +11,5 @@ 'Node', 'Pipeline', 'Buffer', + 'State', ] diff --git a/juturna/components/_node.py b/juturna/components/_node.py index ec16e4a3..27f12699 100644 --- a/juturna/components/_node.py +++ b/juturna/components/_node.py @@ -21,6 +21,7 @@ from juturna.meta import JUTURNA_TELEMETRY_BATCH_SIZE from juturna.components._buffer import Buffer +from juturna.components._state import State from juturna.components._telemetry_manager import TelemetryManager from juturna.components._synchronisers import _SYNCHRONISERS @@ -65,6 +66,7 @@ def __init__( ) self._status: ComponentStatus | None = None + self._state: State | None = None self._queue = queue.Queue(maxsize=JUTURNA_MAX_QUEUE_SIZE) self._worker_thread: threading.Thread | None = None @@ -130,6 +132,9 @@ def destinations(self) -> list: def link_telemetry(self, manager: TelemetryManager): self._telemetry_manager = manager + def link_state(self, state: State): + self._state = state + def put(self, message: Message | ControlSignal): if self._draining.is_set(): self.logger.debug('message received while draining, discarding...') @@ -372,7 +377,7 @@ def join(self): def configure(self): ... - def update(self, message: Message[T_Input]): ... + def update(self, message: Message[T_Input], state: State): ... def set_on_config(self, prop: str, value: Any): ... @@ -415,18 +420,22 @@ def _update(self): batch.payload, ControlPayload ): self._handle_control(batch) + if batch.payload.signal < 0: break + continue self._last_data_source_evt_id = batch.id + with self._pending_condition: self._pending_updates += 1 try: - self.update(batch) + self.update(batch, self._state) finally: with self._pending_condition: self._pending_updates -= 1 + if self._pending_updates == 0: self._pending_condition.notify_all() diff --git a/juturna/components/_pipeline.py b/juturna/components/_pipeline.py index 4306fc47..bb44ffdd 100644 --- a/juturna/components/_pipeline.py +++ b/juturna/components/_pipeline.py @@ -16,6 +16,7 @@ from juturna.payloads import ControlSignal, ControlPayload from juturna.components._dag import DAG +from juturna.components._state import State from juturna.components._node_builder import _builder from juturna.components._telemetry_manager import TelemetryManager @@ -46,6 +47,7 @@ def __init__(self, config: dict): self._nodes: dict[str, Node] = dict() self._links: list = list() self._dag: DAG = DAG() + self._node_state_store = dict() self._telemetry_manager: TelemetryManager | None = None self._telemetry = False @@ -168,6 +170,9 @@ def warmup(self): self._nodes[node_name] = _node self._dag.add_node(node_name) + self._node_state_store[node_name] = State() + + _node.link_state(self._node_state_store[node_name]) for link in links: from_node = link['from'] diff --git a/juturna/components/_state.py b/juturna/components/_state.py new file mode 100644 index 00000000..8cd389fd --- /dev/null +++ b/juturna/components/_state.py @@ -0,0 +1,36 @@ +from collections import UserDict + + +class State(UserDict): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + + self._ops = dict() + + def __setitem__(self, key, value): + super().__setitem__(key, value) + + self._ops[key] = ('SET', value) + + def __delitem__(self, key): + _deleted = super().__delitem__(key) + + self._ops[key] = ('DEL', None) + + return _deleted + + def deltas(self): + _ops = [ + (action, key, value) for key, (action, value) in self._ops.items() + ] + + self._ops.clear() + + return _ops + + def apply(self, deltas): + for _op, key, value in deltas: + if _op == 'SET': + super().__setitem__(key, value) + elif _op == 'DEL': + super().__delitem__(key) diff --git a/plugins/nodes/proc/_passthrough_identity/passthrough_identity.py b/plugins/nodes/proc/_passthrough_identity/passthrough_identity.py index 86ffb71c..b1db7aaf 100644 --- a/plugins/nodes/proc/_passthrough_identity/passthrough_identity.py +++ b/plugins/nodes/proc/_passthrough_identity/passthrough_identity.py @@ -11,6 +11,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import BasePayload @@ -33,7 +34,7 @@ def __init__(self, delay: int, **kwargs): self._delay = delay self._transmitted = 0 - def update(self, message: Message[BasePayload]): + def update(self, message: Message[BasePayload], state: State): """Receive a message from downstream, transmit a message upstream""" self.logger.info( f'message {message.version} received from: {message.creator}' diff --git a/tests/test_node_threads.py b/tests/test_node_threads.py index 8d49266c..71368212 100644 --- a/tests/test_node_threads.py +++ b/tests/test_node_threads.py @@ -1,10 +1,10 @@ import time import threading -from juturna.components import Message, Node +from juturna.components import Message, Node, State from juturna.payloads import BytesPayload, ControlPayload, ControlSignal class SlowNode(Node): - def update(self, message: Message): + def update(self, message: Message, state: State): time.sleep(0.01) def generate_stop_message(): diff --git a/tests/test_plugins/nodes/proc/_aggregator/aggregator.py b/tests/test_plugins/nodes/proc/_aggregator/aggregator.py index e2aa2d64..2a17b2ac 100644 --- a/tests/test_plugins/nodes/proc/_aggregator/aggregator.py +++ b/tests/test_plugins/nodes/proc/_aggregator/aggregator.py @@ -8,10 +8,9 @@ Test node. Collect a number of messages, aggregate their content, then return the concatenated version. """ -import typing - from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads._payloads import Batch, AudioPayload @@ -23,7 +22,7 @@ def __init__(self, size: int, **kwargs): self._size = size self._received = 0 - def update(self, message: Message): + def update(self, message: Message, state: State): self.dump_json(message, f'batch_{self._received}.json') self._received += 1 diff --git a/tests/test_plugins/nodes/proc/_amplifier/amplifier.py b/tests/test_plugins/nodes/proc/_amplifier/amplifier.py index 8d266998..69b34053 100644 --- a/tests/test_plugins/nodes/proc/_amplifier/amplifier.py +++ b/tests/test_plugins/nodes/proc/_amplifier/amplifier.py @@ -11,6 +11,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State # BasePayload type is intended to be a placehoder for the input-output types # you intend to use in the node implementation @@ -56,7 +57,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[BasePayload]): + def update(self, message: Message[BasePayload], state: State): """Receive data from upstream, transmit data downstream""" ... diff --git a/tests/test_plugins/nodes/sink/_crasher/crasher.py b/tests/test_plugins/nodes/sink/_crasher/crasher.py index 15ae1513..d3c88188 100644 --- a/tests/test_plugins/nodes/sink/_crasher/crasher.py +++ b/tests/test_plugins/nodes/sink/_crasher/crasher.py @@ -7,10 +7,9 @@ Test sink node. Accumulate received messages. """ -import typing - from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import BasePayload @@ -27,5 +26,5 @@ def start(self): def stop(self): super().stop() - def update(self, message: Message[BasePayload]): + def update(self, message: Message[BasePayload], state: State): self.messages.append(message) diff --git a/tests/test_plugins/nodes/sink/_dumper/dumper.py b/tests/test_plugins/nodes/sink/_dumper/dumper.py index 618d065c..33ad9c51 100644 --- a/tests/test_plugins/nodes/sink/_dumper/dumper.py +++ b/tests/test_plugins/nodes/sink/_dumper/dumper.py @@ -10,6 +10,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import BasePayload @@ -27,7 +28,7 @@ def stop(self): super().stop() self.logger.info(f"{self._received} messages received in total") - def update(self, message: Message[BasePayload]): + def update(self, message: Message[BasePayload], state: State): self._received += 1 self.dump_json(message, f"message_{message.version}.json") self.logger.info( diff --git a/tests/test_plugins/nodes/source/_data_streamer/data_streamer.py b/tests/test_plugins/nodes/source/_data_streamer/data_streamer.py index 122af06f..7cf34d3a 100644 --- a/tests/test_plugins/nodes/source/_data_streamer/data_streamer.py +++ b/tests/test_plugins/nodes/source/_data_streamer/data_streamer.py @@ -12,6 +12,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import BytesPayload @@ -58,7 +59,7 @@ def start(self): def stop(self): super().stop() - def update(self, message: Message[BytesPayload]): + def update(self, message: Message[BytesPayload], state: State): self.transmit(message) self.dump_json(message, f'message_{self._transmitted}.json') diff --git a/tests/test_plugins/nodes/source/_sequencer/sequencer.py b/tests/test_plugins/nodes/source/_sequencer/sequencer.py index 3992e097..ecf6669a 100644 --- a/tests/test_plugins/nodes/source/_sequencer/sequencer.py +++ b/tests/test_plugins/nodes/source/_sequencer/sequencer.py @@ -66,7 +66,7 @@ def start(self): def stop(self): super().stop() - def update(self, message: Message[AudioPayload]): + def update(self, message: Message[AudioPayload], state): self.transmit(message) self.dump_json(message, f'message_{self._transmitted}.json') diff --git a/tests/test_state.py b/tests/test_state.py new file mode 100644 index 00000000..fc6091b6 --- /dev/null +++ b/tests/test_state.py @@ -0,0 +1,57 @@ +from juturna.components import State + + +def test_state_accessors(): + state = State() + + state['key_a'] = 10 + state['key_b'] = 20 + + assert state['key_a'] == 10 + assert state['key_b'] == 20 + + del state['key_a'] + + assert 'key_a' not in state + assert 'key_b' in state + + +def test_state_initial_deltas(): + state = State() + + assert state.deltas() == list() + + +def test_state_deltas(): + state = State() + + state['key_a'] = 10 + state['key_b'] = 20 + + deltas = state.deltas() + + assert ('SET', 'key_a', 10) in deltas + assert ('SET', 'key_b', 20) in deltas + + assert state.deltas() == list() + + +def test_state_delta_overwrite(): + state = State() + + state['key_a'] = 10 + + assert state.deltas() == [('SET', 'key_a', 10)] + assert state.deltas() == list() + + state['key_a'] = 20 + state['key_a'] = 30 + + assert state.deltas() == [('SET', 'key_a', 30)] + assert state.deltas() == list() + + state['key_a'] = 40 + del state['key_a'] + + assert state.deltas() == [('DEL', 'key_a', None)] + assert state.deltas() == list() From 816da43fc5bc9c9eb09b78d4cf2dd6bd5b15f961 Mon Sep 17 00:00:00 2001 From: Antonio Bevilacqua Date: Fri, 12 Jun 2026 14:33:53 +0200 Subject: [PATCH 2/7] feat!: added pipe id to envelope send and received to and from remote service --- .../cli/commands/_juturna_remote_service.py | 15 ++++++++++++- juturna/nodes/proc/_warp/warp.py | 12 +++++++---- juturna/remotizer/c_protos/payloads_pb2.py | 8 +++---- juturna/remotizer/c_protos/state_delta_pb2.py | 21 +++++++++++++++++++ .../c_protos/state_delta_pb2_grpc.py | 13 ++++++++++++ juturna/remotizer/compile_protos.sh | 6 ++++++ juturna/remotizer/expose_protos.sh | 3 ++- juturna/remotizer/protos/payloads.proto | 3 +++ juturna/remotizer/protos/state_delta.proto | 20 ++++++++++++++++++ juturna/remotizer/utils.py | 3 +++ tests/test_node.py | 8 +++---- .../source/_data_streamer/data_streamer.py | 17 ++++++++------- 12 files changed, 107 insertions(+), 22 deletions(-) create mode 100644 juturna/remotizer/c_protos/state_delta_pb2.py create mode 100644 juturna/remotizer/c_protos/state_delta_pb2_grpc.py create mode 100644 juturna/remotizer/protos/state_delta.proto diff --git a/juturna/cli/commands/_juturna_remote_service.py b/juturna/cli/commands/_juturna_remote_service.py index 205f8068..42cc034a 100644 --- a/juturna/cli/commands/_juturna_remote_service.py +++ b/juturna/cli/commands/_juturna_remote_service.py @@ -9,7 +9,7 @@ import grpc -from juturna.components import Message, Node +from juturna.components import Message, Node, State from juturna.remotizer._remote_context import RequestContext from juturna.remotizer._remote_builder import _standalone_builder @@ -73,6 +73,8 @@ def __init__(self, node: Node, remote_name: str): ) self._cleanup_thread.start() + self._pipe_state_store = dict() + logger.info(f'Service initialized for node {node.name}') def _increment_stat(self, stat_name: str): @@ -145,6 +147,13 @@ def SendAndReceive(self, request: ProtoEnvelope, context): tracking_id = next(self._tracking_id_counter) sender = envelope_dict.get('sender') envelope_id = envelope_dict.get('id') + pipe_id = envelope_dict['pipe_id'] + + if self._pipe_state_store.get(pipe_id, None) is None: + self._pipe_state_store[pipe_id] = State() + + # pass the state reference to the node + self.node.link_state(self._pipe_state_store[pipe_id]) if not sender: raise ValueError('Missing sender in request envelope') @@ -189,9 +198,13 @@ def SendAndReceive(self, request: ProtoEnvelope, context): proto_response = message_to_proto(response_message) + state_deltas = self._pipe_state_store[pipe_id] + logger.info(f'deltas for {pipe_id}: {state_deltas}') + response_envelope = create_envelope( message=proto_response, creator=self.remote_name, + pipe_id=pipe_id, configuration={}, metadata={ 'processing_time': time.time() - request_context.created_at diff --git a/juturna/nodes/proc/_warp/warp.py b/juturna/nodes/proc/_warp/warp.py index c6d1f0e0..e5a8b752 100644 --- a/juturna/nodes/proc/_warp/warp.py +++ b/juturna/nodes/proc/_warp/warp.py @@ -19,6 +19,7 @@ from juturna.components import Message from juturna.components import Node +from juturna.components import State class Warp[T_Input, T_Output](Node[T_Input, T_Output]): @@ -82,7 +83,7 @@ def warmup(self): self.logger.info(f'warmup node: {self.name}') - def update(self, message: Message[T_Input]): + def update(self, message: Message[T_Input], state: State): """ Send message via gRPC and wait for response @@ -92,7 +93,9 @@ def update(self, message: Message[T_Input]): Parameters ---------- message : Message[T_Input] - The message to send with input payload type + The message to send with input payload type. + state : State + The node state. """ try: @@ -102,6 +105,7 @@ def update(self, message: Message[T_Input]): envelope = create_envelope( message=message_proto, creator=self.name, + pipe_id=self.pipe_id, request_type=type(T_Input).__name__, response_type=type(T_Output).__name__, priority=0, @@ -111,9 +115,9 @@ def update(self, message: Message[T_Input]): ) envelope.configuration.update(self._remote_config) - self.logger.info(f'sending message (envelope_id={envelope.id})...') - self.logger.info(f'sending message id {message.id}...') + self.logger.info(f'sending message (envelope_id={envelope.id})...') + self.logger.info(f'sending message id {message.id}') response_envelope = self.stub.SendAndReceive( envelope, timeout=self._timeout diff --git a/juturna/remotizer/c_protos/payloads_pb2.py b/juturna/remotizer/c_protos/payloads_pb2.py index 2d2ea354..f8a21ebc 100644 --- a/juturna/remotizer/c_protos/payloads_pb2.py +++ b/juturna/remotizer/c_protos/payloads_pb2.py @@ -8,7 +8,7 @@ _sym_db = _symbol_database.Default() from google.protobuf import any_pb2 as google_dot_protobuf_dot_any__pb2 from google.protobuf import struct_pb2 as google_dot_protobuf_dot_struct__pb2 -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0epayloads.proto\x12\x16juturna.proto.payloads\x1a\x19google/protobuf/any.proto\x1a\x1cgoogle/protobuf/struct.proto"\xa0\x01\n\x11AudioProtoPayload\x12\x12\n\naudio_data\x18\x01 \x01(\x0c\x12\r\n\x05dtype\x18\x02 \x01(\t\x12\r\n\x05shape\x18\x03 \x03(\x05\x12\x15\n\rsampling_rate\x18\x04 \x01(\x05\x12\x10\n\x08channels\x18\x05 \x01(\x05\x12\r\n\x05start\x18\x06 \x01(\x01\x12\x0b\n\x03end\x18\x07 \x01(\x01\x12\x14\n\x0caudio_format\x18\x08 \x01(\t"\x8d\x01\n\x11ImageProtoPayload\x12\x12\n\nimage_data\x18\x01 \x01(\x0c\x12\r\n\x05dtype\x18\x02 \x01(\t\x12\r\n\x05width\x18\x03 \x01(\x05\x12\x0e\n\x06height\x18\x04 \x01(\x05\x12\r\n\x05depth\x18\x05 \x01(\x05\x12\x14\n\x0cpixel_format\x18\x06 \x01(\t\x12\x11\n\ttimestamp\x18\x07 \x01(\x01"\x94\x01\n\x11VideoProtoPayload\x129\n\x06frames\x18\x01 \x03(\x0b2).juturna.proto.payloads.ImageProtoPayload\x12\x19\n\x11frames_per_second\x18\x02 \x01(\x01\x12\r\n\x05start\x18\x03 \x01(\x01\x12\x0b\n\x03end\x18\x04 \x01(\x01\x12\r\n\x05codec\x18\x05 \x01(\t".\n\x11BytesProtoPayload\x12\x0b\n\x03cnt\x18\x01 \x01(\x0c\x12\x0c\n\x04size\x18\x02 \x01(\x03"D\n\nBatchProto\x126\n\x08messages\x18\x01 \x03(\x0b2$.juturna.proto.payloads.ProtoMessage";\n\x12ObjectProtoPayload\x12%\n\x04data\x18\x01 \x01(\x0b2\x17.google.protobuf.Struct"\x8f\x02\n\x0cProtoMessage\x12\x12\n\ncreated_at\x18\x01 \x01(\x01\x12\x0f\n\x07creator\x18\x02 \x01(\t\x12\x0f\n\x07version\x18\x03 \x01(\x05\x12%\n\x07payload\x18\x04 \x01(\x0b2\x14.google.protobuf.Any\x12%\n\x04meta\x18\x05 \x01(\x0b2\x17.google.protobuf.Struct\x12@\n\x06timers\x18\x06 \x03(\x0b20.juturna.proto.payloads.ProtoMessage.TimersEntry\x12\n\n\x02id\x18\n \x01(\x05\x1a-\n\x0bTimersEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\x01:\x028\x01"\xc4\x02\n\rProtoEnvelope\x12\n\n\x02id\x18\x01 \x01(\t\x125\n\x07message\x18\x02 \x01(\x0b2$.juturna.proto.payloads.ProtoMessage\x12\x0e\n\x06sender\x18\x03 \x01(\t\x12\x10\n\x08receiver\x18\x04 \x01(\t\x12\x13\n\x0bresponse_to\x18\x06 \x01(\t\x12\x0b\n\x03ttl\x18\x07 \x01(\x03\x12\x12\n\ncreated_at\x18\x08 \x01(\x01\x12.\n\rconfiguration\x18\t \x01(\x0b2\x17.google.protobuf.Struct\x12)\n\x08metadata\x18\n \x01(\x0b2\x17.google.protobuf.Struct\x12\x10\n\x08priority\x18\x0b \x01(\x05\x12\x14\n\x0crequest_type\x18\x0c \x01(\t\x12\x15\n\rresponse_type\x18\r \x01(\t"\x8d\x01\n\x16CompressedProtoPayload\x12\x13\n\x0bcompression\x18\x01 \x01(\t\x12\x17\n\x0fcompressed_data\x18\x02 \x01(\x0c\x12\x15\n\roriginal_size\x18\x03 \x01(\x03\x12\x17\n\x0fcompressed_size\x18\x04 \x01(\x03\x12\x15\n\roriginal_type\x18\x05 \x01(\tb\x06proto3') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0epayloads.proto\x12\x16juturna.proto.payloads\x1a\x19google/protobuf/any.proto\x1a\x1cgoogle/protobuf/struct.proto"\xa0\x01\n\x11AudioProtoPayload\x12\x12\n\naudio_data\x18\x01 \x01(\x0c\x12\r\n\x05dtype\x18\x02 \x01(\t\x12\r\n\x05shape\x18\x03 \x03(\x05\x12\x15\n\rsampling_rate\x18\x04 \x01(\x05\x12\x10\n\x08channels\x18\x05 \x01(\x05\x12\r\n\x05start\x18\x06 \x01(\x01\x12\x0b\n\x03end\x18\x07 \x01(\x01\x12\x14\n\x0caudio_format\x18\x08 \x01(\t"\x8d\x01\n\x11ImageProtoPayload\x12\x12\n\nimage_data\x18\x01 \x01(\x0c\x12\r\n\x05dtype\x18\x02 \x01(\t\x12\r\n\x05width\x18\x03 \x01(\x05\x12\x0e\n\x06height\x18\x04 \x01(\x05\x12\r\n\x05depth\x18\x05 \x01(\x05\x12\x14\n\x0cpixel_format\x18\x06 \x01(\t\x12\x11\n\ttimestamp\x18\x07 \x01(\x01"\x94\x01\n\x11VideoProtoPayload\x129\n\x06frames\x18\x01 \x03(\x0b2).juturna.proto.payloads.ImageProtoPayload\x12\x19\n\x11frames_per_second\x18\x02 \x01(\x01\x12\r\n\x05start\x18\x03 \x01(\x01\x12\x0b\n\x03end\x18\x04 \x01(\x01\x12\r\n\x05codec\x18\x05 \x01(\t".\n\x11BytesProtoPayload\x12\x0b\n\x03cnt\x18\x01 \x01(\x0c\x12\x0c\n\x04size\x18\x02 \x01(\x03"D\n\nBatchProto\x126\n\x08messages\x18\x01 \x03(\x0b2$.juturna.proto.payloads.ProtoMessage";\n\x12ObjectProtoPayload\x12%\n\x04data\x18\x01 \x01(\x0b2\x17.google.protobuf.Struct"\x8f\x02\n\x0cProtoMessage\x12\x12\n\ncreated_at\x18\x01 \x01(\x01\x12\x0f\n\x07creator\x18\x02 \x01(\t\x12\x0f\n\x07version\x18\x03 \x01(\x05\x12%\n\x07payload\x18\x04 \x01(\x0b2\x14.google.protobuf.Any\x12%\n\x04meta\x18\x05 \x01(\x0b2\x17.google.protobuf.Struct\x12@\n\x06timers\x18\x06 \x03(\x0b20.juturna.proto.payloads.ProtoMessage.TimersEntry\x12\n\n\x02id\x18\n \x01(\x05\x1a-\n\x0bTimersEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\x01:\x028\x01"\xd5\x02\n\rProtoEnvelope\x12\n\n\x02id\x18\x01 \x01(\t\x125\n\x07message\x18\x02 \x01(\x0b2$.juturna.proto.payloads.ProtoMessage\x12\x0e\n\x06sender\x18\x03 \x01(\t\x12\x10\n\x08receiver\x18\x04 \x01(\t\x12\x0f\n\x07pipe_id\x18\x05 \x01(\t\x12\x13\n\x0bresponse_to\x18\x06 \x01(\t\x12\x0b\n\x03ttl\x18\x07 \x01(\x03\x12\x12\n\ncreated_at\x18\x08 \x01(\x01\x12.\n\rconfiguration\x18\t \x01(\x0b2\x17.google.protobuf.Struct\x12)\n\x08metadata\x18\n \x01(\x0b2\x17.google.protobuf.Struct\x12\x10\n\x08priority\x18\x0b \x01(\x05\x12\x14\n\x0crequest_type\x18\x0c \x01(\t\x12\x15\n\rresponse_type\x18\r \x01(\t"\x8d\x01\n\x16CompressedProtoPayload\x12\x13\n\x0bcompression\x18\x01 \x01(\t\x12\x17\n\x0fcompressed_data\x18\x02 \x01(\x0c\x12\x15\n\roriginal_size\x18\x03 \x01(\x03\x12\x17\n\x0fcompressed_size\x18\x04 \x01(\x03\x12\x15\n\roriginal_type\x18\x05 \x01(\tb\x06proto3') _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) _builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'payloads_pb2', _globals) @@ -33,6 +33,6 @@ _globals['_PROTOMESSAGE_TIMERSENTRY']._serialized_start = 963 _globals['_PROTOMESSAGE_TIMERSENTRY']._serialized_end = 1008 _globals['_PROTOENVELOPE']._serialized_start = 1011 - _globals['_PROTOENVELOPE']._serialized_end = 1335 - _globals['_COMPRESSEDPROTOPAYLOAD']._serialized_start = 1338 - _globals['_COMPRESSEDPROTOPAYLOAD']._serialized_end = 1479 + _globals['_PROTOENVELOPE']._serialized_end = 1352 + _globals['_COMPRESSEDPROTOPAYLOAD']._serialized_start = 1355 + _globals['_COMPRESSEDPROTOPAYLOAD']._serialized_end = 1496 diff --git a/juturna/remotizer/c_protos/state_delta_pb2.py b/juturna/remotizer/c_protos/state_delta_pb2.py new file mode 100644 index 00000000..196fb604 --- /dev/null +++ b/juturna/remotizer/c_protos/state_delta_pb2.py @@ -0,0 +1,21 @@ +"""Generated protocol buffer code.""" +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import runtime_version as _runtime_version +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder +_runtime_version.ValidateProtobufRuntimeVersion(_runtime_version.Domain.PUBLIC, 6, 31, 1, '', 'state_delta.proto') +_sym_db = _symbol_database.Default() +from google.protobuf import struct_pb2 as google_dot_protobuf_dot_struct__pb2 +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x11state_delta.proto\x12\x19juturna.proto.state_delta\x1a\x1cgoogle/protobuf/struct.proto"\x9a\x01\n\nDeltaProto\x12<\n\x06action\x18\x01 \x01(\x0e2,.juturna.proto.state_delta.DeltaProto.Action\x12\x0b\n\x03key\x18\x02 \x01(\t\x12%\n\x05value\x18\x03 \x01(\x0b2\x16.google.protobuf.Value"\x1a\n\x06Action\x12\x07\n\x03SET\x10\x00\x12\x07\n\x03DEL\x10\x01"Z\n\x10StateDeltasProto\x12\x0f\n\x07node_id\x18\x01 \x01(\t\x125\n\x06deltas\x18\x02 \x03(\x0b2%.juturna.proto.state_delta.DeltaProtob\x06proto3') +_globals = globals() +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'state_delta_pb2', _globals) +if not _descriptor._USE_C_DESCRIPTORS: + DESCRIPTOR._loaded_options = None + _globals['_DELTAPROTO']._serialized_start = 79 + _globals['_DELTAPROTO']._serialized_end = 233 + _globals['_DELTAPROTO_ACTION']._serialized_start = 207 + _globals['_DELTAPROTO_ACTION']._serialized_end = 233 + _globals['_STATEDELTASPROTO']._serialized_start = 235 + _globals['_STATEDELTASPROTO']._serialized_end = 325 diff --git a/juturna/remotizer/c_protos/state_delta_pb2_grpc.py b/juturna/remotizer/c_protos/state_delta_pb2_grpc.py new file mode 100644 index 00000000..9d38c08b --- /dev/null +++ b/juturna/remotizer/c_protos/state_delta_pb2_grpc.py @@ -0,0 +1,13 @@ +"""Client and server classes corresponding to protobuf-defined services.""" +import grpc +import warnings +GRPC_GENERATED_VERSION = '1.76.0' +GRPC_VERSION = grpc.__version__ +_version_not_supported = False +try: + from grpc._utilities import first_version_is_lower + _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION) +except ImportError: + _version_not_supported = True +if _version_not_supported: + raise RuntimeError(f'The grpc package installed is at version {GRPC_VERSION},' + ' but the generated code in state_delta_pb2_grpc.py depends on' + f' grpcio>={GRPC_GENERATED_VERSION}.' + f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}' + f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.') diff --git a/juturna/remotizer/compile_protos.sh b/juturna/remotizer/compile_protos.sh index 750c5f63..ee47c484 100755 --- a/juturna/remotizer/compile_protos.sh +++ b/juturna/remotizer/compile_protos.sh @@ -19,4 +19,10 @@ python -m grpc_tools.protoc \ --grpc_python_out=$OUT_DIR \ $PROTO_DIR/messaging_service.proto +python -m grpc_tools.protoc \ + -I=$PROTO_DIR \ + --python_out=$OUT_DIR \ + --grpc_python_out=$OUT_DIR \ + $PROTO_DIR/state_delta.proto + echo "protobuf compilation completed!" diff --git a/juturna/remotizer/expose_protos.sh b/juturna/remotizer/expose_protos.sh index 69317691..9e56cc5d 100755 --- a/juturna/remotizer/expose_protos.sh +++ b/juturna/remotizer/expose_protos.sh @@ -10,6 +10,7 @@ protol\ --create-package \ --in-place \ --python-out $OUT_DIR \ - protoc --proto-path=$PROTO_DIR payloads.proto messaging_service.proto + protoc --proto-path=$PROTO_DIR \ + payloads.proto messaging_service.proto state_delta.proto echo "protobuf compilation exposed!" diff --git a/juturna/remotizer/protos/payloads.proto b/juturna/remotizer/protos/payloads.proto index ac1947a8..c6b0c553 100644 --- a/juturna/remotizer/protos/payloads.proto +++ b/juturna/remotizer/protos/payloads.proto @@ -143,6 +143,9 @@ message ProtoEnvelope { // Intended receiver identifier string receiver = 4; + // Original pipe id + string pipe_id = 5; + // Response-to field: ID of the envelope this is responding to // Enables explicit request/response tracking string response_to = 6; diff --git a/juturna/remotizer/protos/state_delta.proto b/juturna/remotizer/protos/state_delta.proto new file mode 100644 index 00000000..0ed0fd9f --- /dev/null +++ b/juturna/remotizer/protos/state_delta.proto @@ -0,0 +1,20 @@ +syntax = "proto3"; + +package juturna.proto.state_delta; + +import "google/protobuf/struct.proto"; + +message DeltaProto { + enum Action { + SET = 0; + DEL = 1; + } + Action action = 1; + string key = 2; + google.protobuf.Value value = 3; +} + +message StateDeltasProto { + string node_id = 1; + repeated DeltaProto deltas = 2; +} diff --git a/juturna/remotizer/utils.py b/juturna/remotizer/utils.py index 4765f711..38e07de2 100644 --- a/juturna/remotizer/utils.py +++ b/juturna/remotizer/utils.py @@ -330,6 +330,7 @@ def create_envelope( configuration: dict[str, Any], metadata: dict[str, Any], creator: str, + pipe_id: str, id: str = str(uuid.uuid4()), priority: int = 1, timeout: int = 30, @@ -343,6 +344,7 @@ def create_envelope( envelope = ProtoEnvelope() envelope.id = id envelope.sender = creator + envelope.pipe_id = pipe_id envelope.created_at = time.time() envelope.ttl = int(timeout) envelope.request_type = request_type @@ -362,6 +364,7 @@ def deserialize_envelope(envelope: ProtoEnvelope) -> dict[str, Any]: envelope_dict = { 'id': envelope.id, 'sender': envelope.sender, + 'pipe_id': envelope.pipe_id, 'response_to': envelope.response_to, 'ttl': envelope.ttl, 'request_type': envelope.request_type, diff --git a/tests/test_node.py b/tests/test_node.py index 5d870cb0..8d505d9e 100644 --- a/tests/test_node.py +++ b/tests/test_node.py @@ -88,8 +88,8 @@ def test_pipeline_draining_on_stop(test_config, wait_for_condition): pipeline.warmup() pipeline.start() - wait_for_condition(lambda: pipeline._nodes['0_stream'].transmitted_count > 10, timeout=5) - sent_count = pipeline._nodes['0_stream'].transmitted_count + wait_for_condition(lambda: pipeline._node_state_store['0_stream']['transmitted'] > 10, timeout=5) + sent_count = pipeline._node_state_store['0_stream']['transmitted'] pipeline.stop() received_messages = pipeline._nodes['2_sink'].messages @@ -140,8 +140,8 @@ def test_pipeline_immediate_stop(test_config, wait_for_condition): pipeline.warmup() pipeline.start() - wait_for_condition(lambda: pipeline._nodes['0_stream'].transmitted_count > 10, timeout=5) - sent_count = pipeline._nodes['0_stream'].transmitted_count + wait_for_condition(lambda: pipeline._node_state_store['0_stream']['transmitted'] > 10, timeout=5) + sent_count = pipeline._node_state_store['0_stream']['transmitted'] received_messages = pipeline._nodes['2_sink'].messages received_count = len(received_messages) diff --git a/tests/test_plugins/nodes/source/_data_streamer/data_streamer.py b/tests/test_plugins/nodes/source/_data_streamer/data_streamer.py index 7cf34d3a..6bf7b7eb 100644 --- a/tests/test_plugins/nodes/source/_data_streamer/data_streamer.py +++ b/tests/test_plugins/nodes/source/_data_streamer/data_streamer.py @@ -31,7 +31,6 @@ def __init__( self._rate_variance = rate_variance self._rate = rate - self._transmitted = 0 self._transmitting = True self.set_source(self._generate, by=1/self._rate, mode='pre') @@ -47,12 +46,15 @@ def _generate(self): return Message[BytesPayload]( creator=self.name, - version=self._transmitted, + version=-1, payload=BytesPayload( cnt=DataStreamer.generate_stream(num_bytes) ) ) + def init_state(self, state: State): + state['transmitted'] = 0 + def start(self): super().start() @@ -60,14 +62,13 @@ def stop(self): super().stop() def update(self, message: Message[BytesPayload], state: State): - self.transmit(message) - self.dump_json(message, f'message_{self._transmitted}.json') + trx = state['transmitted'] + message.version = trx - self._transmitted += 1 + self.transmit(message) + self.dump_json(message, f'message_{trx}.json') - @property - def transmitted_count(self): - return self._transmitted + state['transmitted'] += 1 @staticmethod def generate_stream(sample_length_sec: int, sample_rate: int) -> bytes: From d6ce9f9d8d1c7d59b0b04105332af02f97c63057 Mon Sep 17 00:00:00 2001 From: Antonio Bevilacqua Date: Fri, 12 Jun 2026 15:24:45 +0200 Subject: [PATCH 3/7] fix!: added state to built in node update methods --- juturna/cli/commands/basic_node.template | 3 ++- .../sink/_notifier_http/notifier_http.py | 3 ++- .../nodes/sink/_notifier_udp/notifier_udp.py | 3 ++- .../_notifier_websocket/notifier_websocket.py | 6 ++---- .../_videostream_ffmpeg/videostream_ffmpeg.py | 3 ++- .../nodes/source/_audio_file/audio_file.py | 8 ++++++-- juturna/nodes/source/_audio_rtp/audio_rtp.py | 19 +++++++++++-------- .../source/_audio_rtp_av/audio_rtp_av.py | 3 ++- juturna/nodes/source/_json_http/json_http.py | 11 +++++++---- .../source/_json_websocket/json_websocket.py | 15 ++++++++------- .../nodes/source/_video_file/video_file.py | 10 ++++++---- juturna/nodes/source/_video_rtp/video_rtp.py | 9 ++++++--- .../source/_video_rtp_av/video_rtp_av.py | 12 ++++++++---- tests/test_node.py | 4 ++-- .../source/_data_streamer/data_streamer.py | 7 ++----- 15 files changed, 68 insertions(+), 48 deletions(-) diff --git a/juturna/cli/commands/basic_node.template b/juturna/cli/commands/basic_node.template index 92d5150d..9b8fc15c 100644 --- a/juturna/cli/commands/basic_node.template +++ b/juturna/cli/commands/basic_node.template @@ -11,6 +11,7 @@ import typing from juturna.components import Node from juturna.components import Message +from juturna.components import State # BasePayload type is intended to be a placehoder for the input-output types # you intend to use in the node implementation @@ -56,7 +57,7 @@ class $_node_class_name(Node[BasePayload, BasePayload]): """Destroy the node""" ... - def update(self, message: Message[BasePayload]): + def update(self, message: Message[BasePayload], state: State): """Receive data from upstream, transmit data downstream""" ... diff --git a/juturna/nodes/sink/_notifier_http/notifier_http.py b/juturna/nodes/sink/_notifier_http/notifier_http.py index 7f4f15c6..44f27af9 100644 --- a/juturna/nodes/sink/_notifier_http/notifier_http.py +++ b/juturna/nodes/sink/_notifier_http/notifier_http.py @@ -13,6 +13,7 @@ from juturna.components import Message from juturna.components import Node +from juturna.components import State from juturna.payloads import ObjectPayload @@ -67,7 +68,7 @@ def set_on_config(self, prop: str, value: str): self._endpoint = value - def update(self, message: Message[ObjectPayload]): + def update(self, message: Message[ObjectPayload], state: State): """Receive a message, transmit a message""" to_send = Message[ObjectPayload]( creator=message.creator, diff --git a/juturna/nodes/sink/_notifier_udp/notifier_udp.py b/juturna/nodes/sink/_notifier_udp/notifier_udp.py index 27aa63a5..1af56050 100644 --- a/juturna/nodes/sink/_notifier_udp/notifier_udp.py +++ b/juturna/nodes/sink/_notifier_udp/notifier_udp.py @@ -14,6 +14,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import ObjectPayload @@ -83,7 +84,7 @@ def set_on_config(self, prop: str, value: typing.Any): elif prop == 'port': self._address[1] = value - def update(self, message: Message[ObjectPayload]): + def update(self, message: Message[ObjectPayload], state: State): """Receive a message, transmit a message""" chunks = self._prepare_chunks(message, message.version) diff --git a/juturna/nodes/sink/_notifier_websocket/notifier_websocket.py b/juturna/nodes/sink/_notifier_websocket/notifier_websocket.py index 2304f345..51f34865 100644 --- a/juturna/nodes/sink/_notifier_websocket/notifier_websocket.py +++ b/juturna/nodes/sink/_notifier_websocket/notifier_websocket.py @@ -13,6 +13,7 @@ from juturna.components import Message from juturna.components import Node +from juturna.components import State from juturna.payloads import BasePayload @@ -34,14 +35,13 @@ def __init__(self, endpoint: str, **kwargs): self._endpoint = endpoint - self._sent = 0 self._t = None def warmup(self): """Warmup the node""" self.logger.info(f'[{self.name}] set to endpoint {self._endpoint}') - def update(self, message: Message[BasePayload]): + def update(self, message: Message[BasePayload], state: State): """Receive a message, transmit a message""" meta = dict(message.meta) to_send = Message[BasePayload]( @@ -62,8 +62,6 @@ def update(self, message: Message[BasePayload]): self._t.start() - self._sent += 1 - def _send_message(self, message: Message[BasePayload]): with connect(self._endpoint) as ws: try: diff --git a/juturna/nodes/sink/_videostream_ffmpeg/videostream_ffmpeg.py b/juturna/nodes/sink/_videostream_ffmpeg/videostream_ffmpeg.py index 71bf981a..f22b7dfe 100644 --- a/juturna/nodes/sink/_videostream_ffmpeg/videostream_ffmpeg.py +++ b/juturna/nodes/sink/_videostream_ffmpeg/videostream_ffmpeg.py @@ -13,6 +13,7 @@ from juturna.components import Message from juturna.components import Node +from juturna.components import State from juturna.payloads import ImagePayload @@ -108,7 +109,7 @@ def stop(self): except Exception: ... - def update(self, message: Message[ImagePayload]): + def update(self, message: Message[ImagePayload], state: State): """Receive a message, transmit a message""" frame = message.payload.image frame_bytes = frame.tobytes() diff --git a/juturna/nodes/source/_audio_file/audio_file.py b/juturna/nodes/source/_audio_file/audio_file.py index 281cf5e0..e8b70b0b 100644 --- a/juturna/nodes/source/_audio_file/audio_file.py +++ b/juturna/nodes/source/_audio_file/audio_file.py @@ -18,6 +18,8 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State + from juturna.payloads import AudioPayload from juturna.payloads import ControlPayload from juturna.payloads import ControlSignal @@ -53,7 +55,6 @@ def __init__( self._audio = None self._audio_chunks = None - self._transmitted = 0 def warmup(self): # noqa: D102 resampler = av.audio.resampler.AudioResampler( @@ -92,6 +93,7 @@ def _generate_chunks(self) -> Message[AudioPayload | ControlPayload]: if audio_chunk is None: self.logger.info('last chunk processed, stopping') + return Message[ControlPayload]( creator=self.name, payload=ControlPayload(signal=ControlSignal.STOP), @@ -126,7 +128,9 @@ def _iter_audio_chunks(self): yield chunk, sample_offset sample_offset += wave_len - def update(self, message: Message[AudioPayload | ControlPayload]): # noqa: D102 + def update( # noqa: D102 + self, message: Message[AudioPayload | ControlPayload], state: State + ): message.meta['session_id'] = self.pipe_id self.transmit(message) diff --git a/juturna/nodes/source/_audio_rtp/audio_rtp.py b/juturna/nodes/source/_audio_rtp/audio_rtp.py index 2685abdd..1a4991de 100644 --- a/juturna/nodes/source/_audio_rtp/audio_rtp.py +++ b/juturna/nodes/source/_audio_rtp/audio_rtp.py @@ -15,9 +15,11 @@ import numpy as np -from juturna.components import _resource_broker as rb from juturna.components import Message from juturna.components import Node +from juturna.components import State +from juturna.components import _resource_broker as rb + from juturna.payloads import BytesPayload, AudioPayload from juturna.names import ComponentStatus @@ -79,7 +81,6 @@ def __init__( * self._block_size * self._audio_rate ) - self._abs_recv = 0 # infer incoming channel by encoding_clock_chan self._in_channels = AudioRTP._parse_audio_channels(encoding_clock_chan) @@ -212,7 +213,7 @@ def configuration(self) -> dict: return base_config - def update(self, message: Message[BytesPayload]): + def update(self, message: Message[BytesPayload], state: State): """Read a message, return a message""" if not self._subprocess_running: return @@ -221,24 +222,26 @@ def update(self, message: Message[BytesPayload]): message.payload.cnt, self._in_channels ) + _abs_recv = state.get('abs_recv', 0) + to_send = Message[AudioPayload]( creator=self.name, - version=self._abs_recv, + version=_abs_recv, payload=AudioPayload( audio=waveform, sampling_rate=self._audio_rate, channels=self._channels, - start=self._block_size * self._abs_recv, - end=self._block_size * self._abs_recv + self._block_size, + start=self._block_size * _abs_recv, + end=self._block_size * _abs_recv + self._block_size, ), ) - to_send.meta['source_recv'] = self._abs_recv + to_send.meta['source_recv'] = _abs_recv self.transmit(to_send) self.logger.info(f'transmitting message {to_send.version}') - self._abs_recv += 1 + state['abs_recv'] = _abs_recv + 1 def clear_source(self): """Clear any source functions defined on the node""" diff --git a/juturna/nodes/source/_audio_rtp_av/audio_rtp_av.py b/juturna/nodes/source/_audio_rtp_av/audio_rtp_av.py index a3482b47..4691476e 100644 --- a/juturna/nodes/source/_audio_rtp_av/audio_rtp_av.py +++ b/juturna/nodes/source/_audio_rtp_av/audio_rtp_av.py @@ -16,6 +16,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import AudioPayload from juturna.components import _resource_broker as rb @@ -140,7 +141,7 @@ def stop(self): self._t.join() super().stop() - def update(self, message: Message[AudioPayload]): + def update(self, message: Message[AudioPayload], state: State): """Receive data from upstream, transmit data downstream""" self.logger.debug('update method not implemented for source node') diff --git a/juturna/nodes/source/_json_http/json_http.py b/juturna/nodes/source/_json_http/json_http.py index 6dd5c0e9..85a65153 100644 --- a/juturna/nodes/source/_json_http/json_http.py +++ b/juturna/nodes/source/_json_http/json_http.py @@ -25,6 +25,7 @@ from juturna.components import _resource_broker as rb from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import ObjectPayload @@ -59,7 +60,6 @@ def __init__( self._endpoint: str = endpoint.lstrip('/') self._httpd: HTTPServer | None = None self._thread: threading.Thread | None = None - self._sent: int = 0 def configure(self) -> None: """Configure the node before warming up""" @@ -120,12 +120,17 @@ def destroy(self) -> None: self._httpd.server_close() self._httpd = None - def update(self, message: Message[ObjectPayload]) -> None: + def update(self, message: Message[ObjectPayload], state: State) -> None: """Receive an update message""" self.logger.info(f'HTTP server received a message: {message}') + _sent = state.get('sent', 0) + message.version = _sent + self.transmit(message) + state['sent'] = _sent + 1 + def _make_handler(self) -> type[BaseHTTPRequestHandler]: node = self @@ -164,12 +169,10 @@ def do_POST(self) -> None: msg = Message[ObjectPayload]( creator=node.name, - version=node._sent, payload=ObjectPayload.from_dict(json_content), ) node.put(msg) - node._sent += 1 self.send_response(HTTPStatus.ACCEPTED) self.end_headers() diff --git a/juturna/nodes/source/_json_websocket/json_websocket.py b/juturna/nodes/source/_json_websocket/json_websocket.py index addb8e1e..40547098 100644 --- a/juturna/nodes/source/_json_websocket/json_websocket.py +++ b/juturna/nodes/source/_json_websocket/json_websocket.py @@ -15,6 +15,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import BytesPayload from juturna.payloads import ObjectPayload @@ -41,7 +42,6 @@ def __init__(self, rtx_host: str, rtx_port: int, **kwargs): self._rtx_host = rtx_host self._rtx_port = rtx_port - self._sent = 0 self._queue: queue.Queue[Message[ObjectPayload]] = queue.Queue() self._thread: threading.Thread | None = None self._server = None @@ -70,7 +70,7 @@ def stop(self): # noqa: D102 if self._thread: self._thread.join(timeout=2) - def update(self, message: Message[BytesPayload]): # noqa: D102 + def update(self, message: Message[BytesPayload], state: State): # noqa: D102 self.logger.info(f'ws server message received: {message.payload.cnt}') try: @@ -80,8 +80,10 @@ def update(self, message: Message[BytesPayload]): # noqa: D102 return + _sent = state.get('sent', 0) + to_send = Message[ObjectPayload]( - creator=self.name, version=self._sent, payload=Draft(ObjectPayload) + creator=self.name, version=_sent, payload=Draft(ObjectPayload) ) for k, v in json_content.items(): @@ -89,16 +91,15 @@ def update(self, message: Message[BytesPayload]): # noqa: D102 self.logger.info('ws source transmitting...') self.transmit(to_send) - self._sent += 1 + + state['sent'] = _sent + 1 def _ws_handler(self, websocket): try: for raw in websocket: payload = BytesPayload(cnt=raw) - msg = Message[BytesPayload]( - creator=self.name, version=self._sent, payload=payload - ) + msg = Message[BytesPayload](creator=self.name, payload=payload) self._queue.put(msg) except Exception as exc: diff --git a/juturna/nodes/source/_video_file/video_file.py b/juturna/nodes/source/_video_file/video_file.py index 8499e94e..103ffff7 100644 --- a/juturna/nodes/source/_video_file/video_file.py +++ b/juturna/nodes/source/_video_file/video_file.py @@ -16,6 +16,8 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State + from juturna.names import PixelFormat from juturna.payloads import BytesPayload, ImagePayload @@ -44,7 +46,6 @@ def __init__(self, video_path: str, width: int, height: int, **kwargs): self._height = height self._video_info = dict() - self._sent = 0 self._ffmpeg_launcher_path = None self._ffmpeg_proc = None @@ -128,15 +129,16 @@ def destroy(self): """Destroy the node""" self.stop() - def update(self, message: Message[BytesPayload]): + def update(self, message: Message[BytesPayload], state: State): """Receive a message, transmit a message""" try: full_frame = np.frombuffer(message.payload.cnt, np.uint8).reshape( (self._height, self._width, 3) ) + _sent = state.get('sent', 0) to_send = Message( creator=self.name, - version=self._sent, + version=_sent, payload=ImagePayload( image=full_frame, width=self._width, @@ -147,7 +149,7 @@ def update(self, message: Message[BytesPayload]): ) self.transmit(to_send) - self._sent += 1 + state['sent'] = _sent + 1 except Exception as _: ... diff --git a/juturna/nodes/source/_video_rtp/video_rtp.py b/juturna/nodes/source/_video_rtp/video_rtp.py index 9f8e66b5..64736ea3 100644 --- a/juturna/nodes/source/_video_rtp/video_rtp.py +++ b/juturna/nodes/source/_video_rtp/video_rtp.py @@ -8,6 +8,7 @@ from juturna.components import Message from juturna.components import Node +from juturna.components import State from juturna.components import _resource_broker as rb from juturna.names import PixelFormat @@ -127,16 +128,18 @@ def configuration(self) -> dict: return base_config - def update(self, message: Message[BytesPayload]): + def update(self, message: Message[BytesPayload], state: State): """Receive a message, transmit a message""" try: full_frame = np.frombuffer(message.payload.cnt, np.uint8).reshape( (self._height, self._width, 3) ) + _sent = state.get('sent', 0) + to_send = Message[ImagePayload]( creator=self.name, - version=self._sent, + version=_sent, payload=ImagePayload( image=full_frame, width=full_frame.shape[0], @@ -147,7 +150,7 @@ def update(self, message: Message[BytesPayload]): ) self.transmit(to_send) - self._sent += 1 + state['sent'] = _sent + 1 except Exception as _: ... diff --git a/juturna/nodes/source/_video_rtp_av/video_rtp_av.py b/juturna/nodes/source/_video_rtp_av/video_rtp_av.py index 33a3cda0..04263f5a 100644 --- a/juturna/nodes/source/_video_rtp_av/video_rtp_av.py +++ b/juturna/nodes/source/_video_rtp_av/video_rtp_av.py @@ -15,6 +15,8 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State + from juturna.components import _resource_broker as rb from juturna.names import PixelFormat from juturna.payloads import BytesPayload, ImagePayload @@ -76,7 +78,6 @@ def __init__( self._sdp_file_path = None self._t = None self._stop_event = threading.Event() - self._sent = 0 def configure(self): """Configure the node""" @@ -102,10 +103,15 @@ def stop(self): self._t.join() super().stop() - def update(self, message: Message[ImagePayload]): + def update(self, message: Message[ImagePayload], state: State): """Receive data from upstream, transmit data downstream""" + _sent = state.get('sent', 0) + + message.version = _sent self.transmit(message) + state['sent'] = _sent + 1 + def _stream_video_blocks(self): self._container = None @@ -146,7 +152,6 @@ def _generate_chunks(self): to_send = Message[ImagePayload]( creator=self.name, - version=self._sent, payload=ImagePayload( image=full_frame, width=full_frame.shape[1], @@ -157,7 +162,6 @@ def _generate_chunks(self): ) self.put(to_send) - self._sent += 1 except Exception as e: if not self._stop_event.is_set(): self.logger.info(f'source unavailable ({e}), retrying...') diff --git a/tests/test_node.py b/tests/test_node.py index 8d505d9e..8618ef5b 100644 --- a/tests/test_node.py +++ b/tests/test_node.py @@ -88,7 +88,7 @@ def test_pipeline_draining_on_stop(test_config, wait_for_condition): pipeline.warmup() pipeline.start() - wait_for_condition(lambda: pipeline._node_state_store['0_stream']['transmitted'] > 10, timeout=5) + wait_for_condition(lambda: pipeline._node_state_store['0_stream'].get('transmitted', 0) > 10, timeout=5) sent_count = pipeline._node_state_store['0_stream']['transmitted'] pipeline.stop() @@ -140,7 +140,7 @@ def test_pipeline_immediate_stop(test_config, wait_for_condition): pipeline.warmup() pipeline.start() - wait_for_condition(lambda: pipeline._node_state_store['0_stream']['transmitted'] > 10, timeout=5) + wait_for_condition(lambda: pipeline._node_state_store['0_stream'].get('transmitted', 0) > 10, timeout=5) sent_count = pipeline._node_state_store['0_stream']['transmitted'] received_messages = pipeline._nodes['2_sink'].messages diff --git a/tests/test_plugins/nodes/source/_data_streamer/data_streamer.py b/tests/test_plugins/nodes/source/_data_streamer/data_streamer.py index 6bf7b7eb..79d7c016 100644 --- a/tests/test_plugins/nodes/source/_data_streamer/data_streamer.py +++ b/tests/test_plugins/nodes/source/_data_streamer/data_streamer.py @@ -52,9 +52,6 @@ def _generate(self): ) ) - def init_state(self, state: State): - state['transmitted'] = 0 - def start(self): super().start() @@ -62,13 +59,13 @@ def stop(self): super().stop() def update(self, message: Message[BytesPayload], state: State): - trx = state['transmitted'] + trx = state.get('transmitted', 0) message.version = trx self.transmit(message) self.dump_json(message, f'message_{trx}.json') - state['transmitted'] += 1 + state['transmitted'] = trx + 1 @staticmethod def generate_stream(sample_length_sec: int, sample_rate: int) -> bytes: From 1c86220802d6ce7c2843dfe551a89580cbd4e950 Mon Sep 17 00:00:00 2001 From: Antonio Bevilacqua Date: Fri, 12 Jun 2026 15:50:58 +0200 Subject: [PATCH 4/7] fix!: plugin nodes changed so that the state is included in their update signature --- .../aggregator_transcript.py | 3 +- plugins/nodes/proc/_meet_echo/meet_echo.py | 34 +++++++++++-------- .../passthrough_identity.py | 3 -- .../proc/_prompter_ollama/prompter_ollama.py | 3 +- plugins/nodes/proc/_rag_chroma/rag_chroma.py | 3 +- .../_summarizer_ollama/summarizer_ollama.py | 3 +- .../nodes/proc/_tracker_yolo/tracker_yolo.py | 3 +- .../_transcriber_kroko/transcriber_kroko.py | 3 +- .../transcriber_parakeet.py | 3 +- .../_transcriber_qwen/transcriber_qwen.py | 3 +- .../_transcriber_whispy/transcriber_whispy.py | 3 +- .../proc/_translator_nllb/translator_nllb.py | 3 +- plugins/nodes/proc/_vad_silero/vad_silero.py | 3 +- .../proc/_yolo_detector/yolo_detector.py | 3 +- .../sink/_notifier_mongo/notifier_mongo.py | 3 +- .../_csv_data_loader/csv_data_loader.py | 3 +- .../source/_file_watcher/file_watcher.py | 3 +- .../source/_image_loader/image_loader.py | 3 +- 18 files changed, 51 insertions(+), 34 deletions(-) diff --git a/plugins/nodes/proc/_aggregator_transcript/aggregator_transcript.py b/plugins/nodes/proc/_aggregator_transcript/aggregator_transcript.py index 5151f16a..e4a0f0bb 100644 --- a/plugins/nodes/proc/_aggregator_transcript/aggregator_transcript.py +++ b/plugins/nodes/proc/_aggregator_transcript/aggregator_transcript.py @@ -14,6 +14,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import ObjectPayload from juturna.payloads import Draft @@ -68,7 +69,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[ObjectPayload]): + def update(self, message: Message[ObjectPayload], state: State): """Receive data from upstream, transmit data downstream""" transcript = message.payload['transcript'] diff --git a/plugins/nodes/proc/_meet_echo/meet_echo.py b/plugins/nodes/proc/_meet_echo/meet_echo.py index 1cd2db44..e6ed8e7a 100644 --- a/plugins/nodes/proc/_meet_echo/meet_echo.py +++ b/plugins/nodes/proc/_meet_echo/meet_echo.py @@ -13,6 +13,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import ObjectPayload from juturna.payloads import Draft @@ -36,10 +37,6 @@ def __init__(self, activation: str, target: str, **kwargs): super().__init__(**kwargs) self._activation = activation - self._messages = list() - self._accumulating = False - - self._sent = 0 def configure(self): """Configure the node""" @@ -68,10 +65,13 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[ObjectPayload]): + def update(self, message: Message[ObjectPayload], state: State): """Receive data from upstream, transmit data downstream""" + _accumulating = state.get('accumulating', False) + _messages = state.get('messages', list()) + # not accumulating, no activation: ignore - if not self._accumulating and not self._has_activation( + if not _accumulating and not self._has_activation( message.payload[self._target] ): self.logger.info('meet echo inactive - skipping') @@ -79,31 +79,32 @@ def update(self, message: Message[ObjectPayload]): return # not accumulating, activation: start accumulating - if not self._accumulating and self._has_activation( + if not _accumulating and self._has_activation( message.payload[self._target] ): - self._accumulating = True + state['accumulating'] = True return # accumulating, message content: keep accumulating - if self._accumulating and not message.payload['silence']: - self._messages.append(message) + if _accumulating and not message.payload['silence']: + _messages.append(message) + + state['messages'] = _messages return # accumulating, silence: command is complete - if self._accumulating and message.payload['silence']: + if _accumulating and message.payload['silence']: full_query = ' '.join( [m.payload['suggestion'] for m in self._messages] ) - self._accumulating = False - self._messages = list() + _sent = state.get('sent', 0) to_send = Message[ObjectPayload]( creator=self.name, - version=self._sent, + version=_sent, payload=Draft(ObjectPayload), timers_from=message, ) @@ -111,7 +112,10 @@ def update(self, message: Message[ObjectPayload]): to_send.payload['input_query'] = full_query self.transmit(to_send) - self._sent += 1 + + state['sent'] = _sent + 1 + state['accumulating'] = False + state['messages'] = list() def _has_activation(self, text: str): clean_activation = ( diff --git a/plugins/nodes/proc/_passthrough_identity/passthrough_identity.py b/plugins/nodes/proc/_passthrough_identity/passthrough_identity.py index b1db7aaf..53506455 100644 --- a/plugins/nodes/proc/_passthrough_identity/passthrough_identity.py +++ b/plugins/nodes/proc/_passthrough_identity/passthrough_identity.py @@ -32,7 +32,6 @@ def __init__(self, delay: int, **kwargs): super().__init__(**kwargs) self._delay = delay - self._transmitted = 0 def update(self, message: Message[BasePayload], state: State): """Receive a message from downstream, transmit a message upstream""" @@ -49,8 +48,6 @@ def update(self, message: Message[BasePayload], state: State): to_send.meta = dict(message.meta) - self._transmitted += 1 - with to_send.timeit(f'{self.name}_delay'): time.sleep(self._delay) diff --git a/plugins/nodes/proc/_prompter_ollama/prompter_ollama.py b/plugins/nodes/proc/_prompter_ollama/prompter_ollama.py index 825aefca..70502bd1 100644 --- a/plugins/nodes/proc/_prompter_ollama/prompter_ollama.py +++ b/plugins/nodes/proc/_prompter_ollama/prompter_ollama.py @@ -15,6 +15,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import ObjectPayload from juturna.payloads import Batch @@ -108,7 +109,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[ObjectPayload | Batch]): + def update(self, message: Message[ObjectPayload | Batch], state: State): """Receive data from upstream, transmit data downstream""" if isinstance(message.payload, Batch): content = ' '.join( diff --git a/plugins/nodes/proc/_rag_chroma/rag_chroma.py b/plugins/nodes/proc/_rag_chroma/rag_chroma.py index c42244c2..922ba8ff 100644 --- a/plugins/nodes/proc/_rag_chroma/rag_chroma.py +++ b/plugins/nodes/proc/_rag_chroma/rag_chroma.py @@ -14,6 +14,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import ObjectPayload from juturna.payloads import Draft @@ -87,7 +88,7 @@ def start(self): super().start() - def update(self, message: Message[ObjectPayload]): + def update(self, message: Message[ObjectPayload], state: State): """Receive data from upstream, transmit data downstream""" query = message.payload[self._target] diff --git a/plugins/nodes/proc/_summarizer_ollama/summarizer_ollama.py b/plugins/nodes/proc/_summarizer_ollama/summarizer_ollama.py index 47b084b0..202dab35 100644 --- a/plugins/nodes/proc/_summarizer_ollama/summarizer_ollama.py +++ b/plugins/nodes/proc/_summarizer_ollama/summarizer_ollama.py @@ -18,6 +18,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import ObjectPayload from juturna.payloads import Batch @@ -130,7 +131,7 @@ def warmup(self): self.logger.info(f'model {self._model_name} loaded') @safe_exec - def update(self, message: Message[ObjectPayload | Batch]): + def update(self, message: Message[ObjectPayload | Batch], state: State): """Receive data from upstream, transmit data downstream""" msgs = ( message.payload.messages diff --git a/plugins/nodes/proc/_tracker_yolo/tracker_yolo.py b/plugins/nodes/proc/_tracker_yolo/tracker_yolo.py index dffdea5f..d85e31ed 100644 --- a/plugins/nodes/proc/_tracker_yolo/tracker_yolo.py +++ b/plugins/nodes/proc/_tracker_yolo/tracker_yolo.py @@ -12,6 +12,7 @@ from juturna.components import Message from juturna.components import Node +from juturna.components import State from juturna.payloads import ImagePayload @@ -69,7 +70,7 @@ def warmup(self): self.logger.info('tracker ready') - def update(self, message: Message[ImagePayload]): + def update(self, message: Message[ImagePayload], state: State): """Receive a message, transmit a message""" image = message.payload.image results = self._model.predict( diff --git a/plugins/nodes/proc/_transcriber_kroko/transcriber_kroko.py b/plugins/nodes/proc/_transcriber_kroko/transcriber_kroko.py index c6f9d1ee..d62a2513 100644 --- a/plugins/nodes/proc/_transcriber_kroko/transcriber_kroko.py +++ b/plugins/nodes/proc/_transcriber_kroko/transcriber_kroko.py @@ -14,6 +14,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import AudioPayload from juturna.payloads import ObjectPayload @@ -67,7 +68,7 @@ def destroy(self): except Exception: return - def update(self, message: Message[AudioPayload]): + def update(self, message: Message[AudioPayload], state: State): """Receive data from upstream, transmit data downstream""" self.logger.info(f'received {message.version}') diff --git a/plugins/nodes/proc/_transcriber_parakeet/transcriber_parakeet.py b/plugins/nodes/proc/_transcriber_parakeet/transcriber_parakeet.py index fefe333f..e537e667 100644 --- a/plugins/nodes/proc/_transcriber_parakeet/transcriber_parakeet.py +++ b/plugins/nodes/proc/_transcriber_parakeet/transcriber_parakeet.py @@ -19,6 +19,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import AudioPayload from juturna.payloads import ObjectPayload @@ -100,7 +101,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[AudioPayload]): + def update(self, message: Message[AudioPayload], state: State): """Receive data from upstream, transmit data downstream""" self.logger.info(f'trx received {message.version}') diff --git a/plugins/nodes/proc/_transcriber_qwen/transcriber_qwen.py b/plugins/nodes/proc/_transcriber_qwen/transcriber_qwen.py index 151dfbeb..26a30162 100644 --- a/plugins/nodes/proc/_transcriber_qwen/transcriber_qwen.py +++ b/plugins/nodes/proc/_transcriber_qwen/transcriber_qwen.py @@ -19,6 +19,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import AudioPayload from juturna.payloads import ObjectPayload @@ -85,7 +86,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[AudioPayload]): + def update(self, message: Message[AudioPayload], state: State): """Receive data from upstream, transmit data downstream""" self.logger.info(f'trx received {message.version}') diff --git a/plugins/nodes/proc/_transcriber_whispy/transcriber_whispy.py b/plugins/nodes/proc/_transcriber_whispy/transcriber_whispy.py index f112f73f..9f084cde 100644 --- a/plugins/nodes/proc/_transcriber_whispy/transcriber_whispy.py +++ b/plugins/nodes/proc/_transcriber_whispy/transcriber_whispy.py @@ -20,6 +20,7 @@ from juturna.components import Message from juturna.components import Node +from juturna.components import State from juturna.payloads import AudioPayload from juturna.payloads import ObjectPayload @@ -92,7 +93,7 @@ def warmup(self): self.logger.info(f'warmup sources: {self.origins}') - def update(self, message: Message[AudioPayload]): + def update(self, message: Message[AudioPayload], state: State): """Receive data from upstream, transmit data downstream""" self.logger.info(f'received {message.version}') diff --git a/plugins/nodes/proc/_translator_nllb/translator_nllb.py b/plugins/nodes/proc/_translator_nllb/translator_nllb.py index ba1b4b5a..fd0dc48e 100644 --- a/plugins/nodes/proc/_translator_nllb/translator_nllb.py +++ b/plugins/nodes/proc/_translator_nllb/translator_nllb.py @@ -20,6 +20,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import ObjectPayload from juturna.payloads import Draft @@ -124,7 +125,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[ObjectPayload]): + def update(self, message: Message[ObjectPayload], state: State): """Receive data from upstream, transmit data downstream""" self._buffer.append(message) diff --git a/plugins/nodes/proc/_vad_silero/vad_silero.py b/plugins/nodes/proc/_vad_silero/vad_silero.py index d49c9de1..d043db02 100644 --- a/plugins/nodes/proc/_vad_silero/vad_silero.py +++ b/plugins/nodes/proc/_vad_silero/vad_silero.py @@ -16,6 +16,7 @@ from juturna.components import Message from juturna.components import Node +from juturna.copmonents import State from juturna.payloads import Draft from juturna.payloads import AudioPayload @@ -77,7 +78,7 @@ def __init__( self._data = deque(maxlen=self._keep) - def update(self, message: Message[AudioPayload]): + def update(self, message: Message[AudioPayload], state: State): """Update the node""" assert isinstance(self._data, deque) self.logger.info(f'receive: {message.version}') diff --git a/plugins/nodes/proc/_yolo_detector/yolo_detector.py b/plugins/nodes/proc/_yolo_detector/yolo_detector.py index 422f7100..336fca69 100644 --- a/plugins/nodes/proc/_yolo_detector/yolo_detector.py +++ b/plugins/nodes/proc/_yolo_detector/yolo_detector.py @@ -14,6 +14,7 @@ from juturna.components import Message from juturna.components import Node +from juturna.components import State from juturna.payloads._payloads import ImagePayload @@ -84,7 +85,7 @@ def warmup(self): self.logger.info('tracker ready') - def update(self, message: Message[ImagePayload]): + def update(self, message: Message[ImagePayload], state: State): """Process an incoming message""" assert self._model is not None diff --git a/plugins/nodes/sink/_notifier_mongo/notifier_mongo.py b/plugins/nodes/sink/_notifier_mongo/notifier_mongo.py index 1f8e0d71..adc625c2 100644 --- a/plugins/nodes/sink/_notifier_mongo/notifier_mongo.py +++ b/plugins/nodes/sink/_notifier_mongo/notifier_mongo.py @@ -11,6 +11,7 @@ from juturna.components import Message from juturna.components import Node +from juturna.components import State from juturna.payloads import ObjectPayload @@ -59,7 +60,7 @@ def warmup(self): self.logger.info(f'[{self.name}] set to endpoint {self._endpoint}') - def update(self, message: Message[ObjectPayload]): + def update(self, message: Message[ObjectPayload], state: State): """Receive a message, transmit a message""" message = message.to_dict() message['session_id'] = self.pipe_id diff --git a/plugins/nodes/source/_csv_data_loader/csv_data_loader.py b/plugins/nodes/source/_csv_data_loader/csv_data_loader.py index fdc13e26..2ceecc22 100644 --- a/plugins/nodes/source/_csv_data_loader/csv_data_loader.py +++ b/plugins/nodes/source/_csv_data_loader/csv_data_loader.py @@ -12,6 +12,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import ObjectPayload from juturna.payloads import ControlSignal @@ -101,7 +102,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[ObjectPayload]): + def update(self, message: Message[ObjectPayload], state: State): """Receive data from upstream, transmit data downstream""" self.logger.info(f'transmitting {message}') self.transmit(message) diff --git a/plugins/nodes/source/_file_watcher/file_watcher.py b/plugins/nodes/source/_file_watcher/file_watcher.py index bb921abd..91722878 100644 --- a/plugins/nodes/source/_file_watcher/file_watcher.py +++ b/plugins/nodes/source/_file_watcher/file_watcher.py @@ -22,6 +22,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import ObjectPayload @@ -106,7 +107,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[ObjectPayload]): + def update(self, message: Message[ObjectPayload], state: State): """Receive data from upstream, transmit data downstream""" self.transmit(message) diff --git a/plugins/nodes/source/_image_loader/image_loader.py b/plugins/nodes/source/_image_loader/image_loader.py index bb99c68d..329d36c9 100644 --- a/plugins/nodes/source/_image_loader/image_loader.py +++ b/plugins/nodes/source/_image_loader/image_loader.py @@ -26,6 +26,7 @@ from juturna.components import Node from juturna.components import Message +from juturna.components import State from juturna.payloads import ObjectPayload from juturna.payloads import ImagePayload @@ -120,7 +121,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[ObjectPayload]): + def update(self, message: Message[ObjectPayload], state: State): """Receive data from upstream, transmit data downstream""" try: image = Image.open(message.payload['src_path']) From da6e9101bfe84f7c38f1e88173ff09d2ecfacb18 Mon Sep 17 00:00:00 2001 From: Antonio Bevilacqua Date: Fri, 3 Jul 2026 13:37:19 +0200 Subject: [PATCH 5/7] refactor: state is now passed among kwargs so that extending node signature will not require changing it --- .../cli/commands/_juturna_remote_service.py | 10 +++-- juturna/cli/commands/basic_node.template | 4 +- juturna/components/_node.py | 4 +- juturna/components/_state.py | 30 -------------- juturna/nodes/proc/_warp/warp.py | 12 ++++-- .../sink/_notifier_http/notifier_http.py | 3 +- .../nodes/sink/_notifier_udp/notifier_udp.py | 3 +- .../_notifier_websocket/notifier_websocket.py | 3 +- .../_videostream_ffmpeg/videostream_ffmpeg.py | 3 +- .../nodes/source/_audio_file/audio_file.py | 5 +-- juturna/nodes/source/_audio_rtp/audio_rtp.py | 4 +- .../source/_audio_rtp_av/audio_rtp_av.py | 3 +- juturna/nodes/source/_json_http/json_http.py | 5 ++- .../source/_json_websocket/json_websocket.py | 4 +- .../nodes/source/_video_file/video_file.py | 4 +- juturna/nodes/source/_video_rtp/video_rtp.py | 4 +- .../source/_video_rtp_av/video_rtp_av.py | 4 +- juturna/remotizer/c_protos/payloads_pb2.py | 8 ++-- juturna/remotizer/compile_protos.sh | 6 --- juturna/remotizer/expose_protos.sh | 3 +- juturna/remotizer/protos/payloads.proto | 19 +++++---- juturna/remotizer/protos/state_delta.proto | 20 --------- juturna/remotizer/utils.py | 1 + .../aggregator_transcript.py | 3 +- plugins/nodes/proc/_meet_echo/meet_echo.py | 3 +- .../passthrough_identity.py | 3 +- .../proc/_prompter_ollama/prompter_ollama.py | 3 +- plugins/nodes/proc/_rag_chroma/rag_chroma.py | 3 +- .../_summarizer_ollama/summarizer_ollama.py | 3 +- .../nodes/proc/_tracker_yolo/tracker_yolo.py | 3 +- .../_transcriber_kroko/transcriber_kroko.py | 3 +- .../transcriber_parakeet.py | 3 +- .../_transcriber_qwen/transcriber_qwen.py | 3 +- .../_transcriber_whispy/transcriber_whispy.py | 3 +- .../proc/_translator_nllb/translator_nllb.py | 3 +- plugins/nodes/proc/_vad_silero/vad_silero.py | 3 +- .../proc/_yolo_detector/yolo_detector.py | 3 +- .../_csv_data_loader/csv_data_loader.py | 3 +- .../source/_file_watcher/file_watcher.py | 3 +- .../source/_image_loader/image_loader.py | 7 ++-- tests/test_state.py | 41 ------------------- 41 files changed, 77 insertions(+), 181 deletions(-) delete mode 100644 juturna/remotizer/protos/state_delta.proto diff --git a/juturna/cli/commands/_juturna_remote_service.py b/juturna/cli/commands/_juturna_remote_service.py index 42cc034a..49503378 100644 --- a/juturna/cli/commands/_juturna_remote_service.py +++ b/juturna/cli/commands/_juturna_remote_service.py @@ -148,6 +148,12 @@ def SendAndReceive(self, request: ProtoEnvelope, context): sender = envelope_dict.get('sender') envelope_id = envelope_dict.get('id') pipe_id = envelope_dict['pipe_id'] + state = envelope_dict['state'] + + if self._pipe_state_store.get(pipe_id, State()) != state: + logger.warning('stored state mismatch with received state') + + self._pipe_state_store[pipe_id] = state if self._pipe_state_store.get(pipe_id, None) is None: self._pipe_state_store[pipe_id] = State() @@ -198,13 +204,11 @@ def SendAndReceive(self, request: ProtoEnvelope, context): proto_response = message_to_proto(response_message) - state_deltas = self._pipe_state_store[pipe_id] - logger.info(f'deltas for {pipe_id}: {state_deltas}') - response_envelope = create_envelope( message=proto_response, creator=self.remote_name, pipe_id=pipe_id, + state=self._pipe_state_store[pipe_id], configuration={}, metadata={ 'processing_time': time.time() - request_context.created_at diff --git a/juturna/cli/commands/basic_node.template b/juturna/cli/commands/basic_node.template index 9b8fc15c..e54c553c 100644 --- a/juturna/cli/commands/basic_node.template +++ b/juturna/cli/commands/basic_node.template @@ -11,7 +11,6 @@ import typing from juturna.components import Node from juturna.components import Message -from juturna.components import State # BasePayload type is intended to be a placehoder for the input-output types # you intend to use in the node implementation @@ -57,8 +56,9 @@ class $_node_class_name(Node[BasePayload, BasePayload]): """Destroy the node""" ... - def update(self, message: Message[BasePayload], state: State): + def update(self, message: Message[BasePayload], **kwargs): """Receive data from upstream, transmit data downstream""" + # state = kwargs['state'] ... # uncomment next_batch to design custom synchronisation policy diff --git a/juturna/components/_node.py b/juturna/components/_node.py index 27f12699..1790a256 100644 --- a/juturna/components/_node.py +++ b/juturna/components/_node.py @@ -377,7 +377,7 @@ def join(self): def configure(self): ... - def update(self, message: Message[T_Input], state: State): ... + def update(self, message: Message[T_Input], **kwargs): ... def set_on_config(self, prop: str, value: Any): ... @@ -431,7 +431,7 @@ def _update(self): with self._pending_condition: self._pending_updates += 1 try: - self.update(batch, self._state) + self.update(batch, state=self._state) finally: with self._pending_condition: self._pending_updates -= 1 diff --git a/juturna/components/_state.py b/juturna/components/_state.py index 8cd389fd..fdafe06e 100644 --- a/juturna/components/_state.py +++ b/juturna/components/_state.py @@ -4,33 +4,3 @@ class State(UserDict): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) - - self._ops = dict() - - def __setitem__(self, key, value): - super().__setitem__(key, value) - - self._ops[key] = ('SET', value) - - def __delitem__(self, key): - _deleted = super().__delitem__(key) - - self._ops[key] = ('DEL', None) - - return _deleted - - def deltas(self): - _ops = [ - (action, key, value) for key, (action, value) in self._ops.items() - ] - - self._ops.clear() - - return _ops - - def apply(self, deltas): - for _op, key, value in deltas: - if _op == 'SET': - super().__setitem__(key, value) - elif _op == 'DEL': - super().__delitem__(key) diff --git a/juturna/nodes/proc/_warp/warp.py b/juturna/nodes/proc/_warp/warp.py index e5a8b752..24636bf7 100644 --- a/juturna/nodes/proc/_warp/warp.py +++ b/juturna/nodes/proc/_warp/warp.py @@ -83,7 +83,7 @@ def warmup(self): self.logger.info(f'warmup node: {self.name}') - def update(self, message: Message[T_Input], state: State): + def update(self, message: Message[T_Input], **kwargs): """ Send message via gRPC and wait for response @@ -94,10 +94,12 @@ def update(self, message: Message[T_Input], state: State): ---------- message : Message[T_Input] The message to send with input payload type. - state : State - The node state. + kwargs : dict + State and more. """ + state: State = kwargs['state'] + try: self.logger.info('converting message to protobuf...') message_proto = message_to_proto(message) @@ -106,6 +108,7 @@ def update(self, message: Message[T_Input], state: State): message=message_proto, creator=self.name, pipe_id=self.pipe_id, + state=state, request_type=type(T_Input).__name__, response_type=type(T_Output).__name__, priority=0, @@ -135,6 +138,9 @@ def update(self, message: Message[T_Input], state: State): self.transmit(to_send) self.logger.info(f'transmit: {to_send.version}') + for k, v in response_envelope.state.items(): + state[k] = v + except grpc.RpcError as e: self.logger.error(f'gRPC error: {e.code()} - {e.details()}') raise diff --git a/juturna/nodes/sink/_notifier_http/notifier_http.py b/juturna/nodes/sink/_notifier_http/notifier_http.py index 44f27af9..b14fefd9 100644 --- a/juturna/nodes/sink/_notifier_http/notifier_http.py +++ b/juturna/nodes/sink/_notifier_http/notifier_http.py @@ -13,7 +13,6 @@ from juturna.components import Message from juturna.components import Node -from juturna.components import State from juturna.payloads import ObjectPayload @@ -68,7 +67,7 @@ def set_on_config(self, prop: str, value: str): self._endpoint = value - def update(self, message: Message[ObjectPayload], state: State): + def update(self, message: Message[ObjectPayload], **kwargs): """Receive a message, transmit a message""" to_send = Message[ObjectPayload]( creator=message.creator, diff --git a/juturna/nodes/sink/_notifier_udp/notifier_udp.py b/juturna/nodes/sink/_notifier_udp/notifier_udp.py index 1af56050..f924a7e5 100644 --- a/juturna/nodes/sink/_notifier_udp/notifier_udp.py +++ b/juturna/nodes/sink/_notifier_udp/notifier_udp.py @@ -14,7 +14,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import ObjectPayload @@ -84,7 +83,7 @@ def set_on_config(self, prop: str, value: typing.Any): elif prop == 'port': self._address[1] = value - def update(self, message: Message[ObjectPayload], state: State): + def update(self, message: Message[ObjectPayload], **kwargs): """Receive a message, transmit a message""" chunks = self._prepare_chunks(message, message.version) diff --git a/juturna/nodes/sink/_notifier_websocket/notifier_websocket.py b/juturna/nodes/sink/_notifier_websocket/notifier_websocket.py index 51f34865..26d4a5f0 100644 --- a/juturna/nodes/sink/_notifier_websocket/notifier_websocket.py +++ b/juturna/nodes/sink/_notifier_websocket/notifier_websocket.py @@ -13,7 +13,6 @@ from juturna.components import Message from juturna.components import Node -from juturna.components import State from juturna.payloads import BasePayload @@ -41,7 +40,7 @@ def warmup(self): """Warmup the node""" self.logger.info(f'[{self.name}] set to endpoint {self._endpoint}') - def update(self, message: Message[BasePayload], state: State): + def update(self, message: Message[BasePayload], **kwargs): """Receive a message, transmit a message""" meta = dict(message.meta) to_send = Message[BasePayload]( diff --git a/juturna/nodes/sink/_videostream_ffmpeg/videostream_ffmpeg.py b/juturna/nodes/sink/_videostream_ffmpeg/videostream_ffmpeg.py index f22b7dfe..55aadb9b 100644 --- a/juturna/nodes/sink/_videostream_ffmpeg/videostream_ffmpeg.py +++ b/juturna/nodes/sink/_videostream_ffmpeg/videostream_ffmpeg.py @@ -13,7 +13,6 @@ from juturna.components import Message from juturna.components import Node -from juturna.components import State from juturna.payloads import ImagePayload @@ -109,7 +108,7 @@ def stop(self): except Exception: ... - def update(self, message: Message[ImagePayload], state: State): + def update(self, message: Message[ImagePayload], **kwargs): """Receive a message, transmit a message""" frame = message.payload.image frame_bytes = frame.tobytes() diff --git a/juturna/nodes/source/_audio_file/audio_file.py b/juturna/nodes/source/_audio_file/audio_file.py index e8b70b0b..2154d161 100644 --- a/juturna/nodes/source/_audio_file/audio_file.py +++ b/juturna/nodes/source/_audio_file/audio_file.py @@ -18,7 +18,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import AudioPayload from juturna.payloads import ControlPayload @@ -128,9 +127,7 @@ def _iter_audio_chunks(self): yield chunk, sample_offset sample_offset += wave_len - def update( # noqa: D102 - self, message: Message[AudioPayload | ControlPayload], state: State - ): + def update(self, message: Message[AudioPayload | ControlPayload], **kwargs): # noqa: D102 message.meta['session_id'] = self.pipe_id self.transmit(message) diff --git a/juturna/nodes/source/_audio_rtp/audio_rtp.py b/juturna/nodes/source/_audio_rtp/audio_rtp.py index 1a4991de..61e1cf4f 100644 --- a/juturna/nodes/source/_audio_rtp/audio_rtp.py +++ b/juturna/nodes/source/_audio_rtp/audio_rtp.py @@ -17,7 +17,6 @@ from juturna.components import Message from juturna.components import Node -from juturna.components import State from juturna.components import _resource_broker as rb from juturna.payloads import BytesPayload, AudioPayload @@ -213,7 +212,7 @@ def configuration(self) -> dict: return base_config - def update(self, message: Message[BytesPayload], state: State): + def update(self, message: Message[BytesPayload], **kwargs): """Read a message, return a message""" if not self._subprocess_running: return @@ -222,6 +221,7 @@ def update(self, message: Message[BytesPayload], state: State): message.payload.cnt, self._in_channels ) + state: dict = kwargs.get('state') _abs_recv = state.get('abs_recv', 0) to_send = Message[AudioPayload]( diff --git a/juturna/nodes/source/_audio_rtp_av/audio_rtp_av.py b/juturna/nodes/source/_audio_rtp_av/audio_rtp_av.py index 4691476e..1d1f4dec 100644 --- a/juturna/nodes/source/_audio_rtp_av/audio_rtp_av.py +++ b/juturna/nodes/source/_audio_rtp_av/audio_rtp_av.py @@ -16,7 +16,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import AudioPayload from juturna.components import _resource_broker as rb @@ -141,7 +140,7 @@ def stop(self): self._t.join() super().stop() - def update(self, message: Message[AudioPayload], state: State): + def update(self, message: Message[AudioPayload], **kwargs): """Receive data from upstream, transmit data downstream""" self.logger.debug('update method not implemented for source node') diff --git a/juturna/nodes/source/_json_http/json_http.py b/juturna/nodes/source/_json_http/json_http.py index 85a65153..8bd5d1d6 100644 --- a/juturna/nodes/source/_json_http/json_http.py +++ b/juturna/nodes/source/_json_http/json_http.py @@ -25,7 +25,7 @@ from juturna.components import _resource_broker as rb from juturna.components import Node from juturna.components import Message -from juturna.components import State + from juturna.payloads import ObjectPayload @@ -120,10 +120,11 @@ def destroy(self) -> None: self._httpd.server_close() self._httpd = None - def update(self, message: Message[ObjectPayload], state: State) -> None: + def update(self, message: Message[ObjectPayload], **kwargs) -> None: """Receive an update message""" self.logger.info(f'HTTP server received a message: {message}') + state = kwargs.get('state') _sent = state.get('sent', 0) message.version = _sent diff --git a/juturna/nodes/source/_json_websocket/json_websocket.py b/juturna/nodes/source/_json_websocket/json_websocket.py index 40547098..5f67e2f2 100644 --- a/juturna/nodes/source/_json_websocket/json_websocket.py +++ b/juturna/nodes/source/_json_websocket/json_websocket.py @@ -15,7 +15,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import BytesPayload from juturna.payloads import ObjectPayload @@ -70,7 +69,7 @@ def stop(self): # noqa: D102 if self._thread: self._thread.join(timeout=2) - def update(self, message: Message[BytesPayload], state: State): # noqa: D102 + def update(self, message: Message[BytesPayload], **kwargs): # noqa: D102 self.logger.info(f'ws server message received: {message.payload.cnt}') try: @@ -80,6 +79,7 @@ def update(self, message: Message[BytesPayload], state: State): # noqa: D102 return + state = kwargs.get('state') _sent = state.get('sent', 0) to_send = Message[ObjectPayload]( diff --git a/juturna/nodes/source/_video_file/video_file.py b/juturna/nodes/source/_video_file/video_file.py index 103ffff7..2a52d903 100644 --- a/juturna/nodes/source/_video_file/video_file.py +++ b/juturna/nodes/source/_video_file/video_file.py @@ -16,7 +16,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.names import PixelFormat from juturna.payloads import BytesPayload, ImagePayload @@ -129,12 +128,13 @@ def destroy(self): """Destroy the node""" self.stop() - def update(self, message: Message[BytesPayload], state: State): + def update(self, message: Message[BytesPayload], **kwargs): """Receive a message, transmit a message""" try: full_frame = np.frombuffer(message.payload.cnt, np.uint8).reshape( (self._height, self._width, 3) ) + state = kwargs.get('state') _sent = state.get('sent', 0) to_send = Message( creator=self.name, diff --git a/juturna/nodes/source/_video_rtp/video_rtp.py b/juturna/nodes/source/_video_rtp/video_rtp.py index 64736ea3..f179952e 100644 --- a/juturna/nodes/source/_video_rtp/video_rtp.py +++ b/juturna/nodes/source/_video_rtp/video_rtp.py @@ -8,7 +8,6 @@ from juturna.components import Message from juturna.components import Node -from juturna.components import State from juturna.components import _resource_broker as rb from juturna.names import PixelFormat @@ -128,13 +127,14 @@ def configuration(self) -> dict: return base_config - def update(self, message: Message[BytesPayload], state: State): + def update(self, message: Message[BytesPayload], **kwargs): """Receive a message, transmit a message""" try: full_frame = np.frombuffer(message.payload.cnt, np.uint8).reshape( (self._height, self._width, 3) ) + state = kwargs.get('state') _sent = state.get('sent', 0) to_send = Message[ImagePayload]( diff --git a/juturna/nodes/source/_video_rtp_av/video_rtp_av.py b/juturna/nodes/source/_video_rtp_av/video_rtp_av.py index 04263f5a..90e39f21 100644 --- a/juturna/nodes/source/_video_rtp_av/video_rtp_av.py +++ b/juturna/nodes/source/_video_rtp_av/video_rtp_av.py @@ -15,7 +15,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.components import _resource_broker as rb from juturna.names import PixelFormat @@ -103,8 +102,9 @@ def stop(self): self._t.join() super().stop() - def update(self, message: Message[ImagePayload], state: State): + def update(self, message: Message[ImagePayload], **kwargs): """Receive data from upstream, transmit data downstream""" + state = kwargs.get('state') _sent = state.get('sent', 0) message.version = _sent diff --git a/juturna/remotizer/c_protos/payloads_pb2.py b/juturna/remotizer/c_protos/payloads_pb2.py index f8a21ebc..854e9281 100644 --- a/juturna/remotizer/c_protos/payloads_pb2.py +++ b/juturna/remotizer/c_protos/payloads_pb2.py @@ -8,7 +8,7 @@ _sym_db = _symbol_database.Default() from google.protobuf import any_pb2 as google_dot_protobuf_dot_any__pb2 from google.protobuf import struct_pb2 as google_dot_protobuf_dot_struct__pb2 -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0epayloads.proto\x12\x16juturna.proto.payloads\x1a\x19google/protobuf/any.proto\x1a\x1cgoogle/protobuf/struct.proto"\xa0\x01\n\x11AudioProtoPayload\x12\x12\n\naudio_data\x18\x01 \x01(\x0c\x12\r\n\x05dtype\x18\x02 \x01(\t\x12\r\n\x05shape\x18\x03 \x03(\x05\x12\x15\n\rsampling_rate\x18\x04 \x01(\x05\x12\x10\n\x08channels\x18\x05 \x01(\x05\x12\r\n\x05start\x18\x06 \x01(\x01\x12\x0b\n\x03end\x18\x07 \x01(\x01\x12\x14\n\x0caudio_format\x18\x08 \x01(\t"\x8d\x01\n\x11ImageProtoPayload\x12\x12\n\nimage_data\x18\x01 \x01(\x0c\x12\r\n\x05dtype\x18\x02 \x01(\t\x12\r\n\x05width\x18\x03 \x01(\x05\x12\x0e\n\x06height\x18\x04 \x01(\x05\x12\r\n\x05depth\x18\x05 \x01(\x05\x12\x14\n\x0cpixel_format\x18\x06 \x01(\t\x12\x11\n\ttimestamp\x18\x07 \x01(\x01"\x94\x01\n\x11VideoProtoPayload\x129\n\x06frames\x18\x01 \x03(\x0b2).juturna.proto.payloads.ImageProtoPayload\x12\x19\n\x11frames_per_second\x18\x02 \x01(\x01\x12\r\n\x05start\x18\x03 \x01(\x01\x12\x0b\n\x03end\x18\x04 \x01(\x01\x12\r\n\x05codec\x18\x05 \x01(\t".\n\x11BytesProtoPayload\x12\x0b\n\x03cnt\x18\x01 \x01(\x0c\x12\x0c\n\x04size\x18\x02 \x01(\x03"D\n\nBatchProto\x126\n\x08messages\x18\x01 \x03(\x0b2$.juturna.proto.payloads.ProtoMessage";\n\x12ObjectProtoPayload\x12%\n\x04data\x18\x01 \x01(\x0b2\x17.google.protobuf.Struct"\x8f\x02\n\x0cProtoMessage\x12\x12\n\ncreated_at\x18\x01 \x01(\x01\x12\x0f\n\x07creator\x18\x02 \x01(\t\x12\x0f\n\x07version\x18\x03 \x01(\x05\x12%\n\x07payload\x18\x04 \x01(\x0b2\x14.google.protobuf.Any\x12%\n\x04meta\x18\x05 \x01(\x0b2\x17.google.protobuf.Struct\x12@\n\x06timers\x18\x06 \x03(\x0b20.juturna.proto.payloads.ProtoMessage.TimersEntry\x12\n\n\x02id\x18\n \x01(\x05\x1a-\n\x0bTimersEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\x01:\x028\x01"\xd5\x02\n\rProtoEnvelope\x12\n\n\x02id\x18\x01 \x01(\t\x125\n\x07message\x18\x02 \x01(\x0b2$.juturna.proto.payloads.ProtoMessage\x12\x0e\n\x06sender\x18\x03 \x01(\t\x12\x10\n\x08receiver\x18\x04 \x01(\t\x12\x0f\n\x07pipe_id\x18\x05 \x01(\t\x12\x13\n\x0bresponse_to\x18\x06 \x01(\t\x12\x0b\n\x03ttl\x18\x07 \x01(\x03\x12\x12\n\ncreated_at\x18\x08 \x01(\x01\x12.\n\rconfiguration\x18\t \x01(\x0b2\x17.google.protobuf.Struct\x12)\n\x08metadata\x18\n \x01(\x0b2\x17.google.protobuf.Struct\x12\x10\n\x08priority\x18\x0b \x01(\x05\x12\x14\n\x0crequest_type\x18\x0c \x01(\t\x12\x15\n\rresponse_type\x18\r \x01(\t"\x8d\x01\n\x16CompressedProtoPayload\x12\x13\n\x0bcompression\x18\x01 \x01(\t\x12\x17\n\x0fcompressed_data\x18\x02 \x01(\x0c\x12\x15\n\roriginal_size\x18\x03 \x01(\x03\x12\x17\n\x0fcompressed_size\x18\x04 \x01(\x03\x12\x15\n\roriginal_type\x18\x05 \x01(\tb\x06proto3') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0epayloads.proto\x12\x16juturna.proto.payloads\x1a\x19google/protobuf/any.proto\x1a\x1cgoogle/protobuf/struct.proto"\xa0\x01\n\x11AudioProtoPayload\x12\x12\n\naudio_data\x18\x01 \x01(\x0c\x12\r\n\x05dtype\x18\x02 \x01(\t\x12\r\n\x05shape\x18\x03 \x03(\x05\x12\x15\n\rsampling_rate\x18\x04 \x01(\x05\x12\x10\n\x08channels\x18\x05 \x01(\x05\x12\r\n\x05start\x18\x06 \x01(\x01\x12\x0b\n\x03end\x18\x07 \x01(\x01\x12\x14\n\x0caudio_format\x18\x08 \x01(\t"\x8d\x01\n\x11ImageProtoPayload\x12\x12\n\nimage_data\x18\x01 \x01(\x0c\x12\r\n\x05dtype\x18\x02 \x01(\t\x12\r\n\x05width\x18\x03 \x01(\x05\x12\x0e\n\x06height\x18\x04 \x01(\x05\x12\r\n\x05depth\x18\x05 \x01(\x05\x12\x14\n\x0cpixel_format\x18\x06 \x01(\t\x12\x11\n\ttimestamp\x18\x07 \x01(\x01"\x94\x01\n\x11VideoProtoPayload\x129\n\x06frames\x18\x01 \x03(\x0b2).juturna.proto.payloads.ImageProtoPayload\x12\x19\n\x11frames_per_second\x18\x02 \x01(\x01\x12\r\n\x05start\x18\x03 \x01(\x01\x12\x0b\n\x03end\x18\x04 \x01(\x01\x12\r\n\x05codec\x18\x05 \x01(\t".\n\x11BytesProtoPayload\x12\x0b\n\x03cnt\x18\x01 \x01(\x0c\x12\x0c\n\x04size\x18\x02 \x01(\x03"D\n\nBatchProto\x126\n\x08messages\x18\x01 \x03(\x0b2$.juturna.proto.payloads.ProtoMessage";\n\x12ObjectProtoPayload\x12%\n\x04data\x18\x01 \x01(\x0b2\x17.google.protobuf.Struct"\x8f\x02\n\x0cProtoMessage\x12\x12\n\ncreated_at\x18\x01 \x01(\x01\x12\x0f\n\x07creator\x18\x02 \x01(\t\x12\x0f\n\x07version\x18\x03 \x01(\x05\x12%\n\x07payload\x18\x04 \x01(\x0b2\x14.google.protobuf.Any\x12%\n\x04meta\x18\x05 \x01(\x0b2\x17.google.protobuf.Struct\x12@\n\x06timers\x18\x06 \x03(\x0b20.juturna.proto.payloads.ProtoMessage.TimersEntry\x12\n\n\x02id\x18\n \x01(\x05\x1a-\n\x0bTimersEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\x01:\x028\x01"\xe4\x02\n\rProtoEnvelope\x12\n\n\x02id\x18\x01 \x01(\t\x125\n\x07message\x18\x02 \x01(\x0b2$.juturna.proto.payloads.ProtoMessage\x12\x0e\n\x06sender\x18\x03 \x01(\t\x12\x10\n\x08receiver\x18\x04 \x01(\t\x12\x0f\n\x07pipe_id\x18\x05 \x01(\t\x12\r\n\x05state\x18\x06 \x01(\t\x12\x13\n\x0bresponse_to\x18\x07 \x01(\t\x12\x0b\n\x03ttl\x18\x08 \x01(\x03\x12\x12\n\ncreated_at\x18\t \x01(\x01\x12.\n\rconfiguration\x18\n \x01(\x0b2\x17.google.protobuf.Struct\x12)\n\x08metadata\x18\x0b \x01(\x0b2\x17.google.protobuf.Struct\x12\x10\n\x08priority\x18\x0c \x01(\x05\x12\x14\n\x0crequest_type\x18\r \x01(\t\x12\x15\n\rresponse_type\x18\x0e \x01(\t"\x8d\x01\n\x16CompressedProtoPayload\x12\x13\n\x0bcompression\x18\x01 \x01(\t\x12\x17\n\x0fcompressed_data\x18\x02 \x01(\x0c\x12\x15\n\roriginal_size\x18\x03 \x01(\x03\x12\x17\n\x0fcompressed_size\x18\x04 \x01(\x03\x12\x15\n\roriginal_type\x18\x05 \x01(\tb\x06proto3') _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) _builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'payloads_pb2', _globals) @@ -33,6 +33,6 @@ _globals['_PROTOMESSAGE_TIMERSENTRY']._serialized_start = 963 _globals['_PROTOMESSAGE_TIMERSENTRY']._serialized_end = 1008 _globals['_PROTOENVELOPE']._serialized_start = 1011 - _globals['_PROTOENVELOPE']._serialized_end = 1352 - _globals['_COMPRESSEDPROTOPAYLOAD']._serialized_start = 1355 - _globals['_COMPRESSEDPROTOPAYLOAD']._serialized_end = 1496 + _globals['_PROTOENVELOPE']._serialized_end = 1367 + _globals['_COMPRESSEDPROTOPAYLOAD']._serialized_start = 1370 + _globals['_COMPRESSEDPROTOPAYLOAD']._serialized_end = 1511 diff --git a/juturna/remotizer/compile_protos.sh b/juturna/remotizer/compile_protos.sh index ee47c484..750c5f63 100755 --- a/juturna/remotizer/compile_protos.sh +++ b/juturna/remotizer/compile_protos.sh @@ -19,10 +19,4 @@ python -m grpc_tools.protoc \ --grpc_python_out=$OUT_DIR \ $PROTO_DIR/messaging_service.proto -python -m grpc_tools.protoc \ - -I=$PROTO_DIR \ - --python_out=$OUT_DIR \ - --grpc_python_out=$OUT_DIR \ - $PROTO_DIR/state_delta.proto - echo "protobuf compilation completed!" diff --git a/juturna/remotizer/expose_protos.sh b/juturna/remotizer/expose_protos.sh index 9e56cc5d..69317691 100755 --- a/juturna/remotizer/expose_protos.sh +++ b/juturna/remotizer/expose_protos.sh @@ -10,7 +10,6 @@ protol\ --create-package \ --in-place \ --python-out $OUT_DIR \ - protoc --proto-path=$PROTO_DIR \ - payloads.proto messaging_service.proto state_delta.proto + protoc --proto-path=$PROTO_DIR payloads.proto messaging_service.proto echo "protobuf compilation exposed!" diff --git a/juturna/remotizer/protos/payloads.proto b/juturna/remotizer/protos/payloads.proto index c6b0c553..a5253596 100644 --- a/juturna/remotizer/protos/payloads.proto +++ b/juturna/remotizer/protos/payloads.proto @@ -146,36 +146,39 @@ message ProtoEnvelope { // Original pipe id string pipe_id = 5; + // Node state object + string state = 6; + // Response-to field: ID of the envelope this is responding to // Enables explicit request/response tracking - string response_to = 6; + string response_to = 7; // Time-to-live in seconds // After this duration, the message may be considered stale/expired - int64 ttl = 7; + int64 ttl = 8; // Envelope creation timestamp (Unix epoch in seconds) - double created_at = 8; + double created_at = 9; // Configuration parameters as key-value pairs // these message are used to the set_on_config call - google.protobuf.Struct configuration = 9; + google.protobuf.Struct configuration = 10; // Envelope metadata as key-value pairs // Different from message.meta - this is envelope-level metadata - google.protobuf.Struct metadata = 10; + google.protobuf.Struct metadata = 11; // message priority (0 = lowest, higher = more urgent) // for future use - int32 priority = 11; + int32 priority = 12; // request payload (not proto) type hint for faster routing/dispatching // e.g., "AudioPayload" - string request_type = 12; + string request_type = 13; // response payload (not proto) type hint for faster serialization // e.g., "ImagePayload" - string response_type = 13; + string response_type = 14; } diff --git a/juturna/remotizer/protos/state_delta.proto b/juturna/remotizer/protos/state_delta.proto deleted file mode 100644 index 0ed0fd9f..00000000 --- a/juturna/remotizer/protos/state_delta.proto +++ /dev/null @@ -1,20 +0,0 @@ -syntax = "proto3"; - -package juturna.proto.state_delta; - -import "google/protobuf/struct.proto"; - -message DeltaProto { - enum Action { - SET = 0; - DEL = 1; - } - Action action = 1; - string key = 2; - google.protobuf.Value value = 3; -} - -message StateDeltasProto { - string node_id = 1; - repeated DeltaProto deltas = 2; -} diff --git a/juturna/remotizer/utils.py b/juturna/remotizer/utils.py index 38e07de2..a4bda952 100644 --- a/juturna/remotizer/utils.py +++ b/juturna/remotizer/utils.py @@ -331,6 +331,7 @@ def create_envelope( metadata: dict[str, Any], creator: str, pipe_id: str, + state: dict[str, Any], id: str = str(uuid.uuid4()), priority: int = 1, timeout: int = 30, diff --git a/plugins/nodes/proc/_aggregator_transcript/aggregator_transcript.py b/plugins/nodes/proc/_aggregator_transcript/aggregator_transcript.py index e4a0f0bb..f2c8f02c 100644 --- a/plugins/nodes/proc/_aggregator_transcript/aggregator_transcript.py +++ b/plugins/nodes/proc/_aggregator_transcript/aggregator_transcript.py @@ -14,7 +14,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import ObjectPayload from juturna.payloads import Draft @@ -69,7 +68,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[ObjectPayload], state: State): + def update(self, message: Message[ObjectPayload], **kwargs): """Receive data from upstream, transmit data downstream""" transcript = message.payload['transcript'] diff --git a/plugins/nodes/proc/_meet_echo/meet_echo.py b/plugins/nodes/proc/_meet_echo/meet_echo.py index e6ed8e7a..55e2a639 100644 --- a/plugins/nodes/proc/_meet_echo/meet_echo.py +++ b/plugins/nodes/proc/_meet_echo/meet_echo.py @@ -65,8 +65,9 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[ObjectPayload], state: State): + def update(self, message: Message[ObjectPayload], **kwargs): """Receive data from upstream, transmit data downstream""" + state: State = kwargs['state'] _accumulating = state.get('accumulating', False) _messages = state.get('messages', list()) diff --git a/plugins/nodes/proc/_passthrough_identity/passthrough_identity.py b/plugins/nodes/proc/_passthrough_identity/passthrough_identity.py index 53506455..1008c9b0 100644 --- a/plugins/nodes/proc/_passthrough_identity/passthrough_identity.py +++ b/plugins/nodes/proc/_passthrough_identity/passthrough_identity.py @@ -11,7 +11,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import BasePayload @@ -33,7 +32,7 @@ def __init__(self, delay: int, **kwargs): self._delay = delay - def update(self, message: Message[BasePayload], state: State): + def update(self, message: Message[BasePayload], **kwargs): """Receive a message from downstream, transmit a message upstream""" self.logger.info( f'message {message.version} received from: {message.creator}' diff --git a/plugins/nodes/proc/_prompter_ollama/prompter_ollama.py b/plugins/nodes/proc/_prompter_ollama/prompter_ollama.py index 70502bd1..3b3bc415 100644 --- a/plugins/nodes/proc/_prompter_ollama/prompter_ollama.py +++ b/plugins/nodes/proc/_prompter_ollama/prompter_ollama.py @@ -15,7 +15,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import ObjectPayload from juturna.payloads import Batch @@ -109,7 +108,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[ObjectPayload | Batch], state: State): + def update(self, message: Message[ObjectPayload | Batch], **kwargs): """Receive data from upstream, transmit data downstream""" if isinstance(message.payload, Batch): content = ' '.join( diff --git a/plugins/nodes/proc/_rag_chroma/rag_chroma.py b/plugins/nodes/proc/_rag_chroma/rag_chroma.py index 922ba8ff..262e2b5c 100644 --- a/plugins/nodes/proc/_rag_chroma/rag_chroma.py +++ b/plugins/nodes/proc/_rag_chroma/rag_chroma.py @@ -14,7 +14,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import ObjectPayload from juturna.payloads import Draft @@ -88,7 +87,7 @@ def start(self): super().start() - def update(self, message: Message[ObjectPayload], state: State): + def update(self, message: Message[ObjectPayload], **kwargs): """Receive data from upstream, transmit data downstream""" query = message.payload[self._target] diff --git a/plugins/nodes/proc/_summarizer_ollama/summarizer_ollama.py b/plugins/nodes/proc/_summarizer_ollama/summarizer_ollama.py index 202dab35..efe0b942 100644 --- a/plugins/nodes/proc/_summarizer_ollama/summarizer_ollama.py +++ b/plugins/nodes/proc/_summarizer_ollama/summarizer_ollama.py @@ -18,7 +18,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import ObjectPayload from juturna.payloads import Batch @@ -131,7 +130,7 @@ def warmup(self): self.logger.info(f'model {self._model_name} loaded') @safe_exec - def update(self, message: Message[ObjectPayload | Batch], state: State): + def update(self, message: Message[ObjectPayload | Batch], **kwargs): """Receive data from upstream, transmit data downstream""" msgs = ( message.payload.messages diff --git a/plugins/nodes/proc/_tracker_yolo/tracker_yolo.py b/plugins/nodes/proc/_tracker_yolo/tracker_yolo.py index d85e31ed..afde759f 100644 --- a/plugins/nodes/proc/_tracker_yolo/tracker_yolo.py +++ b/plugins/nodes/proc/_tracker_yolo/tracker_yolo.py @@ -12,7 +12,6 @@ from juturna.components import Message from juturna.components import Node -from juturna.components import State from juturna.payloads import ImagePayload @@ -70,7 +69,7 @@ def warmup(self): self.logger.info('tracker ready') - def update(self, message: Message[ImagePayload], state: State): + def update(self, message: Message[ImagePayload], **kwargs): """Receive a message, transmit a message""" image = message.payload.image results = self._model.predict( diff --git a/plugins/nodes/proc/_transcriber_kroko/transcriber_kroko.py b/plugins/nodes/proc/_transcriber_kroko/transcriber_kroko.py index d62a2513..42da67a1 100644 --- a/plugins/nodes/proc/_transcriber_kroko/transcriber_kroko.py +++ b/plugins/nodes/proc/_transcriber_kroko/transcriber_kroko.py @@ -14,7 +14,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import AudioPayload from juturna.payloads import ObjectPayload @@ -68,7 +67,7 @@ def destroy(self): except Exception: return - def update(self, message: Message[AudioPayload], state: State): + def update(self, message: Message[AudioPayload], **kwargs): """Receive data from upstream, transmit data downstream""" self.logger.info(f'received {message.version}') diff --git a/plugins/nodes/proc/_transcriber_parakeet/transcriber_parakeet.py b/plugins/nodes/proc/_transcriber_parakeet/transcriber_parakeet.py index e537e667..37583a2d 100644 --- a/plugins/nodes/proc/_transcriber_parakeet/transcriber_parakeet.py +++ b/plugins/nodes/proc/_transcriber_parakeet/transcriber_parakeet.py @@ -19,7 +19,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import AudioPayload from juturna.payloads import ObjectPayload @@ -101,7 +100,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[AudioPayload], state: State): + def update(self, message: Message[AudioPayload], **kwargs): """Receive data from upstream, transmit data downstream""" self.logger.info(f'trx received {message.version}') diff --git a/plugins/nodes/proc/_transcriber_qwen/transcriber_qwen.py b/plugins/nodes/proc/_transcriber_qwen/transcriber_qwen.py index 26a30162..f14b3957 100644 --- a/plugins/nodes/proc/_transcriber_qwen/transcriber_qwen.py +++ b/plugins/nodes/proc/_transcriber_qwen/transcriber_qwen.py @@ -19,7 +19,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import AudioPayload from juturna.payloads import ObjectPayload @@ -86,7 +85,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[AudioPayload], state: State): + def update(self, message: Message[AudioPayload], **kwargs): """Receive data from upstream, transmit data downstream""" self.logger.info(f'trx received {message.version}') diff --git a/plugins/nodes/proc/_transcriber_whispy/transcriber_whispy.py b/plugins/nodes/proc/_transcriber_whispy/transcriber_whispy.py index 9f084cde..4f99af8c 100644 --- a/plugins/nodes/proc/_transcriber_whispy/transcriber_whispy.py +++ b/plugins/nodes/proc/_transcriber_whispy/transcriber_whispy.py @@ -20,7 +20,6 @@ from juturna.components import Message from juturna.components import Node -from juturna.components import State from juturna.payloads import AudioPayload from juturna.payloads import ObjectPayload @@ -93,7 +92,7 @@ def warmup(self): self.logger.info(f'warmup sources: {self.origins}') - def update(self, message: Message[AudioPayload], state: State): + def update(self, message: Message[AudioPayload], **kwargs): """Receive data from upstream, transmit data downstream""" self.logger.info(f'received {message.version}') diff --git a/plugins/nodes/proc/_translator_nllb/translator_nllb.py b/plugins/nodes/proc/_translator_nllb/translator_nllb.py index fd0dc48e..143de33d 100644 --- a/plugins/nodes/proc/_translator_nllb/translator_nllb.py +++ b/plugins/nodes/proc/_translator_nllb/translator_nllb.py @@ -20,7 +20,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import ObjectPayload from juturna.payloads import Draft @@ -125,7 +124,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[ObjectPayload], state: State): + def update(self, message: Message[ObjectPayload], **kwargs): """Receive data from upstream, transmit data downstream""" self._buffer.append(message) diff --git a/plugins/nodes/proc/_vad_silero/vad_silero.py b/plugins/nodes/proc/_vad_silero/vad_silero.py index d043db02..e2c17897 100644 --- a/plugins/nodes/proc/_vad_silero/vad_silero.py +++ b/plugins/nodes/proc/_vad_silero/vad_silero.py @@ -16,7 +16,6 @@ from juturna.components import Message from juturna.components import Node -from juturna.copmonents import State from juturna.payloads import Draft from juturna.payloads import AudioPayload @@ -78,7 +77,7 @@ def __init__( self._data = deque(maxlen=self._keep) - def update(self, message: Message[AudioPayload], state: State): + def update(self, message: Message[AudioPayload], **kwargs): """Update the node""" assert isinstance(self._data, deque) self.logger.info(f'receive: {message.version}') diff --git a/plugins/nodes/proc/_yolo_detector/yolo_detector.py b/plugins/nodes/proc/_yolo_detector/yolo_detector.py index 336fca69..6ad10564 100644 --- a/plugins/nodes/proc/_yolo_detector/yolo_detector.py +++ b/plugins/nodes/proc/_yolo_detector/yolo_detector.py @@ -14,7 +14,6 @@ from juturna.components import Message from juturna.components import Node -from juturna.components import State from juturna.payloads._payloads import ImagePayload @@ -85,7 +84,7 @@ def warmup(self): self.logger.info('tracker ready') - def update(self, message: Message[ImagePayload], state: State): + def update(self, message: Message[ImagePayload], **kwargs): """Process an incoming message""" assert self._model is not None diff --git a/plugins/nodes/source/_csv_data_loader/csv_data_loader.py b/plugins/nodes/source/_csv_data_loader/csv_data_loader.py index 2ceecc22..75f4126b 100644 --- a/plugins/nodes/source/_csv_data_loader/csv_data_loader.py +++ b/plugins/nodes/source/_csv_data_loader/csv_data_loader.py @@ -12,7 +12,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import ObjectPayload from juturna.payloads import ControlSignal @@ -102,7 +101,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[ObjectPayload], state: State): + def update(self, message: Message[ObjectPayload], **kwargs): """Receive data from upstream, transmit data downstream""" self.logger.info(f'transmitting {message}') self.transmit(message) diff --git a/plugins/nodes/source/_file_watcher/file_watcher.py b/plugins/nodes/source/_file_watcher/file_watcher.py index 91722878..eba4a618 100644 --- a/plugins/nodes/source/_file_watcher/file_watcher.py +++ b/plugins/nodes/source/_file_watcher/file_watcher.py @@ -22,7 +22,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import ObjectPayload @@ -107,7 +106,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[ObjectPayload], state: State): + def update(self, message: Message[ObjectPayload], **kwargs): """Receive data from upstream, transmit data downstream""" self.transmit(message) diff --git a/plugins/nodes/source/_image_loader/image_loader.py b/plugins/nodes/source/_image_loader/image_loader.py index 329d36c9..d685f2e6 100644 --- a/plugins/nodes/source/_image_loader/image_loader.py +++ b/plugins/nodes/source/_image_loader/image_loader.py @@ -26,7 +26,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import ObjectPayload from juturna.payloads import ImagePayload @@ -121,7 +120,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[ObjectPayload], state: State): + def update(self, message: Message[ObjectPayload], **kwargs): """Receive data from upstream, transmit data downstream""" try: image = Image.open(message.payload['src_path']) @@ -131,7 +130,9 @@ def update(self, message: Message[ObjectPayload], state: State): image.load() except UnidentifiedImageError: - self.logger.warn(f'cannot load image {message.payload["src_path"]}') + self.logger.warning( + f'cannot load image {message.payload["src_path"]}' + ) return diff --git a/tests/test_state.py b/tests/test_state.py index fc6091b6..ee4bf70b 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -14,44 +14,3 @@ def test_state_accessors(): assert 'key_a' not in state assert 'key_b' in state - - -def test_state_initial_deltas(): - state = State() - - assert state.deltas() == list() - - -def test_state_deltas(): - state = State() - - state['key_a'] = 10 - state['key_b'] = 20 - - deltas = state.deltas() - - assert ('SET', 'key_a', 10) in deltas - assert ('SET', 'key_b', 20) in deltas - - assert state.deltas() == list() - - -def test_state_delta_overwrite(): - state = State() - - state['key_a'] = 10 - - assert state.deltas() == [('SET', 'key_a', 10)] - assert state.deltas() == list() - - state['key_a'] = 20 - state['key_a'] = 30 - - assert state.deltas() == [('SET', 'key_a', 30)] - assert state.deltas() == list() - - state['key_a'] = 40 - del state['key_a'] - - assert state.deltas() == [('DEL', 'key_a', None)] - assert state.deltas() == list() From 116d099a9cd176284579342e22d3aee77e0efdbe Mon Sep 17 00:00:00 2001 From: Antonio Bevilacqua Date: Fri, 3 Jul 2026 16:32:11 +0200 Subject: [PATCH 6/7] fix: node state properly added to envelope, fixed bugs with evenlope ids --- juturna/nodes/proc/_warp/warp.py | 12 ++++++------ juturna/remotizer/utils.py | 6 ++++-- 2 files changed, 10 insertions(+), 8 deletions(-) diff --git a/juturna/nodes/proc/_warp/warp.py b/juturna/nodes/proc/_warp/warp.py index 24636bf7..290bd661 100644 --- a/juturna/nodes/proc/_warp/warp.py +++ b/juturna/nodes/proc/_warp/warp.py @@ -14,7 +14,7 @@ from juturna.remotizer.utils import ( message_to_proto, create_envelope, - deserialize_message, + deserialize_envelope, ) from juturna.components import Message @@ -108,7 +108,7 @@ def update(self, message: Message[T_Input], **kwargs): message=message_proto, creator=self.name, pipe_id=self.pipe_id, - state=state, + state=dict(state), request_type=type(T_Input).__name__, response_type=type(T_Output).__name__, priority=0, @@ -131,14 +131,14 @@ def update(self, message: Message[T_Input], **kwargs): ) self.logger.info('converting response to Message...') - to_send: Message[T_Output] = deserialize_message( - response_envelope.message - ) + envelope_dict = deserialize_envelope(response_envelope) + + to_send: Message[T_Output] = envelope_dict['message'] self.transmit(to_send) self.logger.info(f'transmit: {to_send.version}') - for k, v in response_envelope.state.items(): + for k, v in envelope_dict['state'].items(): state[k] = v except grpc.RpcError as e: diff --git a/juturna/remotizer/utils.py b/juturna/remotizer/utils.py index a4bda952..10136902 100644 --- a/juturna/remotizer/utils.py +++ b/juturna/remotizer/utils.py @@ -332,7 +332,7 @@ def create_envelope( creator: str, pipe_id: str, state: dict[str, Any], - id: str = str(uuid.uuid4()), + id: str = '', priority: int = 1, timeout: int = 30, response_to: str = None, @@ -343,9 +343,10 @@ def create_envelope( assert message is not None envelope = ProtoEnvelope() - envelope.id = id + envelope.id = id or str(uuid.uuid4()) envelope.sender = creator envelope.pipe_id = pipe_id + envelope.state = state envelope.created_at = time.time() envelope.ttl = int(timeout) envelope.request_type = request_type @@ -366,6 +367,7 @@ def deserialize_envelope(envelope: ProtoEnvelope) -> dict[str, Any]: 'id': envelope.id, 'sender': envelope.sender, 'pipe_id': envelope.pipe_id, + 'state': envelope.state, 'response_to': envelope.response_to, 'ttl': envelope.ttl, 'request_type': envelope.request_type, From 2e36e1e14bb4a8e1e39f11581365d513f1571c60 Mon Sep 17 00:00:00 2001 From: Antonio Bevilacqua Date: Fri, 3 Jul 2026 17:04:05 +0200 Subject: [PATCH 7/7] fix: updated signatures for test plugins --- plugins/nodes/sink/_notifier_mongo/notifier_mongo.py | 3 +-- tests/test_node_threads.py | 2 +- tests/test_plugins/nodes/proc/_aggregator/aggregator.py | 3 +-- tests/test_plugins/nodes/proc/_amplifier/amplifier.py | 3 +-- tests/test_plugins/nodes/sink/_crasher/crasher.py | 3 +-- tests/test_plugins/nodes/sink/_dumper/dumper.py | 3 +-- .../test_plugins/nodes/source/_data_streamer/data_streamer.py | 4 ++-- tests/test_plugins/nodes/source/_sequencer/sequencer.py | 2 +- 8 files changed, 9 insertions(+), 14 deletions(-) diff --git a/plugins/nodes/sink/_notifier_mongo/notifier_mongo.py b/plugins/nodes/sink/_notifier_mongo/notifier_mongo.py index adc625c2..5577d09f 100644 --- a/plugins/nodes/sink/_notifier_mongo/notifier_mongo.py +++ b/plugins/nodes/sink/_notifier_mongo/notifier_mongo.py @@ -11,7 +11,6 @@ from juturna.components import Message from juturna.components import Node -from juturna.components import State from juturna.payloads import ObjectPayload @@ -60,7 +59,7 @@ def warmup(self): self.logger.info(f'[{self.name}] set to endpoint {self._endpoint}') - def update(self, message: Message[ObjectPayload], state: State): + def update(self, message: Message[ObjectPayload], **kwargs): """Receive a message, transmit a message""" message = message.to_dict() message['session_id'] = self.pipe_id diff --git a/tests/test_node_threads.py b/tests/test_node_threads.py index 71368212..483b70a4 100644 --- a/tests/test_node_threads.py +++ b/tests/test_node_threads.py @@ -4,7 +4,7 @@ from juturna.payloads import BytesPayload, ControlPayload, ControlSignal class SlowNode(Node): - def update(self, message: Message, state: State): + def update(self, message: Message, **kwargs): time.sleep(0.01) def generate_stop_message(): diff --git a/tests/test_plugins/nodes/proc/_aggregator/aggregator.py b/tests/test_plugins/nodes/proc/_aggregator/aggregator.py index 2a17b2ac..142d26ed 100644 --- a/tests/test_plugins/nodes/proc/_aggregator/aggregator.py +++ b/tests/test_plugins/nodes/proc/_aggregator/aggregator.py @@ -10,7 +10,6 @@ """ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads._payloads import Batch, AudioPayload @@ -22,7 +21,7 @@ def __init__(self, size: int, **kwargs): self._size = size self._received = 0 - def update(self, message: Message, state: State): + def update(self, message: Message, **kwargs): self.dump_json(message, f'batch_{self._received}.json') self._received += 1 diff --git a/tests/test_plugins/nodes/proc/_amplifier/amplifier.py b/tests/test_plugins/nodes/proc/_amplifier/amplifier.py index 69b34053..bba0c944 100644 --- a/tests/test_plugins/nodes/proc/_amplifier/amplifier.py +++ b/tests/test_plugins/nodes/proc/_amplifier/amplifier.py @@ -11,7 +11,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State # BasePayload type is intended to be a placehoder for the input-output types # you intend to use in the node implementation @@ -57,7 +56,7 @@ def destroy(self): """Destroy the node""" ... - def update(self, message: Message[BasePayload], state: State): + def update(self, message: Message[BasePayload], **kwargs): """Receive data from upstream, transmit data downstream""" ... diff --git a/tests/test_plugins/nodes/sink/_crasher/crasher.py b/tests/test_plugins/nodes/sink/_crasher/crasher.py index d3c88188..e632efdc 100644 --- a/tests/test_plugins/nodes/sink/_crasher/crasher.py +++ b/tests/test_plugins/nodes/sink/_crasher/crasher.py @@ -9,7 +9,6 @@ """ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import BasePayload @@ -26,5 +25,5 @@ def start(self): def stop(self): super().stop() - def update(self, message: Message[BasePayload], state: State): + def update(self, message: Message[BasePayload], **kwargs): self.messages.append(message) diff --git a/tests/test_plugins/nodes/sink/_dumper/dumper.py b/tests/test_plugins/nodes/sink/_dumper/dumper.py index 33ad9c51..79cdcd9f 100644 --- a/tests/test_plugins/nodes/sink/_dumper/dumper.py +++ b/tests/test_plugins/nodes/sink/_dumper/dumper.py @@ -10,7 +10,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import BasePayload @@ -28,7 +27,7 @@ def stop(self): super().stop() self.logger.info(f"{self._received} messages received in total") - def update(self, message: Message[BasePayload], state: State): + def update(self, message: Message[BasePayload], **kwargs): self._received += 1 self.dump_json(message, f"message_{message.version}.json") self.logger.info( diff --git a/tests/test_plugins/nodes/source/_data_streamer/data_streamer.py b/tests/test_plugins/nodes/source/_data_streamer/data_streamer.py index 79d7c016..b2fcc3c5 100644 --- a/tests/test_plugins/nodes/source/_data_streamer/data_streamer.py +++ b/tests/test_plugins/nodes/source/_data_streamer/data_streamer.py @@ -12,7 +12,6 @@ from juturna.components import Node from juturna.components import Message -from juturna.components import State from juturna.payloads import BytesPayload @@ -58,7 +57,8 @@ def start(self): def stop(self): super().stop() - def update(self, message: Message[BytesPayload], state: State): + def update(self, message: Message[BytesPayload], **kwargs): + state = kwargs['state'] trx = state.get('transmitted', 0) message.version = trx diff --git a/tests/test_plugins/nodes/source/_sequencer/sequencer.py b/tests/test_plugins/nodes/source/_sequencer/sequencer.py index ecf6669a..1baa82ed 100644 --- a/tests/test_plugins/nodes/source/_sequencer/sequencer.py +++ b/tests/test_plugins/nodes/source/_sequencer/sequencer.py @@ -66,7 +66,7 @@ def start(self): def stop(self): super().stop() - def update(self, message: Message[AudioPayload], state): + def update(self, message: Message[AudioPayload], **kwargs): self.transmit(message) self.dump_json(message, f'message_{self._transmitted}.json')