Support output to webhook (#4598)

Author: itallix

backend/app/main.py

@@ -118,7 +118,8 @@ def main() -> None:
         app,
         host=settings.host,
         port=settings.port,
-        reload=settings.debug and settings.environment == "dev",
+        # FIXME: reload mode currently does not work with multiple workers
+        # reload=settings.debug and settings.environment == "dev",
         log_level="debug" if settings.debug else "info",
     )
 

backend/app/schemas/sink.py

@@ -2,12 +2,16 @@
 # SPDX-License-Identifier: Apache-2.0
 
 from enum import StrEnum
+from os import getenv
 from typing import Annotated, Literal
 
 from pydantic import Field, TypeAdapter
 
 from app.schemas.base import BaseIDNameModel
 
+MQTT_USERNAME = "MQTT_USERNAME"
+MQTT_PASSWORD = "MQTT_PASSWORD"  # noqa: S105
+
 
 class SinkType(StrEnum):
     DISCONNECTED = "disconnected"
@@ -67,8 +71,7 @@ class MqttSinkConfig(BaseSinkConfig):
     broker_host: str
     broker_port: int
     topic: str
-    username: str | None = None
-    password: str | None = None
+    auth_required: bool = False
 
     model_config = {
         "json_schema_extra": {
@@ -80,10 +83,24 @@ class MqttSinkConfig(BaseSinkConfig):
                 "broker_port": 1883,
                 "topic": "predictions",
                 "output_formats": ["predictions"],
+                "auth_required": True,
             }
         }
     }
 
+    def get_credentials(self) -> tuple[str | None, str | None]:
+        """Configure stream URL with authentication if required."""
+        if not self.auth_required:
+            return None, None
+
+        username = getenv(MQTT_USERNAME)
+        password = getenv(MQTT_PASSWORD)
+
+        if not username or not password:
+            raise RuntimeError("MQTT credentials not provided.")
+
+        return username, password
+
 
 class RosSinkConfig(BaseSinkConfig):
     sink_type: Literal[SinkType.ROS]
@@ -102,9 +119,16 @@ class RosSinkConfig(BaseSinkConfig):
     }
 
 
+HttpMethod = Literal["POST", "PUT", "PATCH"]
+HttpHeaders = dict[str, str]
+
+
 class WebhookSinkConfig(BaseSinkConfig):
     sink_type: Literal[SinkType.WEBHOOK]
     webhook_url: str
+    http_method: HttpMethod = "POST"
+    headers: HttpHeaders | None = None
+    timeout: int = 10  # seconds
 
     model_config = {
         "json_schema_extra": {
@@ -113,6 +137,8 @@ class WebhookSinkConfig(BaseSinkConfig):
                 "sink_type": "webhook",
                 "name": "Webhook Endpoint",
                 "webhook_url": "https://example.com/webhook",
+                "http_method": "PUT",
+                "headers": {"Authorization": "Bearer YOUR_TOKEN"},
                 "output_formats": ["predictions"],
             }
         }

backend/app/services/dispatch_service.py

@@ -4,7 +4,7 @@
 from collections.abc import Callable, Sequence
 
 from app.schemas import Sink, SinkType
-from app.services.dispatchers import Dispatcher, FolderDispatcher, MqttDispatcher
+from app.services.dispatchers import Dispatcher, FolderDispatcher, MqttDispatcher, WebhookDispatcher
 
 
 class DispatchService:
@@ -13,11 +13,11 @@ class DispatchService:
         SinkType.FOLDER: lambda config: FolderDispatcher(output_config=config),  # type: ignore[union-attr, arg-type]
         SinkType.MQTT: lambda config: MqttDispatcher(output_config=config),  # type: ignore[union-attr, arg-type]
         SinkType.ROS: lambda _: _raise_not_implemented("ROS output is not implemented yet"),
-        SinkType.WEBHOOK: lambda _: _raise_not_implemented("WEBHOOK output is not implemented yet"),
+        SinkType.WEBHOOK: lambda config: WebhookDispatcher(output_config=config),  # type: ignore[union-attr, arg-type]
     }
 
     @classmethod
-    def get_destination(cls, output_config: Sink) -> Dispatcher | None:
+    def _get_destination(cls, output_config: Sink) -> Dispatcher | None:
         # TODO handle exceptions: if some output cannot be initialized, exclude it and raise a warning
         factory = cls._dispatcher_registry.get(output_config.sink_type)
         if factory is None:
@@ -33,7 +33,7 @@ def get_destinations(cls, output_configs: Sequence[Sink]) -> list[Dispatcher]:
         Args:
             output_configs (Sequence[OutputConfig]): A sequence of output configurations.
         """
-        return [dispatcher for config in output_configs if (dispatcher := cls.get_destination(config)) is not None]
+        return [dispatcher for config in output_configs if (dispatcher := cls._get_destination(config)) is not None]
 
 
 def _raise_not_implemented(message: str) -> None:

backend/app/services/dispatchers/__init__.py

@@ -3,7 +3,8 @@
 
 from .filesystem import FolderDispatcher
 from .mqtt import MqttDispatcher
+from .webhook import WebhookDispatcher
 
-Dispatcher = FolderDispatcher | MqttDispatcher
+Dispatcher = FolderDispatcher | MqttDispatcher | WebhookDispatcher
 
-__all__ = ["Dispatcher", "FolderDispatcher", "MqttDispatcher"]
+__all__ = ["Dispatcher", "FolderDispatcher", "MqttDispatcher", "WebhookDispatcher"]

backend/app/services/dispatchers/base.py

@@ -1,13 +1,25 @@
 # Copyright (C) 2025 Intel Corporation
 # SPDX-License-Identifier: Apache-2.0
 
+import base64
 import time
 from abc import ABCMeta, abstractmethod
+from datetime import datetime
+from typing import Any
 
+import cv2
 import numpy as np
 from model_api.models.result import Result
 
-from app.schemas import Sink
+from app.schemas import OutputFormat, Sink
+
+
+def numpy_to_base64(image: np.ndarray, fmt: str = ".jpg") -> str:
+    """Convert a numpy array image to a base64 string."""
+    success, img_buf = cv2.imencode(fmt, image)
+    if success:
+        return base64.b64encode(img_buf).decode("utf-8")
+    raise ValueError(f"Failed to encode image in format {fmt}")
 
 
 class DispatchError(Exception):
@@ -51,6 +63,30 @@ def _dispatch(
             predictions (Result): Predictions generated by the model
         """
 
+    def _create_payload(
+        self, original_image: np.ndarray, image_with_visualization: np.ndarray, predictions: Result
+    ) -> dict[str, Any]:
+        """Create a JSON payload with the requested output formats."""
+        result: dict[str, Any] = {}
+        payload = {"timestamp": datetime.now().isoformat(), "result": result}
+
+        if OutputFormat.IMAGE_ORIGINAL in self.output_formats:
+            result[OutputFormat.IMAGE_ORIGINAL] = {
+                "data": numpy_to_base64(original_image),
+                "format": "jpeg",
+            }
+
+        if OutputFormat.IMAGE_WITH_PREDICTIONS in self.output_formats:
+            result[OutputFormat.IMAGE_WITH_PREDICTIONS] = {
+                "data": numpy_to_base64(image_with_visualization),
+                "format": "jpeg",
+            }
+
+        if OutputFormat.PREDICTIONS in self.output_formats:
+            result[OutputFormat.PREDICTIONS] = str(predictions)
+
+        return payload
+
     def dispatch(
         self,
         original_image: np.ndarray,

backend/app/services/dispatchers/mqtt.py

@@ -1,19 +1,16 @@
 # Copyright (C) 2025 Intel Corporation
 # SPDX-License-Identifier: Apache-2.0
 
-import base64
 import json
 import logging
 import threading
 import time
-from datetime import datetime
 from typing import Any
 
-import cv2
 import numpy as np
 from model_api.models.result import Result
 
-from app.schemas.sink import MqttSinkConfig, OutputFormat
+from app.schemas.sink import MqttSinkConfig
 from app.services.dispatchers.base import BaseDispatcher
 
 try:
@@ -28,17 +25,6 @@
 CONNECT_TIMEOUT = 10
 
 
-def _encode_image_to_base64(image: np.ndarray, fmt: str = ".jpg") -> str:
-    success, img_buf = cv2.imencode(fmt, image)
-    if success:
-        return base64.b64encode(img_buf.tobytes()).decode("utf-8")
-    raise ValueError(f"Failed to encode image in format {fmt}")
-
-
-def _create_mqtt_payload(data_type: str, **kwargs) -> dict[str, Any]:
-    return {"timestamp": datetime.now().isoformat(), "type": data_type, **kwargs}
-
-
 class MqttDispatcher(BaseDispatcher):
     def __init__(
         self,
@@ -65,8 +51,7 @@ def __init__(
         self.broker_host = output_config.broker_host
         self.broker_port = output_config.broker_port
         self.topic = output_config.topic
-        self.username = output_config.username
-        self.password = output_config.password
+        self.username, self.password = output_config.get_credentials()
 
         self._connected = False
         self._connection_lock = threading.Lock()
@@ -82,7 +67,7 @@ def _create_default_client(self) -> "mqtt.Client":
         client = mqtt.Client(client_id=client_id)
         client.on_connect = self._on_connect
         client.on_disconnect = self._on_disconnect
-        if self.username and self.password:
+        if self.username is not None and self.password is not None:
             client.username_pw_set(self.username, self.password)
         return client
 
@@ -119,54 +104,26 @@ def _on_disconnect(self, _client: "mqtt.Client", _userdata: Any, rc: int):
     def is_connected(self) -> bool:
         return self._connected
 
-    def _publish_message(self, topic: str, payload: dict[str, Any]) -> bool:
+    def __publish_message(self, topic: str, payload: dict[str, Any]) -> None:
         if not self._connected:
             logger.warning("Client not connected. Reconnecting...")
             try:
                 self._connect()
-            except Exception:
+            except ConnectionError:
                 logger.exception("Reconnect failed")
-                return False
 
         try:
             result = self.client.publish(topic, json.dumps(payload))
-            if result.rc == mqtt.MQTT_ERR_SUCCESS:
-                if self._track_messages:
-                    self._published_messages.append({"topic": topic, "payload": payload, "timestamp": datetime.now()})
-                return True
+            if result.rc == mqtt.MQTT_ERR_SUCCESS and self._track_messages:
+                self._published_messages.append({"topic": topic, "payload": payload})
             logger.error(f"Publish failed: {mqtt.error_string(result.rc)}")
-        except Exception:
-            logger.exception("Publish exception")
-        return False
-
-    def _dispatch_image(self, image: np.ndarray, data_type: str):
-        try:
-            image_b64 = _encode_image_to_base64(image)
-            payload = _create_mqtt_payload(
-                data_type=data_type,
-                image=image_b64,
-                format="jpeg",
-            )
-            self._publish_message(self.topic, payload)
-        except Exception:
-            logger.exception("Failed to dispatch %s", data_type)
-
-    def _dispatch_predictions(self, predictions: Result):
-        try:
-            payload = _create_mqtt_payload(data_type=OutputFormat.PREDICTIONS.value, predictions=str(predictions))
-            self._publish_message(self.topic, payload)
-        except Exception:
-            logger.exception("Failed to dispatch predictions")
+        except ValueError:
+            logger.exception("Invalid payload for MQTT publish")
 
     def _dispatch(self, original_image: np.ndarray, image_with_visualization: np.ndarray, predictions: Result) -> None:
-        if OutputFormat.IMAGE_ORIGINAL in self.output_formats:
-            self._dispatch_image(original_image, OutputFormat.IMAGE_ORIGINAL)
-
-        if OutputFormat.IMAGE_WITH_PREDICTIONS in self.output_formats:
-            self._dispatch_image(image_with_visualization, OutputFormat.IMAGE_WITH_PREDICTIONS)
+        payload = self._create_payload(original_image, image_with_visualization, predictions)
 
-        if OutputFormat.PREDICTIONS in self.output_formats:
-            self._dispatch_predictions(predictions)
+        self.__publish_message(self.topic, payload)
 
     def get_published_messages(self) -> list:
         return self._published_messages.copy()
@@ -175,11 +132,11 @@ def clear_published_messages(self) -> None:
         self._published_messages.clear()
 
     def close(self) -> None:
-        try:
-            self.client.loop_stop()
-            self.client.disconnect()
-        except Exception:
-            logger.exception("Error closing dispatcher")
-        finally:
-            self._connected = False
-            self._connection_event.clear()
+        err = self.client.loop_stop()
+        if err != mqtt.MQTT_ERR_SUCCESS:
+            logger.warning(f"Error stopping MQTT loop: {mqtt.error_string(err)}")
+        err = self.client.disconnect()
+        if err != mqtt.MQTT_ERR_SUCCESS:
+            logger.warning(f"Error disconnecting MQTT client: {mqtt.error_string(err)}")
+        self._connected = False
+        self._connection_event.clear()

backend/app/services/dispatchers/webhook.py

@@ -0,0 +1,64 @@
+# Copyright (C) 2025 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+"""This module contains the WebhookDispatcher class for dispatching images and predictions to a webhook endpoint."""
+
+import logging
+from typing import Any
+
+import numpy as np
+import requests
+from model_api.models.result import Result
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+
+from app.schemas.sink import WebhookSinkConfig
+from app.services.dispatchers.base import BaseDispatcher
+
+logger = logging.getLogger(__name__)
+
+MAX_RETRIES = 3
+BACKOFF_FACTOR = 0.3
+RETRY_ON_STATUS = [500, 502, 503, 504]
+
+
+class WebhookDispatcher(BaseDispatcher):
+    def __init__(self, output_config: WebhookSinkConfig) -> None:
+        """
+        Initialize the WebhookDispatcher.
+        Args:
+            output_config: Configuration for the webhook-based output destination
+        """
+        super().__init__(output_config=output_config)
+        self.webhook_url = output_config.webhook_url
+        self.http_method = output_config.http_method
+        self.headers = output_config.headers
+        self.timeout = output_config.timeout
+        self.session = requests.Session()
+        retries = Retry(
+            total=MAX_RETRIES,
+            backoff_factor=BACKOFF_FACTOR,
+            status_forcelist=RETRY_ON_STATUS,
+            allowed_methods=["PATCH", "POST", "PUT"],
+        )
+        adapter = HTTPAdapter(max_retries=retries)
+        self.session.mount("http://", adapter)
+        self.session.mount("https://", adapter)
+
+    def __send_to_webhook(self, payload: dict[str, Any]) -> None:
+        logger.debug("Sending payload to webhook at %s", self.webhook_url)
+        response = self.session.request(
+            self.http_method, self.webhook_url, headers=self.headers, json=payload, timeout=self.timeout
+        )
+        response.raise_for_status()
+        logger.debug("Response from webhook: %s", response.text)
+
+    def _dispatch(
+        self,
+        original_image: np.ndarray,
+        image_with_visualization: np.ndarray,
+        predictions: Result,
+    ) -> None:
+        payload = self._create_payload(original_image, image_with_visualization, predictions)
+
+        self.__send_to_webhook(payload)