Add support for Shelly Cloud API
This commit is contained in:
59
src/services/shelly.py
Normal file
59
src/services/shelly.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user