77 lines
2.8 KiB
Python
77 lines
2.8 KiB
Python
"""HTTP client for the standalone GREE Controller service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from aiohttp import ClientError, ClientSession
|
|
|
|
|
|
class GreeControllerApiError(Exception):
|
|
"""Raised when the controller API cannot be used."""
|
|
|
|
|
|
class GreeControllerClient:
|
|
"""Small async client backed by Home Assistant's shared ClientSession."""
|
|
|
|
def __init__(self, session: ClientSession, base_url: str, token: str = "") -> None:
|
|
self._session = session
|
|
self._base_url = base_url.rstrip("/")
|
|
self._token = token.strip()
|
|
|
|
@property
|
|
def base_url(self) -> str:
|
|
"""Return the normalized controller URL."""
|
|
return self._base_url
|
|
|
|
def _headers(self) -> dict[str, str]:
|
|
headers = {"Accept": "application/json"}
|
|
if self._token:
|
|
headers["Authorization"] = f"Bearer {self._token}"
|
|
return headers
|
|
|
|
async def _request(self, method: str, path: str, **kwargs: Any) -> Any:
|
|
try:
|
|
async with self._session.request(
|
|
method,
|
|
f"{self._base_url}{path}",
|
|
headers=self._headers(),
|
|
timeout=10,
|
|
**kwargs,
|
|
) as response:
|
|
if response.status == 401:
|
|
raise GreeControllerApiError("Controller authentication failed")
|
|
if response.status >= 400:
|
|
try:
|
|
body = await response.json()
|
|
message = body.get("error", f"HTTP {response.status}")
|
|
except (ValueError, TypeError):
|
|
message = f"HTTP {response.status}"
|
|
raise GreeControllerApiError(message)
|
|
if response.status == 204:
|
|
return None
|
|
return await response.json()
|
|
except GreeControllerApiError:
|
|
raise
|
|
except (ClientError, TimeoutError) as err:
|
|
raise GreeControllerApiError(str(err)) from err
|
|
|
|
async def health(self) -> dict[str, Any]:
|
|
"""Return the public controller health payload."""
|
|
return await self._request("GET", "/api/health")
|
|
|
|
async def devices(self) -> list[dict[str, Any]]:
|
|
"""Return all controller devices."""
|
|
data = await self._request("GET", "/api/integrations/home-assistant/devices")
|
|
if not isinstance(data, list):
|
|
raise GreeControllerApiError("Controller returned an invalid devices payload")
|
|
return data
|
|
|
|
async def command(self, device_id: str, payload: dict[str, Any]) -> dict[str, Any]:
|
|
"""Send a device command and return the updated device state."""
|
|
return await self._request(
|
|
"POST",
|
|
f"/api/integrations/home-assistant/devices/{device_id}/command",
|
|
json=payload,
|
|
)
|