Add Home Assistant network control API

luz-musica.py

@@ -2,10 +2,13 @@
 from __future__ import annotations
 
 import colorsys
+import json
 import math
 import os
 import subprocess
+import threading
 import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from io import BytesIO
 from urllib.parse import parse_qs, urlparse
 
@@ -31,6 +34,9 @@ OKLCH_LIGHTNESS = 0.72
 OKLCH_CHROMA = 0.24
 MAX_BRIGHTNESS_STEP = 0.10
 PLAYERCTL_BIN = os.environ.get("PLAYERCTL_BIN", "playerctl")
+CONTROL_HOST = os.environ.get("CONTROL_HOST", "0.0.0.0")
+CONTROL_PORT = int(os.environ.get("CONTROL_PORT", "8765"))
+CONTROL_TOKEN = os.environ.get("CONTROL_TOKEN", "")
 YELLOW_HUE_START = 0.12
 YELLOW_HUE_END = 0.18
 AMBER_TARGET_HUE = 0.10
@@ -41,6 +47,101 @@ NEUTRAL_SATURATION_THRESHOLD = 0.18
 PURPLE_FALLBACK_HUE = 0.78
 
 
+class ControlState:
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._paused = False
+        self._force_update = False
+
+    def pause(self) -> None:
+        with self._lock:
+            self._paused = True
+
+    def resume(self) -> None:
+        with self._lock:
+            self._paused = False
+            self._force_update = True
+
+    def toggle(self) -> bool:
+        with self._lock:
+            self._paused = not self._paused
+            if not self._paused:
+                self._force_update = True
+            return self._paused
+
+    def is_paused(self) -> bool:
+        with self._lock:
+            return self._paused
+
+    def consume_force_update(self) -> bool:
+        with self._lock:
+            force_update = self._force_update
+            self._force_update = False
+            return force_update
+
+    def snapshot(self) -> dict[str, object]:
+        with self._lock:
+            return {
+                "paused": self._paused,
+                "running": True,
+            }
+
+
+CONTROL_STATE = ControlState()
+
+
+class ControlHandler(BaseHTTPRequestHandler):
+    def log_message(self, format: str, *args: object) -> None:
+        return
+
+    def do_GET(self) -> None:
+        if self.path == "/status":
+            self.write_json(200, CONTROL_STATE.snapshot())
+            return
+        self.write_json(404, {"error": "not found"})
+
+    def do_POST(self) -> None:
+        if not self.authorized():
+            self.write_json(401, {"error": "unauthorized"})
+            return
+
+        if self.path == "/pause":
+            CONTROL_STATE.pause()
+        elif self.path == "/resume":
+            CONTROL_STATE.resume()
+        elif self.path == "/toggle":
+            CONTROL_STATE.toggle()
+        else:
+            self.write_json(404, {"error": "not found"})
+            return
+
+        self.write_json(200, CONTROL_STATE.snapshot())
+
+    def authorized(self) -> bool:
+        if not CONTROL_TOKEN:
+            return True
+
+        header_token = self.headers.get("X-Music-Light-Token", "")
+        bearer_token = self.headers.get("Authorization", "").removeprefix("Bearer ")
+        return CONTROL_TOKEN in {header_token, bearer_token}
+
+    def write_json(self, status: int, payload: dict[str, object]) -> None:
+        body = json.dumps(payload).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+
+def start_control_server() -> ThreadingHTTPServer:
+    server = ThreadingHTTPServer((CONTROL_HOST, CONTROL_PORT), ControlHandler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    print(f"Controle HTTP ouvindo em {CONTROL_HOST}:{CONTROL_PORT}")
+    return server
+
+
 def run_playerctl(field: str, player: str | None = None) -> str:
     command = [PLAYERCTL_BIN]
     if player:
@@ -399,11 +500,16 @@ def set_light_color(rgb: tuple[int, int, int]) -> None:
 
 def main() -> None:
     print("Monitorando música atual. Ctrl+C para sair.")
+    control_server = start_control_server()
     last_track = ""
     last_rgb: tuple[int, int, int] | None = None
 
     while True:
         try:
+            if CONTROL_STATE.is_paused():
+                time.sleep(2)
+                continue
+
             player = choose_active_player()
             if not player:
                 time.sleep(2)
@@ -415,7 +521,7 @@ def main() -> None:
                 time.sleep(2)
                 continue
 
-            if track_id != last_track:
+            if CONTROL_STATE.consume_force_update() or track_id != last_track:
                 art_url = get_album_art_url(player)
                 if not art_url:
                     print("Sem capa disponível para a faixa atual.")
@@ -438,6 +544,8 @@ def main() -> None:
 
         except KeyboardInterrupt:
             print("\nEncerrado.")
+            control_server.shutdown()
+            control_server.server_close()
             break
         except Exception as exc:
             print(f"Erro: {exc}")