Initial Commit

app/switchbot.py

@@ -0,0 +1,72 @@
+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", {})