201 lines
7.3 KiB
Python
201 lines
7.3 KiB
Python
"""Asynchronous client for the vendor-neutral Bridge API v1."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from collections.abc import AsyncIterator
|
|
from typing import Any
|
|
from urllib.parse import quote, urlsplit
|
|
|
|
import aiohttp
|
|
|
|
from .const import API_VERSION, MIN_TOKEN_LENGTH
|
|
from .models import (
|
|
BridgeDataError,
|
|
BridgeInfo,
|
|
BridgeSnapshot,
|
|
CommandResult,
|
|
DomainEvent,
|
|
)
|
|
|
|
|
|
class BridgeError(Exception):
|
|
"""Base bridge client error."""
|
|
|
|
|
|
class BridgeConnectionError(BridgeError):
|
|
"""Raised when the bridge cannot be reached."""
|
|
|
|
|
|
class BridgeAuthenticationError(BridgeError):
|
|
"""Raised when the local bridge token is rejected."""
|
|
|
|
|
|
class BridgeVersionError(BridgeError):
|
|
"""Raised when the bridge API version is unsupported."""
|
|
|
|
|
|
class BridgeResponseError(BridgeError):
|
|
"""Raised when the bridge returns an invalid response."""
|
|
|
|
|
|
def normalize_base_url(host: str, port: int) -> str:
|
|
"""Build a normalized local bridge base URL."""
|
|
candidate = host.strip()
|
|
if not candidate:
|
|
raise ValueError("host must not be empty")
|
|
if "://" not in candidate:
|
|
candidate = f"http://{candidate}"
|
|
parsed = urlsplit(candidate)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.hostname:
|
|
raise ValueError("host must be a valid HTTP or HTTPS address")
|
|
if parsed.path not in {"", "/"} or parsed.query or parsed.fragment:
|
|
raise ValueError("host must not contain a path, query, or fragment")
|
|
effective_port = parsed.port or port
|
|
host_part = f"[{parsed.hostname}]" if ":" in parsed.hostname else parsed.hostname
|
|
return f"{parsed.scheme}://{host_part}:{effective_port}"
|
|
|
|
|
|
class BridgeClient:
|
|
"""Client for Bridge API v1."""
|
|
|
|
def __init__(
|
|
self,
|
|
session: aiohttp.ClientSession,
|
|
*,
|
|
host: str,
|
|
port: int,
|
|
token: str,
|
|
) -> None:
|
|
"""Initialize the client without logging credentials."""
|
|
if len(token) < MIN_TOKEN_LENGTH:
|
|
raise ValueError("token is too short")
|
|
self._session = session
|
|
self._base_url = normalize_base_url(host, port)
|
|
self._headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
@property
|
|
def base_url(self) -> str:
|
|
"""Return the normalized non-secret endpoint."""
|
|
return self._base_url
|
|
|
|
async def async_get_info(self) -> BridgeInfo:
|
|
"""Return bridge metadata and enforce API compatibility."""
|
|
info = BridgeInfo.from_dict(await self._async_request_json("GET", "/v1/info"))
|
|
if info.api_version != API_VERSION:
|
|
raise BridgeVersionError(
|
|
f"unsupported Bridge API version {info.api_version}"
|
|
)
|
|
return info
|
|
|
|
async def async_get_snapshot(self) -> BridgeSnapshot:
|
|
"""Return the current normalized snapshot."""
|
|
return BridgeSnapshot.from_dict(
|
|
await self._async_request_json("GET", "/v1/snapshot")
|
|
)
|
|
|
|
async def async_get_diagnostics(self) -> dict[str, Any]:
|
|
"""Return diagnostics already sanitized by the bridge."""
|
|
return await self._async_request_json("GET", "/v1/diagnostics")
|
|
|
|
async def async_arm_area(self, object_id: str, mode: str) -> CommandResult:
|
|
"""Arm an area using a bridge-reported mode capability."""
|
|
return await self._async_command(
|
|
f"/v1/areas/{quote(object_id, safe='')}/arm", {"mode": mode}
|
|
)
|
|
|
|
async def async_disarm_area(self, object_id: str) -> CommandResult:
|
|
"""Disarm an area."""
|
|
return await self._async_command(
|
|
f"/v1/areas/{quote(object_id, safe='')}/disarm"
|
|
)
|
|
|
|
async def async_set_output(self, object_id: str, state: str) -> CommandResult:
|
|
"""Set an output state."""
|
|
return await self._async_command(
|
|
f"/v1/outputs/{quote(object_id, safe='')}/set", {"state": state}
|
|
)
|
|
|
|
async def async_run_command(self, command: str) -> CommandResult:
|
|
"""Run a named capability reported by the bridge."""
|
|
return await self._async_command(f"/v1/commands/{quote(command, safe='')}")
|
|
|
|
async def async_events(self) -> AsyncIterator[DomainEvent]:
|
|
"""Yield normalized server-sent events until disconnected."""
|
|
try:
|
|
async with self._session.get(
|
|
f"{self._base_url}/v1/events",
|
|
headers=self._headers,
|
|
timeout=aiohttp.ClientTimeout(
|
|
total=None, sock_connect=10, sock_read=None
|
|
),
|
|
) as response:
|
|
self._raise_for_status(response.status)
|
|
data_lines: list[str] = []
|
|
async for raw_line in response.content:
|
|
line = raw_line.decode("utf-8").rstrip("\r\n")
|
|
if not line:
|
|
if data_lines:
|
|
yield self._parse_event("\n".join(data_lines))
|
|
data_lines.clear()
|
|
continue
|
|
if line.startswith("data:"):
|
|
data_lines.append(line[5:].lstrip())
|
|
if data_lines:
|
|
yield self._parse_event("\n".join(data_lines))
|
|
raise BridgeConnectionError("bridge event stream ended")
|
|
except (TimeoutError, aiohttp.ClientError, UnicodeDecodeError) as err:
|
|
raise BridgeConnectionError("bridge event stream is unavailable") from err
|
|
|
|
async def _async_command(
|
|
self, path: str, payload: dict[str, Any] | None = None
|
|
) -> CommandResult:
|
|
result = CommandResult.from_dict(
|
|
await self._async_request_json("POST", path, payload)
|
|
)
|
|
if not result.accepted:
|
|
raise BridgeResponseError("bridge rejected the command")
|
|
return result
|
|
|
|
async def _async_request_json(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
request_payload: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
try:
|
|
async with asyncio.timeout(10):
|
|
async with self._session.request(
|
|
method,
|
|
f"{self._base_url}{path}",
|
|
headers=self._headers,
|
|
json=request_payload,
|
|
) as response:
|
|
self._raise_for_status(response.status)
|
|
response_payload = await response.json(content_type=None)
|
|
except (TimeoutError, aiohttp.ClientError) as err:
|
|
raise BridgeConnectionError("bridge is unavailable") from err
|
|
except ValueError as err:
|
|
raise BridgeResponseError("bridge returned invalid JSON") from err
|
|
if not isinstance(response_payload, dict):
|
|
raise BridgeResponseError("bridge response must be an object")
|
|
return response_payload
|
|
|
|
@staticmethod
|
|
def _raise_for_status(status: int) -> None:
|
|
if status in {401, 403}:
|
|
raise BridgeAuthenticationError("bridge authentication failed")
|
|
if status >= 400:
|
|
raise BridgeResponseError(f"bridge request failed with HTTP {status}")
|
|
|
|
@staticmethod
|
|
def _parse_event(payload: str) -> DomainEvent:
|
|
try:
|
|
value = json.loads(payload)
|
|
if not isinstance(value, dict):
|
|
raise BridgeDataError("event must be an object")
|
|
return DomainEvent.from_dict(value)
|
|
except ValueError as err:
|
|
raise BridgeResponseError("bridge returned an invalid event") from err
|