diff --git a/src/services/shelly.py b/src/services/shelly.py new file mode 100644 index 0000000..f8291fb --- /dev/null +++ b/src/services/shelly.py @@ -0,0 +1,59 @@ +import logging + +import requests + +logger = logging.getLogger(__name__) + + +class ShellyCloudAPI: + """Shelly Cloud Control API v2 client. + + *server_uri* — base URL of your Shelly Cloud server + (e.g. ``https://shelly-3.eu.shelly.cloud``). + *auth_key* — long-lived API key generated in the Shelly Cloud portal. + + Reference: https://shelly-api-docs.shelly.cloud/cloud-control-api/communication-v2 + """ + + def __init__(self, server_uri: str, auth_key: str): + self._server_uri = server_uri.rstrip("/") + self._auth_key = auth_key + + def open_gate(self, device_id: str, channel: int = 0) -> None: + """Send a switch-on command to the device via the v2 API. + + Raises on HTTP errors or API-level errors. + """ + url = f"{self._server_uri}/v2/devices/api/set/switch" + params = {"auth_key": self._auth_key} + payload = {"id": device_id, "channel": channel, "on": True} + logger.debug("Shelly v2 open_gate: device_id=%s channel=%d", device_id, channel) + response = requests.post(url, params=params, json=payload, timeout=10) + if not response.ok: + # v2 error body: {"error": "...", "data": {"messages": [...]}} + try: + body = response.json() + error_str = body.get("error", response.text) + messages = body.get("data", {}).get("messages", []) + detail = f"{error_str}: {'; '.join(messages)}" if messages else error_str + except Exception: + detail = response.text + raise Exception(f"Shelly Cloud API error ({response.status_code}): {detail}") + + def validate_credentials(self) -> bool: + """Validate the auth key by issuing a v2 get-state probe. + + Any response other than 401 (Unauthorized) is treated as valid auth. + Raises on unexpected network errors. + """ + url = f"{self._server_uri}/v2/devices/api/get" + params = {"auth_key": self._auth_key} + # Send a single dummy id; the server will return an empty/not-found result + # but will authenticate the key first. A 401 means the key is invalid. + payload = {"ids": ["validate"]} + response = requests.post(url, params=params, json=payload, timeout=10) + if response.status_code == 401: + logger.warning("Shelly credentials validation failed: 401 Unauthorized") + return False + logger.debug("Shelly credentials valid (status=%d)", response.status_code) + return True