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", {})