228 lines
7.9 KiB
Python
228 lines
7.9 KiB
Python
"""Vendor-neutral Bridge API v1 models."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, replace
|
|
from typing import Any
|
|
|
|
|
|
class BridgeDataError(ValueError):
|
|
"""Raised when a bridge response violates API v1."""
|
|
|
|
|
|
def _required_string(data: dict[str, Any], key: str) -> str:
|
|
value = data.get(key)
|
|
if not isinstance(value, str) or not value:
|
|
raise BridgeDataError(f"{key} must be a non-empty string")
|
|
return value
|
|
|
|
|
|
def _string(data: dict[str, Any], key: str) -> str:
|
|
value = data.get(key)
|
|
if not isinstance(value, str):
|
|
raise BridgeDataError(f"{key} must be a string")
|
|
return value
|
|
|
|
|
|
def _optional_string(data: dict[str, Any], key: str) -> str | None:
|
|
value = data.get(key)
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, str) or not value:
|
|
raise BridgeDataError(f"{key} must be null or a non-empty string")
|
|
return value
|
|
|
|
|
|
def _capabilities(data: dict[str, Any]) -> frozenset[str]:
|
|
value = data.get("capabilities")
|
|
if not isinstance(value, list) or not all(isinstance(item, str) for item in value):
|
|
raise BridgeDataError("capabilities must be a list of strings")
|
|
return frozenset(value)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class BridgeInfo:
|
|
"""Bridge metadata."""
|
|
|
|
bridge_id: str | None
|
|
bridge_version: str
|
|
api_version: int
|
|
connected: bool
|
|
capabilities: frozenset[str]
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> BridgeInfo:
|
|
"""Parse bridge metadata while tolerating additive fields."""
|
|
api_version = data.get("api_version")
|
|
connected = data.get("connected")
|
|
if not isinstance(api_version, int) or isinstance(api_version, bool):
|
|
raise BridgeDataError("api_version must be an integer")
|
|
if not isinstance(connected, bool):
|
|
raise BridgeDataError("connected must be a boolean")
|
|
return cls(
|
|
bridge_id=_optional_string(data, "bridge_id"),
|
|
bridge_version=_required_string(data, "bridge_version"),
|
|
api_version=api_version,
|
|
connected=connected,
|
|
capabilities=_capabilities(data),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class BridgeObject:
|
|
"""Normalized object reported by the bridge."""
|
|
|
|
id: str
|
|
name: str
|
|
state: str
|
|
capabilities: frozenset[str]
|
|
controller_id: str | None = None
|
|
area_id: str | None = None
|
|
module_id: str | None = None
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> BridgeObject:
|
|
"""Parse a normalized object while tolerating additive fields."""
|
|
return cls(
|
|
id=_required_string(data, "id"),
|
|
name=_string(data, "name"),
|
|
state=_string(data, "state"),
|
|
capabilities=_capabilities(data),
|
|
controller_id=_optional_string(data, "controller_id"),
|
|
area_id=_optional_string(data, "area_id"),
|
|
module_id=_optional_string(data, "module_id"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class DomainEvent:
|
|
"""Normalized state event."""
|
|
|
|
revision: int
|
|
event_type: str
|
|
object_type: str
|
|
object_id: str
|
|
state: str
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> DomainEvent:
|
|
"""Parse a normalized event."""
|
|
revision = data.get("revision")
|
|
if not isinstance(revision, int) or isinstance(revision, bool) or revision < 0:
|
|
raise BridgeDataError("revision must be a non-negative integer")
|
|
return cls(
|
|
revision=revision,
|
|
event_type=_required_string(data, "event_type"),
|
|
object_type=_required_string(data, "object_type"),
|
|
object_id=_required_string(data, "object_id"),
|
|
state=_required_string(data, "state"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CommandResult:
|
|
"""Result of a normalized bridge command."""
|
|
|
|
accepted: bool
|
|
command_id: str
|
|
message: str | None
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> CommandResult:
|
|
"""Parse a command result."""
|
|
accepted = data.get("accepted")
|
|
if not isinstance(accepted, bool):
|
|
raise BridgeDataError("accepted must be a boolean")
|
|
return cls(
|
|
accepted=accepted,
|
|
command_id=_required_string(data, "command_id"),
|
|
message=_optional_string(data, "message"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class BridgeSnapshot:
|
|
"""Complete normalized bridge state."""
|
|
|
|
revision: int
|
|
controllers: dict[str, BridgeObject]
|
|
areas: dict[str, BridgeObject]
|
|
points: dict[str, BridgeObject]
|
|
modules: dict[str, BridgeObject]
|
|
outputs: dict[str, BridgeObject]
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> BridgeSnapshot:
|
|
"""Parse and validate a full snapshot."""
|
|
revision = data.get("revision")
|
|
if not isinstance(revision, int) or isinstance(revision, bool) or revision < 0:
|
|
raise BridgeDataError("revision must be a non-negative integer")
|
|
|
|
def parse_collection(key: str) -> dict[str, BridgeObject]:
|
|
value = data.get(key)
|
|
if not isinstance(value, list):
|
|
raise BridgeDataError(f"{key} must be a list")
|
|
parsed = [
|
|
BridgeObject.from_dict(item) for item in value if isinstance(item, dict)
|
|
]
|
|
if len(parsed) != len(value):
|
|
raise BridgeDataError(f"{key} must contain only objects")
|
|
result = {item.id: item for item in parsed}
|
|
if len(result) != len(parsed):
|
|
raise BridgeDataError(f"{key} contains duplicate IDs")
|
|
return result
|
|
|
|
snapshot = cls(
|
|
revision=revision,
|
|
controllers=parse_collection("controllers"),
|
|
areas=parse_collection("areas"),
|
|
points=parse_collection("points"),
|
|
modules=parse_collection("modules"),
|
|
outputs=parse_collection("outputs"),
|
|
)
|
|
snapshot._validate_references()
|
|
return snapshot
|
|
|
|
def _validate_references(self) -> None:
|
|
for collection_name in ("areas", "points", "modules", "outputs"):
|
|
for item in getattr(self, collection_name).values():
|
|
if item.controller_id not in self.controllers:
|
|
raise BridgeDataError(
|
|
f"{collection_name} contains an unknown controller reference"
|
|
)
|
|
for point in self.points.values():
|
|
if point.area_id is not None and point.area_id not in self.areas:
|
|
raise BridgeDataError("points contains an unknown area reference")
|
|
if point.module_id is not None and point.module_id not in self.modules:
|
|
raise BridgeDataError("points contains an unknown module reference")
|
|
for output in self.outputs.values():
|
|
if output.module_id is not None and output.module_id not in self.modules:
|
|
raise BridgeDataError("outputs contains an unknown module reference")
|
|
|
|
def apply_event(self, event: DomainEvent) -> BridgeSnapshot | None:
|
|
"""Apply a contiguous state event or request snapshot reconciliation."""
|
|
if event.revision != self.revision + 1 or event.event_type != "state_changed":
|
|
return None
|
|
collection_name = {
|
|
"controller": "controllers",
|
|
"area": "areas",
|
|
"point": "points",
|
|
"module": "modules",
|
|
"output": "outputs",
|
|
}.get(event.object_type)
|
|
if collection_name is None:
|
|
return None
|
|
collection = getattr(self, collection_name)
|
|
current = collection.get(event.object_id)
|
|
if current is None:
|
|
return None
|
|
updated_collection = {
|
|
**collection,
|
|
event.object_id: replace(current, state=event.state),
|
|
}
|
|
return replace(
|
|
self,
|
|
revision=event.revision,
|
|
**{collection_name: updated_collection},
|
|
)
|