73 lines
2.3 KiB
Python
73 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid
|
|
from typing import Any
|
|
|
|
|
|
BASE_URL = "https://api.switch-bot.com/v1.1"
|
|
|
|
|
|
class SwitchBotError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class SwitchBotClient:
|
|
def __init__(self, token: str, secret: str) -> None:
|
|
self.token = token
|
|
self.secret = secret
|
|
|
|
def _headers(self) -> dict[str, str]:
|
|
timestamp_ms = str(int(time.time() * 1000))
|
|
nonce = str(uuid.uuid4())
|
|
message = f"{self.token}{timestamp_ms}{nonce}".encode("utf-8")
|
|
digest = hmac.new(self.secret.encode("utf-8"), message, hashlib.sha256).digest()
|
|
signature = base64.b64encode(digest).decode("utf-8")
|
|
return {
|
|
"Authorization": self.token,
|
|
"Content-Type": "application/json; charset=utf8",
|
|
"sign": signature,
|
|
"t": timestamp_ms,
|
|
"nonce": nonce,
|
|
}
|
|
|
|
def get(self, path: str) -> dict[str, Any]:
|
|
endpoint = path if path.startswith("/") else f"/{path}"
|
|
request = urllib.request.Request(
|
|
f"{BASE_URL}{endpoint}",
|
|
headers=self._headers(),
|
|
method="GET",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=20) as response:
|
|
body = response.read().decode("utf-8")
|
|
except urllib.error.HTTPError as exc:
|
|
details = exc.read().decode("utf-8", errors="replace")
|
|
raise SwitchBotError(f"HTTP {exc.code}: {details}") from exc
|
|
except urllib.error.URLError as exc:
|
|
raise SwitchBotError(f"Could not reach SwitchBot API: {exc.reason}") from exc
|
|
|
|
try:
|
|
payload = json.loads(body)
|
|
except json.JSONDecodeError as exc:
|
|
raise SwitchBotError(f"SwitchBot returned invalid JSON: {body}") from exc
|
|
|
|
if payload.get("statusCode") != 100:
|
|
raise SwitchBotError(f"SwitchBot API error: {payload}")
|
|
return payload
|
|
|
|
def devices(self) -> list[dict[str, Any]]:
|
|
payload = self.get("/devices")
|
|
body = payload.get("body", {})
|
|
return body.get("deviceList", [])
|
|
|
|
def status(self, device_id: str) -> dict[str, Any]:
|
|
payload = self.get(f"/devices/{device_id}/status")
|
|
return payload.get("body", {})
|