159 lines
6.5 KiB
Python
159 lines
6.5 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 snapshot(self) -> dict[str, Any]:
|
|
"""Return devices, groups and the control plan in one restricted request."""
|
|
data = await self._request("GET", "/api/integrations/home-assistant/snapshot")
|
|
if not isinstance(data, dict):
|
|
raise GreeControllerApiError("Controller returned an invalid snapshot payload")
|
|
if not isinstance(data.get("devices"), list):
|
|
raise GreeControllerApiError("Controller returned invalid snapshot devices")
|
|
if not isinstance(data.get("control_plan"), dict):
|
|
raise GreeControllerApiError("Controller returned an invalid snapshot control plan")
|
|
if not isinstance(data.get("groups"), list):
|
|
raise GreeControllerApiError("Controller returned invalid snapshot groups")
|
|
return data
|
|
|
|
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,
|
|
)
|
|
|
|
async def control_plan(self) -> dict[str, Any]:
|
|
"""Return the current whole-house and zone automation plan."""
|
|
data = await self._request("GET", "/api/integrations/home-assistant/control-plan")
|
|
if not isinstance(data, dict):
|
|
raise GreeControllerApiError("Controller returned an invalid control plan payload")
|
|
return data
|
|
|
|
async def groups(self) -> list[dict[str, Any]]:
|
|
"""Return climate groups exposed to Home Assistant."""
|
|
data = await self._request("GET", "/api/integrations/home-assistant/groups")
|
|
if not isinstance(data, list):
|
|
raise GreeControllerApiError("Controller returned an invalid groups payload")
|
|
return data
|
|
|
|
async def group_control(self, group_id: str, payload: dict[str, Any]) -> dict[str, Any]:
|
|
"""Change one controller climate group."""
|
|
data = await self._request(
|
|
"POST",
|
|
f"/api/integrations/home-assistant/groups/{group_id}/control",
|
|
json=payload,
|
|
)
|
|
if not isinstance(data, dict):
|
|
raise GreeControllerApiError("Controller returned an invalid group control payload")
|
|
return data
|
|
|
|
async def house_control(self, mode: str) -> dict[str, Any]:
|
|
"""Change the whole-house thermostat mode."""
|
|
data = await self._request(
|
|
"POST",
|
|
"/api/integrations/home-assistant/house/control",
|
|
json={"mode": mode},
|
|
)
|
|
if not isinstance(data, dict):
|
|
raise GreeControllerApiError("Controller returned an invalid house control payload")
|
|
return data
|
|
|
|
async def house_preset(self, preset: str) -> dict[str, Any]:
|
|
"""Change the whole-house work profile."""
|
|
data = await self._request(
|
|
"POST",
|
|
"/api/integrations/home-assistant/house/preset",
|
|
json={"preset": preset},
|
|
)
|
|
if not isinstance(data, dict):
|
|
raise GreeControllerApiError("Controller returned an invalid house preset payload")
|
|
return data
|
|
|
|
async def house_power(self, power: bool) -> dict[str, Any]:
|
|
"""Turn all enabled air conditioners on or off."""
|
|
data = await self._request(
|
|
"POST",
|
|
"/api/integrations/home-assistant/house/power",
|
|
json={"power": power},
|
|
)
|
|
if not isinstance(data, dict):
|
|
raise GreeControllerApiError("Controller returned an invalid house power payload")
|
|
return data
|
|
|
|
async def zone_control(self, zone_id: str, payload: dict[str, Any]) -> dict[str, Any]:
|
|
"""Change a zone thermostat override through the controller."""
|
|
data = await self._request(
|
|
"POST",
|
|
f"/api/integrations/home-assistant/zones/{zone_id}/control",
|
|
json=payload,
|
|
)
|
|
if not isinstance(data, dict):
|
|
raise GreeControllerApiError("Controller returned an invalid zone payload")
|
|
return data
|