Merge pull request #4 from starwit/feature/AB#2263-register-config-message

Feature/ab#2263 register config message

Author: flonix8

.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

.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: .

.github/workflows/pr_build.yaml

@@ -10,15 +10,15 @@ 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.14.0-py3.14-ptr2.3.4
         volumes:
             - /home/githubrunner/.cache/pypoetry:/root/.cache/pypoetry
     steps:
       #----------------------------------------------
       #       check-out repo and set-up python
       #----------------------------------------------
       - name: Check out repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v7
 
       - name: Setup
         run: |

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 \

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
+```

poetry.lock

@@ -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"

pyproject.toml

@@ -6,21 +6,39 @@
 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'
+    port: Annotated[int, Field(ge=1, le=65535)] = 6379
+
+
+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
-    prometheus_port: Annotated[int, Field(ge=1024, le=65536)] = 8000
+    # 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=65535)] = 8000
 
     model_config = SettingsConfigDict(env_nested_delimiter='__')
 
     @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)
+        return (init_settings, env_settings, YamlConfigSettingsSource(settings_cls, yaml_file=YAML_LOCATION), file_secret_settings)

saeapi/config.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}')

saeapi/event_reporter.py

@@ -0,0 +1,85 @@
+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._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}:{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}')