79 lines
2.4 KiB
Python
79 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Callable, Mapping
|
|
from dataclasses import dataclass
|
|
from typing import Any, Protocol
|
|
|
|
from fastapi import APIRouter
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
STATE_READ_ACTION_ID = "state.read"
|
|
|
|
|
|
class DevicePluginActionRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
input: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class DevicePluginActionAdapter(Protocol):
|
|
plugin_id: str
|
|
action_ids: frozenset[str]
|
|
|
|
async def invoke(self, action_id: str, payload: Mapping[str, Any]) -> dict[str, Any]: ...
|
|
|
|
|
|
def _noop() -> None:
|
|
return None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DevicePluginRuntimeContribution:
|
|
"""One reviewed backend plugin contribution loaded from its manifest factory."""
|
|
|
|
adapter: DevicePluginActionAdapter
|
|
legacy_routers: tuple[APIRouter, ...] = ()
|
|
close: Callable[[], None] = _noop
|
|
|
|
|
|
class PluginNotFoundError(LookupError):
|
|
"""The requested plugin is not installed in the runtime composition."""
|
|
|
|
|
|
class PluginActionNotFoundError(LookupError):
|
|
"""The requested action is not declared by the selected plugin."""
|
|
|
|
|
|
class PluginExecutionError(RuntimeError):
|
|
"""A validated plugin action failed while talking to its device/runtime."""
|
|
|
|
|
|
class DevicePluginDispatcher:
|
|
"""Host-owned dispatcher for allowlisted, namespaced plugin actions."""
|
|
|
|
def __init__(self, adapters: list[DevicePluginActionAdapter]) -> None:
|
|
self._adapters: dict[str, DevicePluginActionAdapter] = {}
|
|
for adapter in adapters:
|
|
if adapter.plugin_id in self._adapters:
|
|
raise ValueError(f"Duplicate runtime device-plugin id: {adapter.plugin_id}")
|
|
self._adapters[adapter.plugin_id] = adapter
|
|
|
|
@property
|
|
def action_declarations(self) -> dict[str, frozenset[str]]:
|
|
return {plugin_id: adapter.action_ids for plugin_id, adapter in self._adapters.items()}
|
|
|
|
async def invoke(
|
|
self,
|
|
plugin_id: str,
|
|
action_id: str,
|
|
payload: Mapping[str, Any],
|
|
) -> dict[str, Any]:
|
|
adapter = self._adapters.get(plugin_id)
|
|
if adapter is None:
|
|
raise PluginNotFoundError(f"Device plugin is not installed: {plugin_id}")
|
|
if action_id not in adapter.action_ids:
|
|
raise PluginActionNotFoundError(
|
|
f"Device plugin {plugin_id} does not declare action {action_id}"
|
|
)
|
|
return await adapter.invoke(action_id, payload)
|