GREE Controller 0.13.9

custom_components/gree_controller/api.py

@@ -0,0 +1,158 @@
+"""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