initial import

Author: beatfactor

.env.example

@@ -0,0 +1,25 @@
+# IoT Edge runtime (set automatically when running as Edge module)
+# LOCAL_ENV=true
+# DEVICE_CONNECTION_STRING=HostName=...;DeviceId=...;SharedAccessKey=...
+
+# Storage
+STORAGE_BACKEND=local
+OUTPUT_BASE_PATH=./output
+
+# Azure Blob Edge (only when STORAGE_BACKEND=azure-blob-edge)
+# AZURE_STORAGE_CONNECTION_STRING=DefaultEndpointsProtocol=http;...
+
+# Stream input
+STREAM_HOST=0.0.0.0
+STREAM_PORT=9100
+
+# File watcher
+WATCH_DIR=/data/sensor
+
+# Metadata
+CAMPAIGN_ID=
+PLATFORM_ID=
+PLATFORM_NAME=
+
+# Logging
+LOG_LEVEL=INFO

.gitignore

@@ -0,0 +1,15 @@
+__pycache__/
+*.pyc
+*.pyo
+.env
+.venv/
+venv/
+output/
+*.egg-info/
+dist/
+build/
+.pytest_cache/
+.mypy_cache/
+test/.test_data_cache/
+test/output_azure/
+test/test_data/

Dockerfile.Linux.amd64

@@ -0,0 +1,31 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+# System dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    git build-essential && \
+    rm -rf /var/lib/apt/lists/*
+
+# Install oceanstream from git
+ARG OCEANSTREAM_BRANCH=main
+ARG OCEANSTREAM_REPO=https://github.com/OceanStreamIO/sd-data-ingest.git
+RUN git clone --depth 1 -b ${OCEANSTREAM_BRANCH} ${OCEANSTREAM_REPO} /tmp/oceanstream && \
+    pip install --no-cache-dir /tmp/oceanstream && \
+    rm -rf /tmp/oceanstream
+
+# Python dependencies
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Application code
+COPY azure_handler/ azure_handler/
+COPY ingest/ ingest/
+COPY process/ process/
+COPY exports/ exports/
+COPY simulate/ simulate/
+COPY config.py main.py standalone.py ./
+
+ENV PYTHONUNBUFFERED=1
+
+CMD ["python", "-u", "main.py"]

Dockerfile.Linux.arm64

@@ -0,0 +1,31 @@
+FROM python:3.12-slim-bookworm
+
+WORKDIR /app
+
+# System dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    git build-essential gcc g++ && \
+    rm -rf /var/lib/apt/lists/*
+
+# Install oceanstream from git
+ARG OCEANSTREAM_BRANCH=main
+ARG OCEANSTREAM_REPO=https://github.com/OceanStreamIO/sd-data-ingest.git
+RUN git clone --depth 1 -b ${OCEANSTREAM_BRANCH} ${OCEANSTREAM_REPO} /tmp/oceanstream && \
+    pip install --no-cache-dir /tmp/oceanstream && \
+    rm -rf /tmp/oceanstream
+
+# Python dependencies
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Application code
+COPY azure_handler/ azure_handler/
+COPY ingest/ ingest/
+COPY process/ process/
+COPY exports/ exports/
+COPY simulate/ simulate/
+COPY config.py main.py standalone.py ./
+
+ENV PYTHONUNBUFFERED=1
+
+CMD ["python", "-u", "main.py"]

azure_handler/__init__.py

@@ -0,0 +1,26 @@
+"""Azure IoT Hub handler — client, messaging, storage."""
+
+from __future__ import annotations
+
+
+def create_client():
+    """Lazy import to avoid pulling azure-iot-device at module level."""
+    from azure_handler.connect_iothub import create_client as _create
+    return _create()
+
+
+def create_storage(backend: str = "local", base_path: str = "/app/processed"):
+    """Factory for storage backends."""
+    from azure_handler.storage import LocalStorage, AzureBlobEdgeStorage
+
+    if backend == "azure-blob-edge":
+        try:
+            return AzureBlobEdgeStorage()
+        except Exception:
+            import logging
+            logging.getLogger("sensorstream").warning(
+                "Azure Blob Edge unavailable, falling back to local storage"
+            )
+            return LocalStorage(base_path)
+
+    return LocalStorage(base_path)

azure_handler/connect_iothub.py

@@ -0,0 +1,46 @@
+"""IoT Hub module client creation for the sensorstream module."""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Optional
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+logger = logging.getLogger("sensorstream")
+
+DEVICE_CONNECTION_STRING: Optional[str] = os.getenv("DEVICE_CONNECTION_STRING")
+LOCAL_ENV: bool = os.getenv("LOCAL_ENV", "").lower() in ("true", "1", "yes")
+
+
+def create_client():
+    """Create the IoT Hub module client.
+
+    In IoT Edge runtime: creates from edge environment.
+    In local/dev mode: creates from device connection string.
+    """
+    from azure.iot.device import IoTHubModuleClient
+
+    client = None
+
+    try:
+        if LOCAL_ENV:
+            logger.info("Running in local environment using device connection string.")
+            if not DEVICE_CONNECTION_STRING:
+                raise ValueError("Missing DEVICE_CONNECTION_STRING environment variable.")
+            client = IoTHubModuleClient.create_from_connection_string(DEVICE_CONNECTION_STRING)
+        else:
+            client = IoTHubModuleClient.create_from_edge_environment()
+
+        if client is None:
+            raise ValueError("Failed to create IoTHubModuleClient.")
+
+        logger.info("IoT Hub module client initialized")
+        return client
+
+    except Exception as e:
+        logger.error("Could not connect: %s", e, exc_info=True)
+        raise

azure_handler/message_handler.py

@@ -0,0 +1,76 @@
+"""Send messages and update twin reported properties via IoT Hub."""
+
+from __future__ import annotations
+
+import json
+import logging
+import uuid
+from datetime import datetime
+from typing import TYPE_CHECKING, Any, Dict
+
+import numpy as np
+import pandas as pd
+
+if TYPE_CHECKING:
+    from azure.iot.device import IoTHubModuleClient
+
+logger = logging.getLogger("sensorstream")
+
+
+def _default_serializer(obj):
+    """Convert non-JSON-serializable objects."""
+    if isinstance(obj, (pd.Timestamp, np.datetime64, datetime)):
+        return obj.isoformat()
+    if isinstance(obj, (np.integer,)):
+        return int(obj)
+    if isinstance(obj, (np.floating,)):
+        return float(obj)
+    if isinstance(obj, np.generic):
+        return obj.item()
+    return obj
+
+
+def serialize_for_json(obj):
+    """Recursively make an object JSON-serializable."""
+    if isinstance(obj, dict):
+        return {k: serialize_for_json(v) for k, v in obj.items()}
+    if isinstance(obj, (list, tuple)):
+        return [serialize_for_json(v) for v in obj]
+    if isinstance(obj, (pd.Timestamp, np.datetime64, datetime)):
+        return obj.isoformat()
+    if isinstance(obj, (np.int64, np.int32)):
+        return int(obj)
+    if isinstance(obj, (np.float64, np.float32)):
+        return float(obj)
+    if isinstance(obj, np.generic):
+        return obj.item()
+    if isinstance(obj, set):
+        return list(obj)
+    return obj
+
+
+def send_to_hub(
+    client: "IoTHubModuleClient",
+    data: Dict[str, Any] | None = None,
+    properties: Dict[str, Any] | None = None,
+    output_name: str = "output1",
+) -> None:
+    """Send data to Azure IoT Hub via output message or twin patch."""
+    from azure.iot.device import Message
+
+    try:
+        if properties:
+            properties = serialize_for_json(properties)
+            client.patch_twin_reported_properties(properties)
+        else:
+            if data is None:
+                data = {}
+
+            payload = json.dumps(data, default=_default_serializer)
+            message = Message(payload)
+            message.message_id = str(uuid.uuid4())
+            client.send_message_to_output(message, output_name)
+
+        logger.info("Sent data to IoT Hub on output '%s'", output_name)
+    except Exception as e:
+        logger.error("Failed to send data to IoT Hub on output '%s': %s", output_name, e, exc_info=True)