""" Panel para OBS: servidor HTTP local que sirve `dock.html` y lo mantiene sincronizado con la app. OBS Studio (28 o más) admite «Docks → Custom Browser Docks»: un panel acoplable que carga una URL. Labmagic levanta este servidor en 127.0.0.1 y el panel habla con la misma conexión Bluetooth que ya tiene la app (la cámara solo admite un proceso a la vez, así que el dock no se conecta solo). GET / -> dock.html GET /state -> estado actual en JSON GET /events -> Server-Sent Events: un `data:` con el estado cada vez que cambia POST /cmd -> JSON {"cmd": "...", ...} que la app ejecuta en su hilo de Tk """ import json import os import queue import threading import time import uuid from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer HERE = os.path.dirname(os.path.abspath(__file__)) DEFAULT_PORT = 8790 # 8765 lo usa el Organizador Labstream class DockServer: def __init__(self, cmd_queue: queue.Queue, port: int = DEFAULT_PORT, html_path: str = None): self.cmd_queue = cmd_queue self.port = port self.html_path = html_path or os.path.join(HERE, "dock.html") self._state = {} self._state_json = "{}" self._clients = set() self._lock = threading.Lock() self._httpd = None self._thread = None self.error = None # ----------------------------------------------------------- ciclo de vida def start(self): server = self class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, *_a): # silencio en consola pass def _send(self, code, body: bytes, ctype="text/plain; charset=utf-8", extra=None): self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") for k, v in (extra or {}).items(): self.send_header(k, v) self.end_headers() self.wfile.write(body) def do_GET(self): path = self.path.split("?", 1)[0] if path in ("/", "/index.html", "/dock.html"): try: with open(server.html_path, "rb") as fh: self._send(200, fh.read(), "text/html; charset=utf-8") except OSError as e: self._send(500, f"No se encuentra dock.html: {e}".encode()) elif path == "/state": self._send(200, server._state_json.encode("utf-8"), "application/json; charset=utf-8") elif path == "/events": self._sse() elif path == "/icon.png": icon = os.path.join(HERE, "icono.png") try: with open(icon, "rb") as fh: self._send(200, fh.read(), "image/png") except OSError: self._send(404, b"") else: self._send(404, b"no") def _sse(self): q = queue.Queue(maxsize=200) with server._lock: server._clients.add(q) try: self.send_response(200) self.send_header("Content-Type", "text/event-stream; charset=utf-8") self.send_header("Cache-Control", "no-store") self.send_header("Connection", "keep-alive") self.end_headers() self.wfile.write(f"retry: 1500\ndata: {server._state_json}\n\n".encode("utf-8")) self.wfile.flush() while True: try: msg = q.get(timeout=15) self.wfile.write(f"data: {msg}\n\n".encode("utf-8")) except queue.Empty: self.wfile.write(b": ping\n\n") self.wfile.flush() except (ConnectionError, OSError, ValueError): pass finally: with server._lock: server._clients.discard(q) def do_POST(self): path = self.path.split("?", 1)[0] n = int(self.headers.get("Content-Length") or 0) raw = self.rfile.read(n) if n else b"" if path != "/cmd": self._send(404, b"no") return try: data = json.loads(raw.decode("utf-8") or "{}") if not isinstance(data, dict) or "cmd" not in data: raise ValueError("falta cmd") except (ValueError, UnicodeDecodeError) as e: self._send(400, f"JSON no válido: {e}".encode()) return data["_t"] = time.time() server.cmd_queue.put(data) self._send(204, b"") # Si el puerto está ocupado por otro programa probamos los siguientes (la URL se guarda en config) last_err = None for port in range(self.port, self.port + 10): try: self._httpd = ThreadingHTTPServer(("127.0.0.1", port), Handler) self._httpd.daemon_threads = True self.port = port break except OSError as e: last_err = e self._httpd = None if self._httpd is None: self.error = f"No se pudo abrir ningún puerto entre {self.port} y {self.port + 9}: {last_err}" return False self._thread = threading.Thread(target=self._httpd.serve_forever, kwargs={"poll_interval": 0.5}, name="labmagic-dock", daemon=True) self._thread.start() return True def stop(self): if self._httpd is not None: try: self._httpd.shutdown() self._httpd.server_close() except Exception: # noqa: BLE001 pass self._httpd = None @property def url(self): return f"http://127.0.0.1:{self.port}/" @property def clients(self): with self._lock: return len(self._clients) # ----------------------------------------------------------- estado def publish(self, state: dict): """Envía el estado a los paneles abiertos solo si ha cambiado.""" js = json.dumps(state, ensure_ascii=False, separators=(",", ":")) if js == self._state_json: return self._state, self._state_json = state, js with self._lock: clients = list(self._clients) for q in clients: try: q.put_nowait(js) except queue.Full: pass # ------------------------------------------------------------------ registro en OBS def obs_user_ini(): return os.path.join(os.environ.get("APPDATA", ""), "obs-studio", "user.ini") def obs_running(): try: import subprocess out = subprocess.run(["tasklist", "/FI", "IMAGENAME eq obs64.exe", "/NH"], capture_output=True, text=True, creationflags=0x08000000).stdout return "obs64.exe" in out except Exception: # noqa: BLE001 return False def obs_add_browser_dock(title: str, url: str): """Añade el panel a «Docks → Custom Browser Docks» de OBS editando user.ini (OBS debe estar cerrado). Devuelve (ok, mensaje).""" ini = obs_user_ini() if not os.path.exists(ini): return False, "No se encuentra la configuración de OBS (user.ini). ¿Está instalado OBS 30 o superior?" if obs_running(): return False, "Cierra OBS antes: guarda su configuración al salir y borraría el cambio." with open(ini, "r", encoding="utf-8-sig") as fh: lines = fh.read().splitlines() docks = [] idx = None section = None basic_end = None for i, line in enumerate(lines): s = line.strip() if s.startswith("[") and s.endswith("]"): if section == "BasicWindow" and basic_end is None: basic_end = i section = s[1:-1] elif section == "BasicWindow" and s.startswith("ExtraBrowserDocks="): idx = i try: docks = json.loads(s.split("=", 1)[1] or "[]") except ValueError: docks = [] for d in docks: if isinstance(d, dict) and d.get("url", "").rstrip("/") == url.rstrip("/"): return True, "El panel ya estaba añadido en OBS (Docks → Labmagic)." docks.append({"title": title, "url": url, "uuid": str(uuid.uuid4())}) new_line = "ExtraBrowserDocks=" + json.dumps(docks, ensure_ascii=False, separators=(",", ":")) if idx is not None: lines[idx] = new_line else: # Sin la clave: la añadimos al final de [BasicWindow] (o creamos la sección) pos = None for i, line in enumerate(lines): if line.strip() == "[BasicWindow]": pos = i + 1 while pos < len(lines) and not lines[pos].strip().startswith("["): pos += 1 break if pos is None: lines += ["", "[BasicWindow]", new_line] else: lines.insert(pos, new_line) backup = ini + ".labmagic.bak" try: if not os.path.exists(backup): with open(ini, "rb") as src, open(backup, "wb") as dst: dst.write(src.read()) with open(ini, "w", encoding="utf-8") as fh: fh.write("\n".join(lines) + "\n") except OSError as e: return False, f"No se pudo escribir user.ini: {e}" return True, "Panel añadido. Abre OBS: aparece en Docks → Labmagic (arrástralo donde quieras)."