feat(bridge): Dynamic Parser Registry – Phase 1

- parsers/registry.py: PARSER_REGISTRY mit shelly_pm-Eintrag; get_parser(),
  get_schema(), list_parser_types() als Public API
- parsers/__init__.py: Registry-Funktionen exportiert
- core/device_manager.py: globalen ShellyParser entfernt; DeviceManager
  verwaltet jetzt ein eigenes parser-Dict pro Device (per get_parser())
- api/server.py: GET /parsers Endpoint hinzugefügt (gibt get_schema() zurück)
- tests/unit/test_parser_registry.py: 17 neue Tests (Registry-API,
  PARSER_REGISTRY-Integrität, DeviceManager-Integration)
- tests/unit/test_device_manager.py: Test auf neues API angepasst (patch
  statt parser=-Argument)

Tests: 63/63 passed
This commit is contained in:
Matthias Lotz 2026-03-10 17:31:03 +01:00
parent 4de3401416
commit ddd1e05f55
6 changed files with 374 additions and 15 deletions

View File

@ -214,6 +214,22 @@ class ConfigServer:
# Return config dict with all fields (including None values)
return self.current_config.dict(exclude_none=False)
@self.app.get("/parsers")
async def get_parsers():
"""
Return available parser types and their schemas.
Odoo uses this endpoint to derive the parser_type Selection field
and conditional form groups for device configuration.
Returns:
Dict mapping parser_type key schema
(label, description, topic_hint, data_unit, parameters).
"""
from parsers.registry import get_schema
return get_schema()
def _persist_config(self, config: BridgeConfig):
"""Persist configuration to /data/config-active.yaml."""
try:

View File

@ -8,7 +8,7 @@ from datetime import datetime
import structlog
from core.session_detector import SessionDetector
from parsers.shelly_parser import ShellyParser
from parsers.registry import get_parser
logger = structlog.get_logger()
@ -25,24 +25,23 @@ class DeviceManager:
- Tracking device online/offline status
"""
def __init__(self, mqtt_client, event_queue, parser=None, status_monitor=None):
def __init__(self, mqtt_client, event_queue, status_monitor=None):
"""
Initialize Device Manager.
Args:
mqtt_client: MQTTClient instance for managing subscriptions
event_queue: EventQueue instance for event handling
parser: Message parser (default: ShellyParser)
status_monitor: DeviceStatusMonitor instance for tracking online/offline
"""
self.mqtt_client = mqtt_client
self.event_queue = event_queue
self.parser = parser or ShellyParser()
self.status_monitor = status_monitor
# Device tracking
self.session_detectors: dict[str, SessionDetector] = {}
self.device_map: dict[str, str] = {} # topic -> device_id
self.parsers: dict[str, object] = {} # device_id -> parser instance
logger.info("device_manager_initialized")
@ -129,6 +128,7 @@ class DeviceManager:
self.session_detectors[device_id] = detector
self.device_map[device.mqtt_topic] = device_id
self.parsers[device_id] = get_parser(device.parser_type)
# Subscribe to MQTT topic
self.mqtt_client.subscribe(device.mqtt_topic)
@ -224,8 +224,9 @@ class DeviceManager:
self.mqtt_client.unsubscribe(topic_to_remove)
del self.device_map[topic_to_remove]
# Remove detector
# Remove detector and parser
del self.session_detectors[device_id]
self.parsers.pop(device_id, None)
logger.info("device_removed", device_id=device_id, topic=topic_to_remove)
@ -257,8 +258,12 @@ class DeviceManager:
if self.status_monitor:
self.status_monitor.update_last_seen(device_id)
# Parse message
parsed = self.parser.parse_message(topic, payload)
# Parse message using per-device parser
parser = self.parsers.get(device_id)
if parser is None:
logger.warning("no_parser_for_device", device_id=device_id)
return
parsed = parser.parse_message(topic, payload)
if parsed and parsed.get("apower") is not None:
power_w = parsed["apower"]

View File

@ -1,10 +1,14 @@
"""Parser modules for MQTT message parsing."""
from parsers.base import BaseParser, MessageParser
from parsers.registry import get_parser, get_schema, list_parser_types
from parsers.shelly_parser import ShellyParser
__all__ = [
"MessageParser",
"BaseParser",
"ShellyParser",
"get_parser",
"get_schema",
"list_parser_types",
]

View File

@ -0,0 +1,117 @@
"""
Parser Registry Single Source of Truth for available parser types.
Each entry defines:
- class: The parser class to instantiate
- label: Human-readable name (used by Odoo UI)
- description: Short description of the parser
- topic_hint: MQTT topic pattern template (informational)
- data_unit: Measurement unit this parser provides (e.g. "W" for Watts)
- parameters: Parser-specific session-detection parameters for Odoo UI
Odoo derives its ``parser_type`` Selection field and conditional form groups
from this schema via ``GET /parsers``.
"""
from parsers.shelly_parser import ShellyParser
# ─────────────────────────────────────────────────────────────────────────────
# Registry
# ─────────────────────────────────────────────────────────────────────────────
PARSER_REGISTRY: dict[str, dict] = {
"shelly_pm": {
"class": ShellyParser,
"label": "Shelly PM (Power Meter)",
"description": (
"Shelly PM Mini G3 liest Leistungsmesswerte (apower in W) "
"aus MQTT-Status-Nachrichten."
),
"topic_hint": "<device-name>/status/pm1:0",
"data_unit": "W",
"parameters": [
{
"name": "standby_threshold_w",
"label": "Standby-Schwellwert (W)",
"type": "float",
"default": 20.0,
"description": "Leistung unterhalb dieser Schwelle → keine aktive Session",
},
{
"name": "working_threshold_w",
"label": "Arbeitsschwellwert (W)",
"type": "float",
"default": 100.0,
"description": "Leistung ab dieser Schwelle → Session als aktiv gewertet",
},
{
"name": "start_debounce_s",
"label": "Start-Entprellung (s)",
"type": "float",
"default": 3.0,
"description": "Wartezeit bis Session-Start bestätigt wird",
},
{
"name": "stop_debounce_s",
"label": "Stop-Entprellung (s)",
"type": "float",
"default": 15.0,
"description": "Wartezeit bis Session-Ende bestätigt wird",
},
],
},
}
# ─────────────────────────────────────────────────────────────────────────────
# Public API
# ─────────────────────────────────────────────────────────────────────────────
def get_parser(parser_type: str):
"""
Return a new parser instance for the given parser_type.
Args:
parser_type: Key from PARSER_REGISTRY (e.g. ``"shelly_pm"``).
Returns:
Fresh parser instance implementing ``parse_message(topic, payload)``.
Raises:
KeyError: If parser_type is not registered.
"""
entry = PARSER_REGISTRY.get(parser_type)
if entry is None:
available = list(PARSER_REGISTRY.keys())
raise KeyError(
f"Unknown parser type: {parser_type!r}. Available: {available}"
)
return entry["class"]()
def get_schema() -> dict:
"""
Return the public schema of all registered parsers.
Used by ``GET /parsers`` so Odoo can derive its UI fields statically.
Returns:
Dict mapping parser_type schema dict (label, description,
topic_hint, data_unit, parameters). ``class`` is excluded.
"""
return {
key: {
"label": entry["label"],
"description": entry["description"],
"topic_hint": entry.get("topic_hint", ""),
"data_unit": entry.get("data_unit", ""),
"parameters": entry.get("parameters", []),
}
for key, entry in PARSER_REGISTRY.items()
}
def list_parser_types() -> list[str]:
"""Return sorted list of registered parser type keys."""
return sorted(PARSER_REGISTRY.keys())

View File

@ -1,6 +1,6 @@
"""Unit tests for DeviceManager lifecycle operations."""
from unittest.mock import Mock
from unittest.mock import Mock, patch
from api.models import BridgeConfig, DeviceConfig, SessionConfig
from core.device_manager import DeviceManager
@ -60,21 +60,25 @@ def test_route_message_updates_status_monitor_and_processes_power() -> None:
event_queue = Mock()
status_monitor = Mock()
parser = Mock()
parser.parse_message.return_value = {"apower": 123.4}
mock_parser = Mock()
mock_parser.parse_message.return_value = {"apower": 123.4}
manager = DeviceManager(
mqtt_client=mqtt_client,
event_queue=event_queue,
parser=parser,
status_monitor=status_monitor,
)
manager.apply_config(
BridgeConfig(mqtt=None, devices=[_device_config("dev-1", "dev-1/status/pm1:0")], version="1.0")
)
with patch("core.device_manager.get_parser", return_value=mock_parser):
manager.apply_config(
BridgeConfig(
mqtt=None,
devices=[_device_config("dev-1", "dev-1/status/pm1:0")],
version="1.0",
)
)
manager.route_message("dev-1/status/pm1:0", {"apower": 123.4})
status_monitor.update_last_seen.assert_called_once_with("dev-1")
parser.parse_message.assert_called_once()
mock_parser.parse_message.assert_called_once()

View File

@ -0,0 +1,213 @@
"""Unit tests for the Parser Registry."""
import pytest
from parsers.registry import PARSER_REGISTRY, get_parser, get_schema, list_parser_types
from parsers.shelly_parser import ShellyParser
# ─────────────────────────────────────────────────────────────────────────────
# get_parser
# ─────────────────────────────────────────────────────────────────────────────
def test_get_parser_shelly_pm_returns_shelly_parser_instance() -> None:
parser = get_parser("shelly_pm")
assert isinstance(parser, ShellyParser)
def test_get_parser_returns_new_instance_each_call() -> None:
p1 = get_parser("shelly_pm")
p2 = get_parser("shelly_pm")
assert p1 is not p2
def test_get_parser_unknown_type_raises_key_error() -> None:
with pytest.raises(KeyError, match="Unknown parser type"):
get_parser("does_not_exist")
def test_get_parser_error_message_contains_available_types() -> None:
with pytest.raises(KeyError) as exc_info:
get_parser("unknown_xyz")
assert "shelly_pm" in str(exc_info.value)
# ─────────────────────────────────────────────────────────────────────────────
# get_schema
# ─────────────────────────────────────────────────────────────────────────────
def test_get_schema_returns_dict() -> None:
schema = get_schema()
assert isinstance(schema, dict)
def test_get_schema_contains_shelly_pm() -> None:
schema = get_schema()
assert "shelly_pm" in schema
def test_get_schema_shelly_pm_has_required_fields() -> None:
entry = get_schema()["shelly_pm"]
assert "label" in entry
assert "description" in entry
assert "topic_hint" in entry
assert "data_unit" in entry
assert "parameters" in entry
def test_get_schema_does_not_expose_class() -> None:
"""The 'class' key must not leak into the public schema."""
schema = get_schema()
for entry in schema.values():
assert "class" not in entry
def test_get_schema_parameters_are_list() -> None:
schema = get_schema()
for entry in schema.values():
assert isinstance(entry["parameters"], list)
def test_get_schema_shelly_pm_parameters_have_required_keys() -> None:
params = get_schema()["shelly_pm"]["parameters"]
assert len(params) > 0
for param in params:
assert "name" in param
assert "label" in param
assert "type" in param
assert "default" in param
# ─────────────────────────────────────────────────────────────────────────────
# list_parser_types
# ─────────────────────────────────────────────────────────────────────────────
def test_list_parser_types_returns_sorted_list() -> None:
types = list_parser_types()
assert isinstance(types, list)
assert types == sorted(types)
def test_list_parser_types_contains_shelly_pm() -> None:
assert "shelly_pm" in list_parser_types()
# ─────────────────────────────────────────────────────────────────────────────
# PARSER_REGISTRY integrity
# ─────────────────────────────────────────────────────────────────────────────
def test_registry_all_entries_have_class() -> None:
for key, entry in PARSER_REGISTRY.items():
assert "class" in entry, f"Entry {key!r} missing 'class'"
def test_registry_all_classes_have_parse_message() -> None:
for key, entry in PARSER_REGISTRY.items():
cls = entry["class"]
assert hasattr(cls, "parse_message"), (
f"Parser class for {key!r} missing parse_message()"
)
def test_registry_all_classes_are_instantiable() -> None:
for key, entry in PARSER_REGISTRY.items():
cls = entry["class"]
instance = cls()
assert instance is not None
# ─────────────────────────────────────────────────────────────────────────────
# Integration: DeviceManager uses registry per device
# ─────────────────────────────────────────────────────────────────────────────
def test_device_manager_creates_per_device_parser() -> None:
"""DeviceManager must store an independent parser per device_id."""
from unittest.mock import Mock
from api.models import BridgeConfig, DeviceConfig, SessionConfig
from core.device_manager import DeviceManager
mqtt_client = Mock()
event_queue = Mock()
manager = DeviceManager(mqtt_client=mqtt_client, event_queue=event_queue)
config = BridgeConfig(
mqtt=None,
devices=[
DeviceConfig(
device_id="dev-a",
machine_name="Machine A",
mqtt_topic="dev-a/status/pm1:0",
parser_type="shelly_pm",
device_status_timeout_s=120,
session_config=SessionConfig(
standby_threshold_w=20,
working_threshold_w=100,
start_debounce_s=3,
stop_debounce_s=15,
message_timeout_s=20,
heartbeat_interval_s=60,
),
),
DeviceConfig(
device_id="dev-b",
machine_name="Machine B",
mqtt_topic="dev-b/status/pm1:0",
parser_type="shelly_pm",
device_status_timeout_s=120,
session_config=SessionConfig(
standby_threshold_w=20,
working_threshold_w=100,
start_debounce_s=3,
stop_debounce_s=15,
message_timeout_s=20,
heartbeat_interval_s=60,
),
),
],
version="1.0",
)
manager.apply_config(config)
assert "dev-a" in manager.parsers
assert "dev-b" in manager.parsers
# Each device gets its own parser instance
assert manager.parsers["dev-a"] is not manager.parsers["dev-b"]
def test_device_manager_removes_parser_on_device_removal() -> None:
from unittest.mock import Mock
from api.models import BridgeConfig, DeviceConfig, SessionConfig
from core.device_manager import DeviceManager
mqtt_client = Mock()
event_queue = Mock()
manager = DeviceManager(mqtt_client=mqtt_client, event_queue=event_queue)
device = DeviceConfig(
device_id="dev-x",
machine_name="Machine X",
mqtt_topic="dev-x/status/pm1:0",
parser_type="shelly_pm",
device_status_timeout_s=120,
session_config=SessionConfig(
standby_threshold_w=20,
working_threshold_w=100,
start_debounce_s=3,
stop_debounce_s=15,
message_timeout_s=20,
heartbeat_interval_s=60,
),
)
manager.apply_config(BridgeConfig(mqtt=None, devices=[device], version="1.0"))
assert "dev-x" in manager.parsers
manager.apply_config(BridgeConfig(mqtt=None, devices=[], version="1.0"))
assert "dev-x" not in manager.parsers