matthias.lotz
19be0903d2
Implements PUSH architecture - Bridge receives config from Odoo via HTTP NEW FEATURES: - HTTP Config Server (FastAPI on port 8080) - POST /config: Receive device configuration from Odoo - GET /config: Query currently active configuration - GET /health: Health check endpoint with device statistics - Optional Bearer token authentication (BRIDGE_API_TOKEN) - Config Validation (Pydantic models) - BridgeConfig, DeviceConfig, SessionConfig schemas - Validation: unique device_id, unique mqtt_topic - Threshold validation: working_threshold_w > standby_threshold_w - Dynamic Device Management (DeviceManager) - Device diff detection (add/update/remove) - Automatic MQTT subscribe/unsubscribe for topic changes - SessionDetector lifecycle management - Active session closing on device removal - Config Persistence - Saves received config to /data/config-active.yaml - Bridge startup: loads config-active.yaml as fallback - Docker volume iot-bridge-data:/data for persistence NEW FILES: - iot_bridge/config_server.py - FastAPI HTTP server - iot_bridge/device_manager.py - Device lifecycle management - iot_bridge/tests/test-config-push.sh - Integration test script - iot_bridge/tests/test-config-push.json - Test configuration MODIFIED FILES: - iot_bridge/main.py - Integrated HTTP server in thread - iot_bridge/mqtt_client.py - Added subscribe()/unsubscribe() - iot_bridge/requirements.txt - Added fastapi, uvicorn, pydantic - iot_bridge/Dockerfile - EXPOSE 8080, HTTP healthcheck - odoo/docker-compose.dev.yaml - Port 8080, volume /data TESTING: Verified via iot_bridge/tests/test-config-push.sh: ✅ Config push via HTTP works (200 OK) ✅ Old devices removed, new devices added ✅ MQTT topics dynamically updated (subscribe/unsubscribe) ✅ Config persisted to /data/config-active.yaml ✅ Health endpoint shows last_config_update timestamp ✅ Bridge restarts load config-active.yaml automatically NEXT STEP: Phase 3.2 - Implement Odoo-side config push (Model hooks)
229 lines
9.1 KiB
Python
229 lines
9.1 KiB
Python
"""
|
|
HTTP Config Server for IoT Bridge
|
|
Receives configuration from Odoo via POST /config
|
|
"""
|
|
import os
|
|
import json
|
|
import yaml
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
from typing import List, Optional, Dict, Any
|
|
import structlog
|
|
|
|
from fastapi import FastAPI, HTTPException, Header, Depends
|
|
from pydantic import BaseModel, Field, validator
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
# ==================== Pydantic Models for Validation ====================
|
|
|
|
class SessionConfig(BaseModel):
|
|
"""Session detection configuration for a device."""
|
|
standby_threshold_w: float = Field(..., gt=0, description="Power threshold for session start (W)")
|
|
working_threshold_w: float = Field(..., gt=0, description="Power threshold for working state (W)")
|
|
start_debounce_s: float = Field(3.0, gt=0, description="Debounce time for session start (s)")
|
|
stop_debounce_s: float = Field(15.0, gt=0, description="Debounce time for session stop (s)")
|
|
message_timeout_s: float = Field(20.0, gt=0, description="Max time without messages before timeout (s)")
|
|
heartbeat_interval_s: float = Field(300.0, gt=0, description="Interval for heartbeat events (s)")
|
|
|
|
@validator('working_threshold_w')
|
|
def working_must_be_greater_than_standby(cls, v, values):
|
|
if 'standby_threshold_w' in values and v <= values['standby_threshold_w']:
|
|
raise ValueError('working_threshold_w must be greater than standby_threshold_w')
|
|
return v
|
|
|
|
|
|
class DeviceConfig(BaseModel):
|
|
"""Configuration for a single IoT device."""
|
|
device_id: str = Field(..., min_length=1, description="Unique device identifier")
|
|
machine_name: str = Field(..., min_length=1, description="Human-readable machine name")
|
|
mqtt_topic: str = Field(..., min_length=1, description="MQTT topic to subscribe to")
|
|
parser_type: str = Field("shelly_pm", description="Parser type (e.g., 'shelly_pm')")
|
|
session_config: SessionConfig
|
|
|
|
|
|
class BridgeConfig(BaseModel):
|
|
"""Complete bridge configuration from Odoo."""
|
|
devices: List[DeviceConfig] = Field(default_factory=list, description="List of devices to monitor")
|
|
timestamp: Optional[str] = Field(default_factory=lambda: datetime.utcnow().isoformat(),
|
|
description="Config timestamp")
|
|
version: Optional[str] = Field("1.0", description="Config format version")
|
|
|
|
@validator('devices')
|
|
def unique_device_ids(cls, v):
|
|
device_ids = [d.device_id for d in v]
|
|
if len(device_ids) != len(set(device_ids)):
|
|
raise ValueError('device_id must be unique')
|
|
return v
|
|
|
|
@validator('devices')
|
|
def unique_mqtt_topics(cls, v):
|
|
topics = [d.mqtt_topic for d in v]
|
|
if len(topics) != len(set(topics)):
|
|
raise ValueError('mqtt_topic must be unique')
|
|
return v
|
|
|
|
|
|
# ==================== FastAPI App ====================
|
|
|
|
class ConfigServer:
|
|
"""HTTP Server for receiving configuration from Odoo."""
|
|
|
|
def __init__(self, config_callback=None, token: Optional[str] = None):
|
|
"""
|
|
Initialize Config Server.
|
|
|
|
Args:
|
|
config_callback: Callback function(new_config) called when config is received
|
|
token: Optional Bearer token for authentication
|
|
"""
|
|
self.app = FastAPI(
|
|
title="IoT Bridge Config API",
|
|
description="Receives device configuration from Odoo",
|
|
version="1.0.0"
|
|
)
|
|
self.config_callback = config_callback
|
|
self.auth_token = token
|
|
self.current_config: Optional[BridgeConfig] = None
|
|
self.device_count = 0
|
|
self.subscription_count = 0
|
|
self.last_config_update: Optional[datetime] = None
|
|
|
|
# Register routes
|
|
self._setup_routes()
|
|
|
|
logger.info("config_server_initialized", auth_enabled=bool(token))
|
|
|
|
def _verify_token(self, authorization: Optional[str] = Header(None)):
|
|
"""Verify Bearer token if authentication is enabled."""
|
|
if not self.auth_token:
|
|
return True # No auth required
|
|
|
|
if not authorization:
|
|
raise HTTPException(status_code=401, detail="Authorization header missing")
|
|
|
|
if not authorization.startswith("Bearer "):
|
|
raise HTTPException(status_code=401, detail="Invalid authorization header format")
|
|
|
|
token = authorization.replace("Bearer ", "")
|
|
if token != self.auth_token:
|
|
raise HTTPException(status_code=403, detail="Invalid token")
|
|
|
|
return True
|
|
|
|
def _setup_routes(self):
|
|
"""Setup FastAPI routes."""
|
|
|
|
@self.app.get("/health")
|
|
async def health():
|
|
"""Health check endpoint."""
|
|
return {
|
|
"status": "ok",
|
|
"devices": self.device_count,
|
|
"subscriptions": self.subscription_count,
|
|
"last_config_update": self.last_config_update.isoformat() if self.last_config_update else None
|
|
}
|
|
|
|
@self.app.post("/config")
|
|
async def receive_config(config: BridgeConfig, authorized: bool = Depends(self._verify_token)):
|
|
"""
|
|
Receive new configuration from Odoo.
|
|
|
|
This endpoint accepts a complete device configuration and triggers:
|
|
1. Config validation
|
|
2. Device diff (added/updated/removed)
|
|
3. Dynamic MQTT subscription updates
|
|
4. Config persistence to /data/config-active.yaml
|
|
"""
|
|
try:
|
|
logger.info("config_received",
|
|
device_count=len(config.devices),
|
|
timestamp=config.timestamp)
|
|
|
|
# Persist config to disk
|
|
self._persist_config(config)
|
|
|
|
# Update internal state
|
|
self.current_config = config
|
|
self.device_count = len(config.devices)
|
|
self.subscription_count = len(config.devices) # Each device = 1 subscription
|
|
self.last_config_update = datetime.utcnow()
|
|
|
|
# Call callback to update bridge internals
|
|
if self.config_callback:
|
|
await self.config_callback(config)
|
|
|
|
logger.info("config_applied",
|
|
devices=self.device_count,
|
|
status="success")
|
|
|
|
return {
|
|
"status": "success",
|
|
"message": "Configuration applied",
|
|
"devices_configured": len(config.devices),
|
|
"timestamp": datetime.utcnow().isoformat()
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("config_apply_failed", error=str(e))
|
|
raise HTTPException(status_code=500, detail=f"Failed to apply config: {str(e)}")
|
|
|
|
@self.app.get("/config")
|
|
async def get_current_config():
|
|
"""Get currently active configuration."""
|
|
if not self.current_config:
|
|
raise HTTPException(status_code=404, detail="No configuration loaded")
|
|
|
|
return self.current_config.dict()
|
|
|
|
def _persist_config(self, config: BridgeConfig):
|
|
"""Persist configuration to /data/config-active.yaml."""
|
|
try:
|
|
# Create /data directory if it doesn't exist
|
|
data_dir = Path("/data")
|
|
data_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
config_path = data_dir / "config-active.yaml"
|
|
|
|
# Convert Pydantic model to dict
|
|
config_dict = config.dict()
|
|
|
|
# Write to YAML
|
|
with open(config_path, 'w') as f:
|
|
yaml.dump(config_dict, f, default_flow_style=False, sort_keys=False)
|
|
|
|
logger.info("config_persisted", path=str(config_path), devices=len(config.devices))
|
|
|
|
except Exception as e:
|
|
logger.error("config_persist_failed", error=str(e))
|
|
# Don't raise - persistence failure shouldn't block config application
|
|
|
|
def load_persisted_config(self) -> Optional[BridgeConfig]:
|
|
"""Load configuration from /data/config-active.yaml if it exists."""
|
|
try:
|
|
config_path = Path("/data/config-active.yaml")
|
|
|
|
if not config_path.exists():
|
|
logger.info("no_persisted_config", path=str(config_path))
|
|
return None
|
|
|
|
with open(config_path, 'r') as f:
|
|
config_dict = yaml.safe_load(f)
|
|
|
|
config = BridgeConfig(**config_dict)
|
|
|
|
self.current_config = config
|
|
self.device_count = len(config.devices)
|
|
self.subscription_count = len(config.devices)
|
|
|
|
logger.info("persisted_config_loaded",
|
|
path=str(config_path),
|
|
devices=len(config.devices))
|
|
|
|
return config
|
|
|
|
except Exception as e:
|
|
logger.error("persisted_config_load_failed", error=str(e))
|
|
return None
|