From c7d3f81dd9c8f4c592f2ff06cc91f0c91c631032 Mon Sep 17 00:00:00 2001 From: flonix8 Date: Fri, 26 Jun 2026 12:38:30 +0200 Subject: [PATCH 01/11] Add initial implementation --- poetry.lock | 8 ++-- pyproject.toml | 2 +- saeapi/config.py | 30 +++++++++++--- saeapi/event_reporter.py | 50 +++++++++++++++++++++++ saeapi/frame_forwarder.py | 86 +++++++++++++++++++++++++++++++++++++++ saeapi/saeapi.py | 45 -------------------- saeapi/stage.py | 57 ++++++++++++-------------- settings.template.yaml | 19 ++++++--- 8 files changed, 204 insertions(+), 93 deletions(-) create mode 100644 saeapi/event_reporter.py create mode 100644 saeapi/frame_forwarder.py delete mode 100644 saeapi/saeapi.py diff --git a/poetry.lock b/poetry.lock index 79af710..52a151e 100644 --- a/poetry.lock +++ b/poetry.lock @@ -800,7 +800,7 @@ ocsp = ["cryptography (>=36.0.1)", "pyopenssl (==23.2.1)", "requests (>=2.31.0)" [[package]] name = "visionapi" -version = "3.5.3" +version = "3.6.0" description = "" optional = false python-versions = "^3.10" @@ -814,8 +814,8 @@ protobuf = "6.31.1" [package.source] type = "git" url = "https://github.com/starwit/vision-api.git" -reference = "3.5.3" -resolved_reference = "e454b95339957b5c99e0f4eebdcd37e095328aee" +reference = "c895601" +resolved_reference = "c89560153c7f3a07c36ebe0ac3392c4dbc64b27d" subdirectory = "python/visionapi" [[package]] @@ -847,4 +847,4 @@ subdirectory = "python" [metadata] lock-version = "2.1" python-versions = ">=3.11,<4.0" -content-hash = "913639b2381be64cb999faab467b911fc28c84cf31e8f3b7362b32e79cb3f146" +content-hash = "4b1baa5fcd36db707d930b47595523a68ac5df8bad76a111fe577f8a18f2ff5f" diff --git a/pyproject.toml b/pyproject.toml index 0c74264..96fde44 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,7 +10,7 @@ package-mode = false [tool.poetry.dependencies] python = ">=3.11,<4.0" pydantic = "2.13.4" -visionapi = { git = "https://github.com/starwit/vision-api.git", subdirectory = "python/visionapi", tag = "3.5.3" } +visionapi = { git = "https://github.com/starwit/vision-api.git", subdirectory = "python/visionapi", rev = "c895601" } visionlib = { git = "https://github.com/starwit/vision-lib.git", subdirectory = "python", tag = "1.0.1" } pydantic-settings = "2.14.1" prometheus-client = "0.25.0" diff --git a/saeapi/config.py b/saeapi/config.py index 138e571..c4c3c39 100644 --- a/saeapi/config.py +++ b/saeapi/config.py @@ -6,16 +6,34 @@ from visionlib.pipeline.settings import LogLevel -class RedisConfig(BaseModel): +class ValkeyConfig(BaseModel): host: str = 'localhost' port: Annotated[int, Field(ge=1, le=65536)] = 6379 - stream_id: str - input_stream_prefix: str - output_stream_prefix: str = 'saeapi' + + +class EventReportingConfig(BaseModel): + # Event messages are written to '{output_stream_prefix}:{instance_id}' + output_stream_prefix: str = 'saeapi-event' + + +class FrameForwardingConfig(BaseModel): + # Source streams are discovered by scanning '{video_source_stream_prefix}:*' + video_source_stream_prefix: str + # Forwarded frames are written to '{output_stream_prefix}:{source_id}' + output_stream_prefix: str = 'saeapi-frame' + poll_interval_s: float = 600.0 + class SaeApiConfig(BaseSettings): log_level: LogLevel = LogLevel.WARNING - redis: RedisConfig + # Identifies this SAE instance; reflected in all output stream names + instance_id: str + # Local instance SAE Valkey, where we read video source frames from + source_valkey: ValkeyConfig + # Remote cloud backend Valkey, where we write all output to + backend_valkey: ValkeyConfig + event_reporting: EventReportingConfig = EventReportingConfig() + frame_forwarding: FrameForwardingConfig prometheus_port: Annotated[int, Field(ge=1024, le=65536)] = 8000 model_config = SettingsConfigDict(env_nested_delimiter='__') @@ -23,4 +41,4 @@ class SaeApiConfig(BaseSettings): @classmethod def settings_customise_sources(cls, settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings): YAML_LOCATION = os.environ.get('SETTINGS_FILE', 'settings.yaml') - return (init_settings, env_settings, YamlConfigSettingsSource(settings_cls, yaml_file=YAML_LOCATION), file_secret_settings) \ No newline at end of file + return (init_settings, env_settings, YamlConfigSettingsSource(settings_cls, yaml_file=YAML_LOCATION), file_secret_settings) diff --git a/saeapi/event_reporter.py b/saeapi/event_reporter.py new file mode 100644 index 0000000..72dc9f8 --- /dev/null +++ b/saeapi/event_reporter.py @@ -0,0 +1,50 @@ +import logging +import time + +from prometheus_client import Counter +from visionapi.sae_pb2 import EventMessage +from visionapi.common_pb2 import MessageType +from visionlib.pipeline import ValkeyPublisher + +from .config import SaeApiConfig + +logger = logging.getLogger(__name__) + +EVENT_MESSAGE_COUNTER = Counter('sae_api_event_message_counter', 'How many event messages have been published', + labelnames=('event',)) + + +class EventReporter: + '''Publishes lifecycle EventMessages about this SAE instance to the backend Valkey.''' + + def __init__(self, config: SaeApiConfig) -> None: + self._config = config + self._stream_key = f'{config.event_reporting.output_stream_prefix}:{config.instance_id}' + self._publish_ctx = ValkeyPublisher(config.backend_valkey.host, config.backend_valkey.port) + self._publish = None + + def __enter__(self): + self._publish = self._publish_ctx.__enter__() + return self + + def __exit__(self, exc_type, exc_value, traceback): + return self._publish_ctx.__exit__(exc_type, exc_value, traceback) + + def report_startup(self): + self.report(EventMessage.EventType.STARTUP) + + def report_shutdown(self): + self.report(EventMessage.EventType.SHUTDOWN) + + def report(self, event_type: EventMessage.EventType): + msg = EventMessage() + msg.instance_id = self._config.instance_id + msg.timestamp_utc_ms = int(time.time() * 1000) + msg.event_type = event_type + msg.type = MessageType.SAE_EVENT + + self._publish(self._stream_key, msg.SerializeToString()) + + event_name = EventMessage.EventType.Name(event_type) + EVENT_MESSAGE_COUNTER.labels(event=event_name).inc() + logger.info(f'Published event message ({event_name}) to {self._stream_key}') diff --git a/saeapi/frame_forwarder.py b/saeapi/frame_forwarder.py new file mode 100644 index 0000000..f7ef5a5 --- /dev/null +++ b/saeapi/frame_forwarder.py @@ -0,0 +1,86 @@ +import logging + +import pybase64 +import valkey +from prometheus_client import Counter, Histogram +from visionapi.common_pb2 import MessageType +from visionapi.sae_pb2 import SaeMessage +from visionlib.pipeline import ValkeyPublisher + +from .config import SaeApiConfig + +logger = logging.getLogger(__name__) + +FRAMES_FORWARDED = Counter('sae_api_frames_forwarded', 'How many frames have been forwarded to the backend') +FORWARD_CYCLE_DURATION = Histogram('sae_api_forward_cycle_duration', 'The time it takes to run one full frame forwarding cycle') + + +class FrameForwarder: + '''Periodically pulls the latest frame from every video source stream on the source Valkey + and forwards a frame-only SaeMessage to the backend Valkey (one output stream per source).''' + + def __init__(self, config: SaeApiConfig) -> None: + self._config = config + self._scan_pattern = f'{config.frame_forwarding.video_source_stream_prefix}:*' + self._output_prefix = config.frame_forwarding.output_stream_prefix + self._instance_id = config.instance_id + + self._source_client = None + self._publish_ctx = ValkeyPublisher(config.backend_valkey.host, config.backend_valkey.port) + self._publish = None + + def __enter__(self): + self._source_client = valkey.Valkey(self._config.source_valkey.host, self._config.source_valkey.port) + self._publish = self._publish_ctx.__enter__() + return self + + def __exit__(self, exc_type, exc_value, traceback): + try: + self._source_client.close() + except Exception as e: + logger.warning('Error while closing source valkey client', exc_info=e) + return self._publish_ctx.__exit__(exc_type, exc_value, traceback) + + @FORWARD_CYCLE_DURATION.time() + def forward_once(self): + stream_keys = list(self._source_client.scan_iter(match=self._scan_pattern, _type='STREAM')) + logger.debug(f'Discovered {len(stream_keys)} source stream(s) matching {self._scan_pattern}') + + for stream_key in stream_keys: + try: + self._forward_stream(stream_key) + except Exception as e: + logger.warning(f'Error while forwarding latest frame from {stream_key!r}', exc_info=e) + + def _forward_stream(self, stream_key: bytes): + entries = self._source_client.xrevrange(stream_key, count=1) + if not entries: + return + + _, fields = entries[0] + proto_data = fields.get(b'proto_data_b64') + if proto_data is not None: + proto_data = pybase64.b64decode(proto_data, validate=True) + else: + proto_data = fields.get(b'proto_data') + if proto_data is None: + logger.warning(f'No proto data field found in latest entry of {stream_key!r}') + return + + sae_msg = SaeMessage() + sae_msg.ParseFromString(proto_data) + + if not sae_msg.HasField('frame'): + logger.debug(f'Latest message in {stream_key!r} has no frame; skipping') + return + + out_msg = SaeMessage() + out_msg.frame.CopyFrom(sae_msg.frame) + out_msg.type = MessageType.SAE + + source_id = stream_key.decode('utf-8').split(':', 1)[1] + output_key = f'{self._output_prefix}:{self._instance_id}:{source_id}' + self._publish(output_key, out_msg.SerializeToString()) + + FRAMES_FORWARDED.inc() + logger.debug(f'Forwarded latest frame from {stream_key!r} to {output_key}') diff --git a/saeapi/saeapi.py b/saeapi/saeapi.py deleted file mode 100644 index 246b2e1..0000000 --- a/saeapi/saeapi.py +++ /dev/null @@ -1,45 +0,0 @@ -import logging -from typing import Any - -from prometheus_client import Counter, Histogram, Summary -from visionapi.sae_pb2 import SaeMessage - -from .config import SaeApiConfig - -logging.basicConfig(format='%(asctime)s %(name)-15s %(levelname)-8s %(processName)-10s %(message)s') -logger = logging.getLogger(__name__) - -GET_DURATION = Histogram('sae_api_get_duration', 'The time it takes to deserialize the proto until returning the tranformed result as a serialized proto', - buckets=(0.0025, 0.005, 0.0075, 0.01, 0.025, 0.05, 0.075, 0.1, 0.15, 0.2, 0.25)) -OBJECT_COUNTER = Counter('sae_api_object_counter', 'How many detections have been transformed') -PROTO_SERIALIZATION_DURATION = Summary('sae_api_proto_serialization_duration', 'The time it takes to create a serialized output proto') -PROTO_DESERIALIZATION_DURATION = Summary('sae_api_proto_deserialization_duration', 'The time it takes to deserialize an input proto') - - -class SaeApi: - def __init__(self, config: SaeApiConfig) -> None: - self.config = config - logger.setLevel(self.config.log_level.value) - - def __call__(self, input_proto) -> Any: - return self.get(input_proto) - - @GET_DURATION.time() - def get(self, input_proto): - sae_msg = self._unpack_proto(input_proto) - - # Your implementation goes (mostly) here - logger.warning('Received SAE message from pipeline') - - return self._pack_proto(sae_msg) - - @PROTO_DESERIALIZATION_DURATION.time() - def _unpack_proto(self, sae_message_bytes): - sae_msg = SaeMessage() - sae_msg.ParseFromString(sae_message_bytes) - - return sae_msg - - @PROTO_SERIALIZATION_DURATION.time() - def _pack_proto(self, sae_msg: SaeMessage): - return sae_msg.SerializeToString() \ No newline at end of file diff --git a/saeapi/stage.py b/saeapi/stage.py index ee2ed4c..fb7274c 100644 --- a/saeapi/stage.py +++ b/saeapi/stage.py @@ -2,26 +2,26 @@ import signal import threading -from prometheus_client import Counter, Histogram, start_http_server -from visionlib.pipeline import ValkeyConsumer, ValkeyPublisher +from prometheus_client import start_http_server from .config import SaeApiConfig -from .saeapi import SaeApi +from .frame_forwarder import FrameForwarder +from .event_reporter import EventReporter +logging.basicConfig(format='%(asctime)s %(name)-15s %(levelname)-8s %(processName)-10s %(message)s') logger = logging.getLogger(__name__) -REDIS_PUBLISH_DURATION = Histogram('sae_api_redis_publish_duration', 'The time it takes to push a message onto the Redis stream', - buckets=(0.0025, 0.005, 0.0075, 0.01, 0.025, 0.05, 0.075, 0.1, 0.15, 0.2, 0.25)) -FRAME_COUNTER = Counter('sae_api_frame_counter', 'How many frames have been consumed from the Redis input stream') def run_stage(): stop_event = threading.Event() - # Register signal handlers + # Register signal handlers. Any termination signal (SIGTERM from an orchestrator, + # SIGINT from Ctrl-C) sets the stop event and triggers a graceful shutdown, including + # the shutdown status report. def sig_handler(signum, _): signame = signal.Signals(signum).name - print(f'Caught signal {signame} ({signum}). Exiting...') + logger.info(f'Caught signal {signame} ({signum}). Exiting...') stop_event.set() signal.signal(signal.SIGTERM, sig_handler) @@ -33,34 +33,29 @@ def sig_handler(signum, _): logger.setLevel(CONFIG.log_level.value) logger.info(f'Starting prometheus metrics endpoint on port {CONFIG.prometheus_port}') - start_http_server(CONFIG.prometheus_port) - logger.info(f'Starting geo mapper stage. Config: {CONFIG.model_dump_json(indent=2)}') - - sae_api = SaeApi(CONFIG) + logger.info(f'Starting sae-api stage. Config: {CONFIG.model_dump_json(indent=2)}') - consumer_ctx = ValkeyConsumer(CONFIG.redis.host, CONFIG.redis.port, - stream_keys=[f'{CONFIG.redis.input_stream_prefix}:{CONFIG.redis.stream_id}']) - publish_ctx = ValkeyPublisher(CONFIG.redis.host, CONFIG.redis.port) - - with consumer_ctx as iter_messages, publish_ctx as publish: - for stream_key, proto_data in iter_messages(): - if stop_event.is_set(): - break + with EventReporter(CONFIG) as status, FrameForwarder(CONFIG) as forwarder: + status.report_startup() - if stream_key is None: - continue + def poll_loop(): + while not stop_event.is_set(): + try: + forwarder.forward_once() + except Exception as e: + logger.error('Error during frame forwarding cycle', exc_info=e) + # Wait returns True as soon as the stop event is set, otherwise after the interval + if stop_event.wait(CONFIG.frame_forwarding.poll_interval_s): + break - stream_id = stream_key.split(':')[1] + poll_thread = threading.Thread(target=poll_loop, name='frame-forwarder', daemon=True) + poll_thread.start() - FRAME_COUNTER.inc() + # Block the main thread until a termination signal arrives + stop_event.wait() - output_proto_data = sae_api.get(proto_data) + poll_thread.join(timeout=10) - if output_proto_data is None: - continue - - with REDIS_PUBLISH_DURATION.time(): - publish(f'{CONFIG.redis.output_stream_prefix}:{stream_id}', output_proto_data) - \ No newline at end of file + status.report_shutdown() diff --git a/settings.template.yaml b/settings.template.yaml index 4a27578..4a31d1e 100644 --- a/settings.template.yaml +++ b/settings.template.yaml @@ -1,8 +1,15 @@ log_level: INFO -redis: - host: redis +instance_id: sae-instance-1 +source_valkey: + host: valkey port: 6379 - stream_id: stream1 - input_stream_prefix: objecttracker - output_stream_prefix: saeapi -prometheus_port: 8000 \ No newline at end of file +backend_valkey: + host: backend-valkey + port: 6379 +event_reporting: + output_stream_prefix: saeapi-event +frame_forwarding: + video_source_stream_prefix: videosource + output_stream_prefix: saeapi-frame + poll_interval_s: 600 +prometheus_port: 8000 From b69a0009cf961f79c20b78afeea6464a5330de33 Mon Sep 17 00:00:00 2001 From: flonix8 Date: Fri, 26 Jun 2026 14:03:53 +0200 Subject: [PATCH 02/11] Add tests --- tests/conftest.py | 34 +++++++- tests/test_config.py | 42 +++++----- tests/test_event_reporter.py | 49 +++++++++++ tests/test_frame_forwarder.py | 92 +++++++++++++++++++++ tests/test_imports.py | 16 +++- tests/test_stage.py | 151 +++++++++++++++++++++++++--------- 6 files changed, 318 insertions(+), 66 deletions(-) create mode 100644 tests/test_event_reporter.py create mode 100644 tests/test_frame_forwarder.py diff --git a/tests/conftest.py b/tests/conftest.py index 2eeec5b..50e96d3 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,6 +1,38 @@ import pytest +from saeapi.config import (EventReportingConfig, FrameForwardingConfig, + SaeApiConfig, ValkeyConfig) + + # This is necessary to prevent tests from accidentally loading real config files @pytest.fixture(autouse=True) def set_settings_file_location(monkeypatch): - monkeypatch.setenv('SETTINGS_FILE', '/tmp/should_not_exist.yaml') \ No newline at end of file + monkeypatch.setenv('SETTINGS_FILE', '/tmp/should_not_exist.yaml') + + +@pytest.fixture +def make_config(): + '''Factory for a fully-populated SaeApiConfig, so tests don't depend on + settings.yaml or environment variables.''' + def _make_config( + instance_id='inst1', + video_source_stream_prefix='videosource', + frame_output_prefix='saeapi-frame', + event_output_prefix='saeapi-event', + poll_interval_s=600.0, + ): + return SaeApiConfig( + log_level='WARNING', + instance_id=instance_id, + source_valkey=ValkeyConfig(host='source-host', port=6379), + backend_valkey=ValkeyConfig(host='backend-host', port=6380), + event_reporting=EventReportingConfig( + output_stream_prefix=event_output_prefix, + ), + frame_forwarding=FrameForwardingConfig( + video_source_stream_prefix=video_source_stream_prefix, + output_stream_prefix=frame_output_prefix, + poll_interval_s=poll_interval_s, + ), + ) + return _make_config diff --git a/tests/test_config.py b/tests/test_config.py index 66ffe0b..755bcdf 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,35 +1,35 @@ import pytest from pydantic import ValidationError -from saeapi.config import SaeApiConfig, RedisConfig +from saeapi.config import (FrameForwardingConfig, SaeApiConfig, ValkeyConfig) def test_incomplete_config(): with pytest.raises(ValidationError): SaeApiConfig( - log_level="INFO", - redis=RedisConfig( - host="localhost", - port=6379 - # Missing stream_id and input_stream_prefix - ) + log_level='INFO', + # Missing instance_id, source_valkey, backend_valkey, frame_forwarding ) -def test_complete_config(): + +def test_minimal_config(): config = SaeApiConfig( - log_level="INFO", - redis=RedisConfig( - host="localhost", - port=6379, - stream_id="stream1", - input_stream_prefix="saeapi_input" + log_level='INFO', + instance_id='inst1', + source_valkey=ValkeyConfig(host='source-host', port=6379), + backend_valkey=ValkeyConfig(host='backend-host', port=6380), + frame_forwarding=FrameForwardingConfig( + video_source_stream_prefix='videosource', ), - prometheus_port=9000 + prometheus_port=9000, ) - assert config.log_level.name == "INFO" - assert config.redis.host == "localhost" - assert config.redis.port == 6379 - assert config.redis.stream_id == "stream1" - assert config.redis.input_stream_prefix == "saeapi_input" - assert config.prometheus_port == 9000 \ No newline at end of file + assert config.log_level.name == 'INFO' + assert config.instance_id == 'inst1' + assert config.source_valkey.host == 'source-host' + assert config.backend_valkey.port == 6380 + assert config.frame_forwarding.video_source_stream_prefix == 'videosource' + # Defaults are applied + assert config.frame_forwarding.output_stream_prefix == 'saeapi-frame' + assert config.event_reporting.output_stream_prefix == 'saeapi-event' + assert config.prometheus_port == 9000 diff --git a/tests/test_event_reporter.py b/tests/test_event_reporter.py new file mode 100644 index 0000000..5d957af --- /dev/null +++ b/tests/test_event_reporter.py @@ -0,0 +1,49 @@ +from unittest.mock import patch + +import pytest +from visionapi.common_pb2 import MessageType +from visionapi.sae_pb2 import EventMessage + +from saeapi.event_reporter import EventReporter + + +@pytest.fixture +def publisher_mock(): + '''Mocks the ValkeyPublisher used by the EventReporter and yields the + publish callable (what `__enter__` returns), so tests can assert on the + messages that were published.''' + with patch('saeapi.event_reporter.ValkeyPublisher') as mock_publisher: + yield mock_publisher.return_value.__enter__.return_value + + +def test_report_startup_publishes_event(make_config, publisher_mock): + config = make_config(instance_id='inst1', event_output_prefix='saeapi-event') + + with EventReporter(config) as reporter: + reporter.report_startup() + + publisher_mock.assert_called_once() + stream_key, proto_data = publisher_mock.call_args.args + + assert stream_key == 'saeapi-event:inst1' + + msg = EventMessage() + msg.ParseFromString(proto_data) + assert msg.instance_id == 'inst1' + assert msg.event_type == EventMessage.EventType.STARTUP + assert msg.type == MessageType.SAE_EVENT + assert msg.timestamp_utc_ms > 0 + + +def test_report_shutdown_publishes_event(make_config, publisher_mock): + config = make_config(instance_id='inst1') + + with EventReporter(config) as reporter: + reporter.report_shutdown() + + publisher_mock.assert_called_once() + _, proto_data = publisher_mock.call_args.args + + msg = EventMessage() + msg.ParseFromString(proto_data) + assert msg.event_type == EventMessage.EventType.SHUTDOWN diff --git a/tests/test_frame_forwarder.py b/tests/test_frame_forwarder.py new file mode 100644 index 0000000..e2c2cb7 --- /dev/null +++ b/tests/test_frame_forwarder.py @@ -0,0 +1,92 @@ +from unittest.mock import MagicMock, patch + +import pybase64 +import pytest +from visionapi.common_pb2 import MessageType +from visionapi.sae_pb2 import SaeMessage + +from saeapi.frame_forwarder import FrameForwarder + + +# Stand-in for raw image bytes carried in a frame +FRAME_DATA = b'\x01\x02\x03fake-frame-bytes\xff' + + +def _make_frame_msg_bytes(timestamp: int = 1) -> bytes: + '''A SaeMessage carrying a frame (the kind FrameForwarder should forward).''' + msg = SaeMessage() + msg.frame.timestamp_utc_ms = timestamp + msg.frame.source_id = 'cam1' + msg.frame.frame_data = FRAME_DATA + msg.type = MessageType.SAE + return msg.SerializeToString() + + +def _stream_entry(proto_data: bytes, b64=True): + '''Mimics a single Valkey stream entry as returned by xrevrange.''' + if b64: + fields = {b'proto_data_b64': pybase64.b64encode(proto_data)} + else: + fields = {b'proto_data': proto_data} + return (b'1700000000000-0', fields) + + +@pytest.fixture +def source_client_mock(): + '''Mocks the source valkey.Valkey client used by FrameForwarder.''' + with patch('saeapi.frame_forwarder.valkey.Valkey') as mock_valkey: + yield mock_valkey.return_value + + +@pytest.fixture +def publisher_mock(): + '''Mocks the ValkeyPublisher and yields the publish callable.''' + with patch('saeapi.frame_forwarder.ValkeyPublisher') as mock_publisher: + yield mock_publisher.return_value.__enter__.return_value + + +def test_forward_once_forwards_latest_frame(make_config, source_client_mock, publisher_mock): + config = make_config( + instance_id='inst1', + video_source_stream_prefix='test', + frame_output_prefix='saeapi-frame', + ) + + source_client_mock.scan_iter.return_value = iter([b'test:cam1']) + source_client_mock.xrevrange.return_value = [_stream_entry(_make_frame_msg_bytes(42))] + + with FrameForwarder(config) as forwarder: + forwarder.forward_once() + + # The source stream was scanned with the configured prefix pattern + source_client_mock.scan_iter.assert_called_once_with(match='test:*', _type='STREAM') + + # A single frame-only message was published to the per-source output stream + publisher_mock.assert_called_once() + output_key, proto_data = publisher_mock.call_args.args + assert output_key == 'saeapi-frame:inst1:cam1' + + out_msg = SaeMessage() + out_msg.ParseFromString(proto_data) + assert out_msg.HasField('frame') + assert out_msg.frame.timestamp_utc_ms == 42 + # The frame image data must be forwarded untouched + assert out_msg.frame.frame_data == FRAME_DATA + assert out_msg.type == MessageType.SAE + + +def test_forward_once_handles_base64_and_plain_proto(make_config, source_client_mock, publisher_mock): + config = make_config(instance_id='inst1', video_source_stream_prefix='videosource') + + source_client_mock.scan_iter.return_value = iter([b'videosource:cam1']) + # Entry without b64 encoding, using the plain proto_data field + source_client_mock.xrevrange.return_value = [_stream_entry(_make_frame_msg_bytes(7), b64=False)] + + with FrameForwarder(config) as forwarder: + forwarder.forward_once() + + publisher_mock.assert_called_once() + _, proto_data = publisher_mock.call_args.args + out_msg = SaeMessage() + out_msg.ParseFromString(proto_data) + assert out_msg.frame.timestamp_utc_ms == 7 diff --git a/tests/test_imports.py b/tests/test_imports.py index c10cdce..c2b996a 100644 --- a/tests/test_imports.py +++ b/tests/test_imports.py @@ -1,9 +1,17 @@ import pytest -def test_saeapi_import(): + +def test_modules_import(): + '''Smoke test that the top-level components import cleanly.''' try: - from saeapi.saeapi import SaeApi + from saeapi.config import SaeApiConfig + from saeapi.event_reporter import EventReporter + from saeapi.frame_forwarder import FrameForwarder + from saeapi.stage import run_stage except ImportError as e: - pytest.fail(f"Failed to import SaeApi: {e}") + pytest.fail(f'Failed to import saeapi components: {e}') - assert SaeApi is not None, "SaeApi should be imported successfully" \ No newline at end of file + assert SaeApiConfig is not None + assert EventReporter is not None + assert FrameForwarder is not None + assert run_stage is not None diff --git a/tests/test_stage.py b/tests/test_stage.py index f8d6b05..2eddca0 100644 --- a/tests/test_stage.py +++ b/tests/test_stage.py @@ -1,12 +1,49 @@ +import threading from unittest.mock import patch +import pybase64 import pytest from visionapi.common_pb2 import MessageType -from visionapi.sae_pb2 import SaeMessage +from visionapi.sae_pb2 import EventMessage, SaeMessage -from saeapi.config import SaeApiConfig, RedisConfig from saeapi.stage import run_stage +# Real Event class captured before any patching, so the helper below can use a +# genuine event internally even while saeapi.stage.threading.Event is patched. +_RealEvent = threading.Event + + +class _OneShotStopEvent: + '''Drop-in replacement for the threading.Event used by run_stage that lets a + smoke test run exactly one frame-forwarding cycle and then shut down cleanly. + + - The poll loop calls wait(timeout); the first such call sets the event and + returns True, so the loop forwards once and then breaks. + - The main thread calls wait() (no timeout) and blocks on a real Event until set.''' + + def __init__(self): + self._event = _RealEvent() + + def set(self): + self._event.set() + + def is_set(self): + return self._event.is_set() + + def wait(self, timeout=None): + if timeout is None: + return self._event.wait(timeout=5) + self.set() + return True + + +def _make_frame_msg_bytes(timestamp: int) -> bytes: + msg = SaeMessage() + msg.frame.timestamp_utc_ms = timestamp + msg.frame.source_id = 'cam1' + msg.type = MessageType.SAE + return msg.SerializeToString() + @pytest.fixture(autouse=True) def disable_prometheus(): @@ -14,49 +51,83 @@ def disable_prometheus(): with patch('saeapi.stage.start_http_server'): yield + +@pytest.fixture(autouse=True) +def one_shot_stop_event(): + # Make run_stage's blocking loop terminate after a single forward cycle + with patch('saeapi.stage.threading.Event', _OneShotStopEvent): + yield + + @pytest.fixture -def set_config(): +def stage_config(make_config): with patch('saeapi.stage.SaeApiConfig') as mock_config: - def _make_mock_config(stream_id: str, input_stream_prefix: str): - mock_config.return_value = SaeApiConfig( - log_level='WARNING', - redis=RedisConfig( - stream_id=stream_id, - input_stream_prefix=input_stream_prefix - ) - ) - yield _make_mock_config + mock_config.return_value = make_config( + instance_id='inst1', + video_source_stream_prefix='videosource', + frame_output_prefix='saeapi-frame', + event_output_prefix='saeapi-event', + poll_interval_s=0.01, + ) + yield mock_config + @pytest.fixture -def redis_publisher_mock(): - with patch('saeapi.stage.ValkeyPublisher') as mock_publisher: +def source_client_mock(): + with patch('saeapi.frame_forwarder.valkey.Valkey') as mock_valkey: + yield mock_valkey.return_value + + +@pytest.fixture +def event_publisher_mock(): + with patch('saeapi.event_reporter.ValkeyPublisher') as mock_publisher: yield mock_publisher.return_value.__enter__.return_value + @pytest.fixture -def inject_consumer_messages(): - with patch('saeapi.stage.ValkeyConsumer') as mock_consumer: - def _inject_messages(messages): - mock_consumer.return_value.__enter__.return_value.return_value.__iter__.return_value = iter(messages) - yield _inject_messages - -def test_example(set_config, redis_publisher_mock, inject_consumer_messages): - set_config(stream_id='test_stream', input_stream_prefix='test_prefix') - - inject_consumer_messages([ - ('test_prefix:test_stream', _make_sae_msg_bytes(1)), - ('test_prefix:test_stream', _make_sae_msg_bytes(2)), - ]) - - # Run the stage (this will process the injected messages) +def frame_publisher_mock(): + with patch('saeapi.frame_forwarder.ValkeyPublisher') as mock_publisher: + yield mock_publisher.return_value.__enter__.return_value + + +def test_run_stage_startup_forward_shutdown(stage_config, source_client_mock, + event_publisher_mock, frame_publisher_mock): + '''Integration smoke test: with valkey input/output mocked, run_stage should + report startup, forward one frame, and report shutdown.''' + source_client_mock.scan_iter.return_value = iter([b'videosource:cam1']) + source_client_mock.xrevrange.return_value = [ + (b'1700000000000-0', {b'proto_data_b64': pybase64.b64encode(_make_frame_msg_bytes(99))}), + ] + run_stage() - - # Verify that messages were published - assert redis_publisher_mock.call_count == 2 - assert redis_publisher_mock.call_args_list[0].args == ('saeapi:test_stream', _make_sae_msg_bytes(1)) - assert redis_publisher_mock.call_args_list[1].args == ('saeapi:test_stream', _make_sae_msg_bytes(2)) - -def _make_sae_msg_bytes(timestamp: int) -> bytes: - sae_msg = SaeMessage() - sae_msg.frame.timestamp_utc_ms = timestamp - sae_msg.type = MessageType.SAE - return sae_msg.SerializeToString() \ No newline at end of file + + # A frame was forwarded to the per-source backend stream + frame_publisher_mock.assert_called_once() + frame_key, frame_proto = frame_publisher_mock.call_args.args + assert frame_key == 'saeapi-frame:inst1:cam1' + forwarded = SaeMessage() + forwarded.ParseFromString(frame_proto) + assert forwarded.frame.timestamp_utc_ms == 99 + + # Startup and shutdown events were both reported + assert event_publisher_mock.call_count == 2 + event_types = [] + for call in event_publisher_mock.call_args_list: + _, proto = call.args + msg = EventMessage() + msg.ParseFromString(proto) + assert msg.type == MessageType.SAE_EVENT + event_types.append(msg.event_type) + assert event_types == [EventMessage.EventType.STARTUP, EventMessage.EventType.SHUTDOWN] + + +def test_run_stage_no_source_streams(stage_config, source_client_mock, + event_publisher_mock, frame_publisher_mock): + '''Smoke test with no source streams: lifecycle events are still reported and + nothing is forwarded.''' + source_client_mock.scan_iter.return_value = iter([]) + + run_stage() + + frame_publisher_mock.assert_not_called() + assert event_publisher_mock.call_count == 2 From 6c9f559e5e876e7b719b2f3f8836d68f3f9107f2 Mon Sep 17 00:00:00 2001 From: flonix8 Date: Fri, 26 Jun 2026 14:08:20 +0200 Subject: [PATCH 03/11] Remove instance_id from frame_forwarder output key (aligned with existing SAE conventions) --- saeapi/frame_forwarder.py | 3 +-- tests/test_frame_forwarder.py | 2 +- tests/test_stage.py | 2 +- 3 files changed, 3 insertions(+), 4 deletions(-) diff --git a/saeapi/frame_forwarder.py b/saeapi/frame_forwarder.py index f7ef5a5..63f7a04 100644 --- a/saeapi/frame_forwarder.py +++ b/saeapi/frame_forwarder.py @@ -23,7 +23,6 @@ def __init__(self, config: SaeApiConfig) -> None: self._config = config self._scan_pattern = f'{config.frame_forwarding.video_source_stream_prefix}:*' self._output_prefix = config.frame_forwarding.output_stream_prefix - self._instance_id = config.instance_id self._source_client = None self._publish_ctx = ValkeyPublisher(config.backend_valkey.host, config.backend_valkey.port) @@ -79,7 +78,7 @@ def _forward_stream(self, stream_key: bytes): out_msg.type = MessageType.SAE source_id = stream_key.decode('utf-8').split(':', 1)[1] - output_key = f'{self._output_prefix}:{self._instance_id}:{source_id}' + output_key = f'{self._output_prefix}:{source_id}' self._publish(output_key, out_msg.SerializeToString()) FRAMES_FORWARDED.inc() diff --git a/tests/test_frame_forwarder.py b/tests/test_frame_forwarder.py index e2c2cb7..02d09e2 100644 --- a/tests/test_frame_forwarder.py +++ b/tests/test_frame_forwarder.py @@ -64,7 +64,7 @@ def test_forward_once_forwards_latest_frame(make_config, source_client_mock, pub # A single frame-only message was published to the per-source output stream publisher_mock.assert_called_once() output_key, proto_data = publisher_mock.call_args.args - assert output_key == 'saeapi-frame:inst1:cam1' + assert output_key == 'saeapi-frame:cam1' out_msg = SaeMessage() out_msg.ParseFromString(proto_data) diff --git a/tests/test_stage.py b/tests/test_stage.py index 2eddca0..9910b14 100644 --- a/tests/test_stage.py +++ b/tests/test_stage.py @@ -104,7 +104,7 @@ def test_run_stage_startup_forward_shutdown(stage_config, source_client_mock, # A frame was forwarded to the per-source backend stream frame_publisher_mock.assert_called_once() frame_key, frame_proto = frame_publisher_mock.call_args.args - assert frame_key == 'saeapi-frame:inst1:cam1' + assert frame_key == 'saeapi-frame:cam1' forwarded = SaeMessage() forwarded.ParseFromString(frame_proto) assert forwarded.frame.timestamp_utc_ms == 99 From 6ecc581f43e95cb2e7660e6fdf08e4599afaefcc Mon Sep 17 00:00:00 2001 From: flonix8 Date: Fri, 26 Jun 2026 14:21:33 +0200 Subject: [PATCH 04/11] Add readme --- README.md | 47 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/README.md b/README.md index a50507e..4bab3a2 100644 --- a/README.md +++ b/README.md @@ -1 +1,48 @@ # SAE sae-api + +The API layer between a SAE instance and the outside world (e.g. a cloud backend +system). It publishes information about the local SAE instance to a backend Valkey, +and is the place where inbound communication from the backend will be handled in the +future. + +## What it does + +- **Reports lifecycle events** — on startup and shutdown it publishes an + `EventMessage` describing the instance to the backend. +- **Forwards frames** — periodically pulls the latest frame from each local video + source stream and forwards a frame-only `SaeMessage` to the backend (one output + stream per source). + +## Interfaces + +It reads from a local **source Valkey** and writes to a remote **backend Valkey**. +Stream keys are built from the prefixes and IDs in the configuration: + +| Direction | Valkey | Stream key | Payload | +| --------- | -------- | --------------------------------------------------- | -------------- | +| Input | source | `{video_source_stream_prefix}:{source_id}` | `SaeMessage` | +| Output | backend | `{frame_forwarding.output_stream_prefix}:{source_id}` | frame-only `SaeMessage` | +| Output | backend | `{event_reporting.output_stream_prefix}:{instance_id}` | `EventMessage` | + +Source streams are discovered by scanning the source Valkey for keys matching +`{video_source_stream_prefix}:*`. + +## Running it + +Configuration is read from `settings.yaml` (see `settings.template.yaml` for all +options); every value can be overridden via environment variables, using `__` as the +nesting delimiter (e.g. `BACKEND_VALKEY__HOST`). Point `SETTINGS_FILE` at a different +file to use another location. + +```sh +poetry install +poetry run python main.py +``` + +A Prometheus metrics endpoint is served on `prometheus_port` (8000 by default). + +## Tests + +```sh +poetry run pytest +``` From 9024bf5338d6e099a691cab1afff13a21ace3266 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Anett=20H=C3=BCbner?= Date: Wed, 1 Jul 2026 14:03:34 +0200 Subject: [PATCH 05/11] use newer image for pr-build --- .github/workflows/pr_build.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/pr_build.yaml b/.github/workflows/pr_build.yaml index 9ef3e42..b4edcda 100644 --- a/.github/workflows/pr_build.yaml +++ b/.github/workflows/pr_build.yaml @@ -10,7 +10,7 @@ jobs: name: PR Build with Poetry runs-on: [self-hosted, linux, X64] container: - image: starwitorg/base-python-image:0.0.15 + image: starwitorg/base-python-image:3.13.2-py3.13-ptr2.3.4 volumes: - /home/githubrunner/.cache/pypoetry:/root/.cache/pypoetry steps: From a23285c8fa3c8b8f747c7fae5d6bd2784642f076 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Anett=20H=C3=BCbner?= Date: Wed, 1 Jul 2026 14:13:11 +0200 Subject: [PATCH 06/11] used latest checkout action for pr build --- .github/workflows/pr_build.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/pr_build.yaml b/.github/workflows/pr_build.yaml index b4edcda..776cffe 100644 --- a/.github/workflows/pr_build.yaml +++ b/.github/workflows/pr_build.yaml @@ -18,7 +18,7 @@ jobs: # check-out repo and set-up python #---------------------------------------------- - name: Check out repository - uses: actions/checkout@v4 + uses: actions/checkout@v5 - name: Setup run: | From 945a6a80bc52aca6ecc1f3a865f75a10374a7daf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Anett=20H=C3=BCbner?= Date: Wed, 1 Jul 2026 14:14:58 +0200 Subject: [PATCH 07/11] used checkout action 7 for pr build --- .github/workflows/pr_build.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/pr_build.yaml b/.github/workflows/pr_build.yaml index 776cffe..53ed7c3 100644 --- a/.github/workflows/pr_build.yaml +++ b/.github/workflows/pr_build.yaml @@ -18,7 +18,7 @@ jobs: # check-out repo and set-up python #---------------------------------------------- - name: Check out repository - uses: actions/checkout@v5 + uses: actions/checkout@v7 - name: Setup run: | From 952d292a42ba76e51f6cfb5fa0eb774d054d971d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Anett=20H=C3=BCbner?= Date: Wed, 1 Jul 2026 14:21:01 +0200 Subject: [PATCH 08/11] used latest versions for release build --- .github/workflows/build-publish-latest.yaml | 8 ++++---- .github/workflows/create-release.yaml | 12 ++++++------ 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/.github/workflows/build-publish-latest.yaml b/.github/workflows/build-publish-latest.yaml index 8c6f6a4..9c78472 100644 --- a/.github/workflows/build-publish-latest.yaml +++ b/.github/workflows/build-publish-latest.yaml @@ -17,23 +17,23 @@ jobs: steps: - name: checkout - uses: actions/checkout@v4 + uses: actions/checkout@v7 - name: Login to Docker Hub - uses: docker/login-action@v3 + uses: docker/login-action@v4 with: username: ${{ env.DOCKERHUB_USERNAME }} password: ${{ secrets.DOCKERHUB_TOKEN }} - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v3 + uses: docker/setup-buildx-action@v4 - name: build and push docker run: | echo "DOCKER_VERSION=$(cat pyproject.toml | grep "version" | head -1 | cut -d'=' -f2 | xargs echo -n)" >> $GITHUB_ENV - name: Build and push - uses: docker/build-push-action@v5 + uses: docker/build-push-action@v7 with: context: . file: ./Dockerfile diff --git a/.github/workflows/create-release.yaml b/.github/workflows/create-release.yaml index 07539c0..045303a 100644 --- a/.github/workflows/create-release.yaml +++ b/.github/workflows/create-release.yaml @@ -25,7 +25,7 @@ jobs: name: create new release runs-on: [self-hosted, linux, X64] container: - image: starwitorg/debian-packaging:0.0.3 + image: starwitorg/debian-packaging:0.0.6 volumes: - /home/githubrunner/.cache/pypoetry:/root/.cache/pypoetry env: @@ -37,7 +37,7 @@ jobs: # check-out repo and set-up python #---------------------------------------------- - name: Check out repository - uses: actions/checkout@v4 + uses: actions/checkout@v7 - name: Install GH CLI uses: dev-hanz-ops/install-gh-cli-action@v0.2.1 @@ -78,7 +78,7 @@ jobs: steps: - name: Checkout - uses: actions/checkout@v4 + uses: actions/checkout@v7 with: fetch-depth: 0 @@ -91,16 +91,16 @@ jobs: echo "APP_VERSION=${APP_VERSION}" >> $GITHUB_ENV - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v3 + uses: docker/setup-buildx-action@v4 - name: Login to Docker Hub - uses: docker/login-action@v3 + uses: docker/login-action@v4 with: username: ${{ env.DOCKERHUB_USERNAME }} password: ${{ secrets.DOCKERHUB_TOKEN }} - name: Build and push - uses: docker/build-push-action@v5 + uses: docker/build-push-action@v7 if: env.APP_VERSION == env.LATEST_TAG with: context: . From 7750682d02f6083434b69d8748b975a9188d70f3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Anett=20H=C3=BCbner?= Date: Wed, 1 Jul 2026 14:34:02 +0200 Subject: [PATCH 09/11] use latest python image --- .github/workflows/pr_build.yaml | 2 +- Dockerfile | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/pr_build.yaml b/.github/workflows/pr_build.yaml index 53ed7c3..bbd509b 100644 --- a/.github/workflows/pr_build.yaml +++ b/.github/workflows/pr_build.yaml @@ -10,7 +10,7 @@ jobs: name: PR Build with Poetry runs-on: [self-hosted, linux, X64] container: - image: starwitorg/base-python-image:3.13.2-py3.13-ptr2.3.4 + image: starwitorg/base-python-image:3.14.0-py3.14-ptr2.3.4 volumes: - /home/githubrunner/.cache/pypoetry:/root/.cache/pypoetry steps: diff --git a/Dockerfile b/Dockerfile index a633491..106a7a5 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,4 @@ -FROM starwitorg/base-python-image:3.13.0 AS build +FROM starwitorg/base-python-image:3.14.0 AS build # Copy only files that are necessary to install dependencies COPY poetry.lock poetry.toml pyproject.toml /code/ @@ -9,7 +9,7 @@ RUN poetry install COPY . /code/ ### Main artifact / deliverable image -FROM python:3.13-slim +FROM python:3.14-slim RUN apt update && apt install --no-install-recommends -y \ libglib2.0-0 \ libgl1 \ From 9eac856322076ef22e1cc87a95bd5e617b279ee3 Mon Sep 17 00:00:00 2001 From: Florian Stanek Date: Thu, 2 Jul 2026 17:29:12 +0200 Subject: [PATCH 10/11] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- saeapi/config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/saeapi/config.py b/saeapi/config.py index c4c3c39..dd9e462 100644 --- a/saeapi/config.py +++ b/saeapi/config.py @@ -8,7 +8,7 @@ class ValkeyConfig(BaseModel): host: str = 'localhost' - port: Annotated[int, Field(ge=1, le=65536)] = 6379 + port: Annotated[int, Field(ge=1, le=65535)] = 6379 class EventReportingConfig(BaseModel): From a72209bcdc9ca9bbf772238a9d89d59cfc2b4654 Mon Sep 17 00:00:00 2001 From: Florian Stanek Date: Thu, 2 Jul 2026 17:29:25 +0200 Subject: [PATCH 11/11] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- saeapi/config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/saeapi/config.py b/saeapi/config.py index dd9e462..cba858e 100644 --- a/saeapi/config.py +++ b/saeapi/config.py @@ -34,7 +34,7 @@ class SaeApiConfig(BaseSettings): backend_valkey: ValkeyConfig event_reporting: EventReportingConfig = EventReportingConfig() frame_forwarding: FrameForwardingConfig - prometheus_port: Annotated[int, Field(ge=1024, le=65536)] = 8000 + prometheus_port: Annotated[int, Field(ge=1024, le=65535)] = 8000 model_config = SettingsConfigDict(env_nested_delimiter='__')