#!/usr/bin/env python3 """ FOSFENO - Motor de visuales audio-reactivas para Raspberry Pi OS. Un unico proceso que: - sirve el PANEL de control (http:///) - sirve el ESCENARIO (http:///stage) -> lo abre Chromium en kiosko - lanza/cierra Chromium y el visualizador nativo projectM - mantiene el estado y lo sincroniza con panel y escenario via WebSocket Motores: projectm (nativo), butterchurn, hydra, shaders (GLSL) y mixer (mezclador VJ basado en Hydra: camara + video + efectos). """ import json import os import re import shutil import socket import subprocess import threading import time from pathlib import Path from flask import Flask, request, send_from_directory from flask_socketio import SocketIO from werkzeug.utils import secure_filename # -------------------------------------------------------------------------- # Rutas y configuracion # -------------------------------------------------------------------------- BASE = Path(__file__).resolve().parent.parent # .../FOSFENO WEB = BASE / "web" DATA = BASE / "data" VIDEOS = DATA / "videos" VIDEO_EXT = (".mp4", ".webm", ".mov", ".m4v", ".ogv") IMAGE_EXT = (".jpg", ".jpeg", ".png", ".gif", ".webp") MEDIA_EXT = VIDEO_EXT + IMAGE_EXT # la galeria del mezclador admite ambos def load_json(path, default): try: with open(path, encoding="utf-8") as fh: return json.load(fh) except (OSError, json.JSONDecodeError) as exc: print(f"[FOSFENO] No se pudo leer {path}: {exc}") return default CFG = load_json(BASE / "config.json", {}) HOST = CFG.get("server", {}).get("host", "0.0.0.0") # El puerto y el modo kiosko se pueden forzar por variable de entorno; asi el # modo portatil (script 'fosfeno') usa el puerto 8080 sin tocar config.json. PORT = int(os.environ.get("FOSFENO_PORT") or CFG.get("server", {}).get("port", 80)) NO_KIOSK = bool(os.environ.get("FOSFENO_NO_KIOSK")) app = Flask(__name__, static_folder=None) app.config["MAX_CONTENT_LENGTH"] = 4 * 1024 ** 3 # subidas de hasta 4 GB socketio = SocketIO(app, async_mode="threading", cors_allowed_origins="*") # -------------------------------------------------------------------------- # Datos cargados de disco # -------------------------------------------------------------------------- HYDRA_SKETCHES = load_json(DATA / "hydra-sketches.json", {}) SHADERS = load_json(DATA / "shaders.json", {}) DEFAULTS = CFG.get("defaults", {}) ENGINES = ("projectm", "butterchurn", "hydra", "shaders", "mixer") def first_code(table, key, field="code"): """Devuelve el codigo de una entrada de un diccionario de presets.""" entry = table.get(key) or next(iter(table.values()), {}) return entry.get(field, "") if isinstance(entry, dict) else "" def list_videos(): if not VIDEOS.is_dir(): return [] return sorted(f.name for f in VIDEOS.iterdir() if f.suffix.lower() in MEDIA_EXT) def list_projectm_presets(): """Categorias y visuales de data/presets-projectm. El pack se organiza como categoria/visual/variante.milk: cada 'visual' (Aurora, Blobby...) es una carpeta con variantes del mismo estilo. Devuelve {categoria: [visuales]}. Las carpetas '!...' se saltan (transiciones).""" base = DATA / "presets-projectm" cats = {} if not base.is_dir(): return cats for d in sorted(base.iterdir()): if not d.is_dir() or d.name.startswith(("!", ".")): continue visuals = sorted( v.name for v in d.iterdir() if v.is_dir() and any(f.suffix.lower() == ".milk" for f in v.iterdir())) if visuals: cats[d.name] = visuals return cats # -------------------------------------------------------------------------- # Estado global (se difunde a todos los clientes) # -------------------------------------------------------------------------- NOTIFY_HISTORY = [] # ultimos avisos mostrados en el panel state = { "engine": DEFAULTS.get("engine", "butterchurn"), "power": True, "sensitivity": float(DEFAULTS.get("sensitivity", 1.0)), # source: "mic" (microfono de la sala) o "monitor" (lo que suena en el # propio equipo: navegador, Spotify... captado por el monitor del sistema). "audio": {"device": "", "bpm": 0, "source": "mic"}, "butterchurn": { "preset": "", "shuffle": True, "interval": 20, "intervalMode": "seconds", "blendTime": 2.7, }, "hydra": { "label": "", "code": first_code(HYDRA_SKETCHES, DEFAULTS.get("hydraSketch", "")), }, "shaders": { "label": "", "code": first_code(SHADERS, DEFAULTS.get("shader", "")), }, "mixer": { # fondo: capa de abajo de la mezcla ("cam" o "butter" = MilkDrop) "source": "cam", "fondo": "cam", "camOn": True, "cameraId": 0, "video": "", "mix": 0.5, "blendMode": "blend", "hue": 0.0, "saturate": 1.0, "contrast": 1.0, "brightness": 0.0, "colorama": 0.0, "posterize": 0, "pixelate": 0, "kaleid": 0, "rotate": 0.0, "feedback": 0.0, "invert": False, "beatPulse": False, }, # projectM: categoria (subcarpeta de presets), preset concreto (fijo), # segundos por visual y orden aleatorio. Cambiarlos relanza projectM. # locked: visual congelado con la tecla de bloqueo (se pierde al relanzar). # nowPlaying/nowCategory/nowVisual: lo que esta en pantalla ahora mismo, # leido del log de projectM (~/.config/projectM/projectMSDL.log). # monitor: "auto" (la pantalla del escenario) o el numero de monitor. # variante: un .milk exacto elegido a dedo ("" = rota dentro del visual). # favoritos: visuales guardados para pincharlos en directo (persisten # en data/pm-favoritos.json). "projectm": {"available": False, "category": "", "preset": "", "duration": 60, "shuffle": True, "locked": False, "nowPlaying": "", "nowCategory": "", "nowVisual": "", "monitor": "auto", "variante": "", "favoritos": load_json(DATA / "pm-favoritos.json", [])}, # mapping: projection mapping in-browser. enabled = deformar la salida; # edit = mostrar los tiradores en el escenario para ajustarlo con el raton; # surfaces = [{id, corners:[[x,y]x4]}] normalizado 0..1. Persiste en # data/mapping.json. "mapping": load_json(DATA / "mapping.json", {"enabled": False, "edit": False, "surfaces": [], "masks": []}), "meta": { "butterchurnPresets": [], "audioDevices": [], "cameraDevices": [], "videos": list_videos(), "projectmPresets": {}, "monitors": [], }, "status": {"label": ""}, "notifications": NOTIFY_HISTORY, "network": {"ip": "", "hostname": "", "port": PORT, "panelSeen": False}, } lock = threading.Lock() def broadcast(): socketio.emit("state", state) def notify(level, message): """Registra un aviso y lo envia al panel. level puede ser 'info', 'warn' o 'error'. Asi el usuario nunca se queda sin saber que ha pasado: cualquier fallo aparece en la parte de arriba del panel de control. """ entry = {"level": level, "message": str(message), "t": int(time.time())} print(f"[FOSFENO] aviso({level}): {message}") NOTIFY_HISTORY.append(entry) while len(NOTIFY_HISTORY) > 20: NOTIFY_HISTORY.pop(0) try: socketio.emit("notify", entry) except Exception: pass def get_lan_ip(): """Devuelve la IP de la Raspberry en la red local.""" sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: sock.connect(("8.8.8.8", 80)) return sock.getsockname()[0] except OSError: return "127.0.0.1" finally: sock.close() def refresh_network(): """Actualiza la IP y el nombre de red de la Raspberry. Devuelve True si la IP ha cambiado, para avisar al panel y al escenario.""" new_ip = get_lan_ip() changed = state["network"]["ip"] != new_ip state["network"]["ip"] = new_ip state["network"]["hostname"] = socket.gethostname() return changed # -------------------------------------------------------------------------- # Gestion de procesos (Chromium kiosko y projectM nativo) # -------------------------------------------------------------------------- procs = {"chromium": None, "projectm": None} def is_running(proc): return proc is not None and proc.poll() is None def start_chromium(): """Abre el escenario en Chromium a pantalla completa.""" if is_running(procs["chromium"]): return kiosk = CFG.get("kiosk", {}) browser = shutil.which(kiosk.get("browser", "chromium-browser")) \ or shutil.which("chromium") if not browser: notify("error", "No se encontro Chromium. Las visuales web no pueden " "mostrarse. Ejecuta install.sh para instalarlo.") return url = f"http://localhost:{PORT}/stage" procs["chromium"] = subprocess.Popen([ browser, "--kiosk", f"--app={url}", "--use-fake-ui-for-media-stream", # acepta micro y camara sin preguntar "--autoplay-policy=no-user-gesture-required", "--noerrdialogs", "--disable-infobars", "--disable-translate", "--disable-features=Translate", "--check-for-update-interval=31536000", "--ozone-platform-hint=auto", ]) print("[FOSFENO] Chromium abierto en", url) def stop_projectm(): proc = procs["projectm"] if is_running(proc): proc.terminate() try: proc.wait(timeout=5) except subprocess.TimeoutExpired: proc.kill() procs["projectm"] = None def save_favoritos(): """Persiste los visuales favoritos de projectM en data/pm-favoritos.json.""" try: with open(DATA / "pm-favoritos.json", "w", encoding="utf-8") as fh: json.dump(state["projectm"]["favoritos"], fh, ensure_ascii=False, indent=2) except OSError as exc: notify("warn", f"No pude guardar los favoritos: {exc}") def list_monitors(): """Monitores conectados segun xrandr, para el selector del panel. Devuelve [{'id': 1, 'name': 'eDP-1 (1920x1080)'}, ...]; el id va en el formato de projectMSDL (1 = primer monitor).""" if not shutil.which("xrandr"): return [] mons = [] try: out = subprocess.run( ["xrandr", "--listmonitors"], capture_output=True, text=True, timeout=3, check=False).stdout for line in out.splitlines(): m = re.search(r"^\s*(\d+):\s+\+?\*?(\S+)\s+(\d+)/\d+x(\d+)/\d+", line) if m: idx, name, w, h = m.groups() mons.append({"id": int(idx) + 1, "name": f"{name} ({w}x{h})"}) except (OSError, subprocess.SubprocessError): pass return mons def monitor_of_stage(): """Numero de monitor (formato projectMSDL: 1 = primero) donde esta la ventana del escenario. Con varias pantallas, projectM debe abrirse siempre encima del escenario (el proyector) y no donde este el raton. Devuelve 0 (posicion por defecto del SO) si no se puede averiguar.""" if not shutil.which("xdotool") or not shutil.which("xrandr"): return 0 try: geo = subprocess.run( ["xdotool", "search", "--name", "FOSFENO :: Escenario", "getwindowgeometry"], capture_output=True, text=True, timeout=3, check=False).stdout pos = re.search(r"Position: (-?\d+),(-?\d+)", geo) if not pos: return 0 wx, wy = int(pos.group(1)), int(pos.group(2)) mons = subprocess.run( ["xrandr", "--listmonitors"], capture_output=True, text=True, timeout=3, check=False).stdout for line in mons.splitlines(): m = re.search( r"^\s*(\d+):\s+\S+\s+(\d+)/\d+x(\d+)/\d+([+-]\d+)([+-]\d+)", line) if not m: continue idx, w, h, x, y = (int(v) for v in m.groups()) if x <= wx < x + w and y <= wy < y + h: return idx + 1 except (OSError, subprocess.SubprocessError, ValueError): pass return 0 def start_projectm(): """Lanza projectM nativo con los ajustes del panel (categoria, preset fijo, duracion, aleatorio); su ventana se superpone a Chromium.""" stop_projectm() script = BASE / "scripts" / "start-projectm.sh" pm = state["projectm"] base = DATA / "presets-projectm" path = base if pm.get("category") and (base / pm["category"]).is_dir(): path = base / pm["category"] try: duration = max(2, int(pm.get("duration") or 20)) except (TypeError, ValueError): duration = 20 shuffle = "1" if pm.get("shuffle", True) else "0" # Visual concreto elegido: cada visual es una carpeta con variantes del # mismo estilo -> projectM rota solo dentro de ese look. if pm.get("preset") and pm.get("category"): vis = base / pm["category"] / pm["preset"] if vis.is_dir(): path = vis # Variante exacta elegida a dedo: projectM solo entiende carpetas, asi # que se le prepara una carpeta con esa unica variante dentro. if pm.get("variante") and pm.get("category") and pm.get("preset"): origen = (base / pm["category"] / pm["preset"] / (pm["variante"] + ".milk")) if origen.is_file(): fijado = DATA / "pm-fijado" shutil.rmtree(fijado, ignore_errors=True) fijado.mkdir(parents=True, exist_ok=True) shutil.copy(origen, fijado / origen.name) path = fijado # Pantalla elegida en el panel: "auto" sigue al escenario; un numero # la fija aunque el escenario este en otra (o no exista la ventana). try: mon = int(pm.get("monitor")) except (TypeError, ValueError): mon = 0 if mon <= 0: mon = monitor_of_stage() env = dict(os.environ, FOSFENO_PM_PATH=str(path), FOSFENO_PM_DURATION=str(duration), FOSFENO_PM_SHUFFLE=shuffle, FOSFENO_PM_MONITOR=str(mon)) # El bloqueo con tecla no sobrevive al relanzamiento del proceso pm["locked"] = False try: procs["projectm"] = subprocess.Popen(["bash", str(script)], env=env) print(f"[FOSFENO] projectM lanzado (presets: {path.name}, " f"{duration}s, shuffle={shuffle}).") except OSError as exc: notify("error", f"No se pudo lanzar projectM: {exc}. " "Prueba con otro motor de visuales.") return # projectM (SDL) abre su captura en la entrada por defecto; si toca # "audio del sistema", mover su stream en cuanto aparezca en pactl. # En la Raspberry projectM tarda varios segundos en arrancar: se le da # margen de sobra (2s + 30 intentos x 0.3s = ~11s). if state["audio"]["source"] == "monitor": threading.Timer( 2.0, lambda: route_captures("monitor", quiet=True, attempts=30)).start() def projectm_key(key): """Envia una tecla a projectM (solo funciona en sesion X11, no Wayland).""" if not shutil.which("xdotool"): return subprocess.run( ["xdotool", "search", "--name", "projectM", "key", "--clearmodifiers", key], timeout=3, check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) # projectM escribe en su log cada visual que muestra, con la ruta completa: # es la forma fiable de saber que hay en pantalla en cada momento. PM_LOG = Path.home() / ".config" / "projectM" / "projectMSDL.log" def projectm_now_playing(): """Ultimo 'Displaying preset' del log -> (categoria, visual, variante).""" try: with open(PM_LOG, "rb") as fh: fh.seek(0, os.SEEK_END) fh.seek(max(0, fh.tell() - 8192)) tail = fh.read().decode("utf-8", "replace") except OSError: return "", "", "" ruta = "" for line in tail.splitlines(): if "Displaying preset:" in line: ruta = line.split("Displaying preset:", 1)[1].strip() if not ruta: return "", "", "" try: cat, visual, variante = Path(ruta).relative_to( DATA / "presets-projectm").parts except ValueError: return "", "", Path(ruta).stem # preset fuera del catalogo propio return cat, visual, Path(variante).stem def watch_projectm_log(): """Bucle de fondo: refleja en el estado el visual que projectM muestra, para que el panel ensene su nombre y pueda fijarlo con un boton.""" while True: time.sleep(2) if not (state["engine"] == "projectm" and state["power"] and is_running(procs["projectm"])): continue cat, visual, variante = projectm_now_playing() if not variante or variante == state["projectm"]["nowPlaying"]: continue with lock: state["projectm"]["nowPlaying"] = variante state["projectm"]["nowCategory"] = cat state["projectm"]["nowVisual"] = visual state["status"]["label"] = variante broadcast() def apply_engine(): """Arranca o detiene projectM segun el motor y el estado de encendido.""" use_native = state["engine"] == "projectm" and state["power"] if state["engine"] == "projectm" and not state["projectm"]["available"]: notify("warn", "projectM no esta instalado en esta Raspberry. " "Elige otro motor o vuelve a compilarlo con " "scripts/build-projectm.sh.") if use_native and state["projectm"]["available"]: start_projectm() else: stop_projectm() # -------------------------------------------------------------------------- # Rutas HTTP (panel, escenario, librerias, datos) # -------------------------------------------------------------------------- @app.route("/") def panel_index(): return send_from_directory(str(WEB / "panel"), "index.html") @app.route("/panel/") def panel_files(fname): return send_from_directory(str(WEB / "panel"), fname) @app.route("/stage") def stage_index(): return send_from_directory(str(WEB / "stage"), "index.html") @app.route("/stage/") def stage_files(fname): return send_from_directory(str(WEB / "stage"), fname) @app.route("/lib/") def lib_files(fname): return send_from_directory(str(WEB / "lib"), fname) @app.route("/data/") def data_files(fname): return send_from_directory(str(DATA), fname) @app.route("/pm/variantes") def pm_variantes(): """Lista las variantes (.milk) de un visual, para el selector del panel.""" base = (DATA / "presets-projectm").resolve() d = (base / request.args.get("categoria", "") / request.args.get("visual", "")).resolve() if not str(d).startswith(str(base) + os.sep) or not d.is_dir(): return {"variantes": []} return {"variantes": sorted( f.stem for f in d.iterdir() if f.suffix.lower() == ".milk")} @app.route("/upload", methods=["POST"]) def upload_media(): """Sube un video o una imagen a la galeria del mezclador (data/videos).""" f = request.files.get("file") if not f or not f.filename: return {"ok": False, "error": "No llego ningun archivo."}, 400 name = secure_filename(f.filename) if not name or not name.lower().endswith(MEDIA_EXT): return {"ok": False, "error": "Formato no valido. Sube video (mp4, webm, mov, " "m4v, ogv) o imagen (jpg, png, gif, webp)."}, 400 try: VIDEOS.mkdir(parents=True, exist_ok=True) f.save(str(VIDEOS / name)) except OSError as exc: notify("error", f"No se pudo guardar '{name}': {exc}") return {"ok": False, "error": f"No se pudo guardar: {exc}"}, 500 with lock: state["meta"]["videos"] = list_videos() notify("info", f"'{name}' anadido a la galeria del mezclador.") broadcast() return {"ok": True, "name": name} # -------------------------------------------------------------------------- # Eventos WebSocket # -------------------------------------------------------------------------- @socketio.on("connect") def on_connect(): socketio.emit("state", state) @socketio.on("hello") def on_hello(data): """El panel se presenta al conectarse. Asi el escenario sabe que alguien ya ha encontrado el panel y puede dejar de mostrar el codigo QR.""" if (data or {}).get("role") == "panel" and not state["network"]["panelSeen"]: with lock: state["network"]["panelSeen"] = True broadcast() @socketio.on("set_engine") def on_set_engine(data): engine = (data or {}).get("engine") if engine not in ENGINES: return with lock: state["engine"] = engine apply_engine() broadcast() @socketio.on("set_power") def on_set_power(data): with lock: state["power"] = bool((data or {}).get("on")) apply_engine() broadcast() @socketio.on("set_sensitivity") def on_set_sensitivity(data): try: value = float((data or {}).get("value", 1.0)) except (TypeError, ValueError): return with lock: state["sensitivity"] = max(0.0, min(4.0, value)) broadcast() @socketio.on("run_code") def on_run_code(data): """Ejecuta codigo (Hydra o GLSL) escrito/pegado o un fragmento de la libreria.""" engine = (data or {}).get("engine") code = (data or {}).get("code") label = (data or {}).get("label", "codigo personalizado") if engine in ("hydra", "shaders") and isinstance(code, str): with lock: state[engine]["code"] = code state[engine]["label"] = label state["status"]["label"] = label broadcast() @socketio.on("update_settings") def on_update_settings(data): """Actualiza ajustes de un motor: butterchurn, mixer, audio o projectm.""" engine = (data or {}).get("engine") patch = (data or {}).get("patch", {}) if engine in ("butterchurn", "mixer", "audio", "projectm") \ and isinstance(patch, dict): with lock: state[engine].update(patch) # Los ajustes de projectM se aplican relanzandolo (tarda ~1s) if engine == "projectm" and state["engine"] == "projectm" \ and state["power"]: start_projectm() # Cambio de fuente de audio: ademas del reenganche que hara el # escenario, enrutamos aqui por si esta cerrado o si projectM (que # no pasa por el navegador) es el motor activo. Con un microfono # concreto elegido no se toca nada: lo fija el propio navegador. src = patch.get("source") if engine == "audio" else None if src == "monitor" or (src == "mic" and not state["audio"]["device"]): threading.Timer( 1.5, lambda: route_captures(src, quiet=True)).start() broadcast() def save_mapping(): try: with open(DATA / "mapping.json", "w", encoding="utf-8") as fh: json.dump(state["mapping"], fh, ensure_ascii=False, indent=2) except OSError as e: notify("warn", f"No se pudo guardar el mapping: {e}") @socketio.on("set_mapping") def on_set_mapping(data): """Projection mapping: activar/editar y guardar las superficies. Lo usan el escenario (al arrastrar esquinas) y el panel (activar, añadir, borrar, reset). Guarda en data/mapping.json y difunde a todos.""" data = data or {} with lock: m = state["mapping"] if "enabled" in data: m["enabled"] = bool(data["enabled"]) if "edit" in data: m["edit"] = bool(data["edit"]) if "surfaces" in data and isinstance(data["surfaces"], list): m["surfaces"] = data["surfaces"] if "masks" in data and isinstance(data["masks"], list): m["masks"] = data["masks"] m.setdefault("masks", []) save_mapping() broadcast() @socketio.on("engine_command") def on_engine_command(data): action = (data or {}).get("action") if state["engine"] == "projectm": pm = state["projectm"] if action in ("next", "prev"): projectm_key({"next": "n", "prev": "p"}[action]) elif action == "lock": # La tecla espacio congela/suelta el visual en pantalla al # instante, sin relanzar projectM (el titulo pone [locked]). projectm_key("space") with lock: pm["locked"] = not pm.get("locked") broadcast() elif action == "fix_current": # Fija en las listas la categoria y el visual que se ven ahora: # projectM se relanza rotando solo entre variantes de ese look. if pm.get("nowCategory") and pm.get("nowVisual"): with lock: pm["category"] = pm["nowCategory"] pm["preset"] = pm["nowVisual"] pm["variante"] = "" start_projectm() broadcast() else: notify("warn", "Todavia no se que visual esta en pantalla: " "espera un par de segundos y vuelve a probar.") else: socketio.emit("stage_command", {"action": action, "value": (data or {}).get("value")}) @socketio.on("pm_favorito") def on_pm_favorito(data): """Favoritos de projectM: guardar el visual en pantalla, quitar uno, o lanzar uno a dedo (para pinchar visuales en directo).""" data = data or {} action = data.get("action") pm = state["projectm"] if action == "add": fav = {"categoria": pm.get("nowCategory"), "visual": pm.get("nowVisual"), "variante": pm.get("nowPlaying")} if not all(fav.values()): notify("warn", "Todavia no se que visual esta en pantalla: " "espera un par de segundos y vuelve a probar.") return if fav not in pm["favoritos"]: with lock: pm["favoritos"].append(fav) save_favoritos() notify("info", f"Favorito guardado: {fav['variante']}") elif action == "del": fav = {k: data.get(k) for k in ("categoria", "visual", "variante")} with lock: pm["favoritos"] = [f for f in pm["favoritos"] if f != fav] save_favoritos() elif action == "play": with lock: pm["category"] = data.get("categoria") or "" pm["preset"] = data.get("visual") or "" pm["variante"] = data.get("variante") or "" if state["engine"] == "projectm" and state["power"]: start_projectm() broadcast() @socketio.on("stage_meta") def on_stage_meta(data): """El escenario nos envia las listas que solo conoce el navegador.""" data = data or {} with lock: if isinstance(data.get("butterchurnPresets"), list): state["meta"]["butterchurnPresets"] = data["butterchurnPresets"] if isinstance(data.get("audioDevices"), list): state["meta"]["audioDevices"] = data["audioDevices"] if isinstance(data.get("cameraDevices"), list): state["meta"]["cameraDevices"] = data["cameraDevices"] broadcast() @socketio.on("rescan_devices") def on_rescan_devices(): """Pide al escenario que vuelva a buscar microfonos y camaras. Aprovecha para refrescar la lista de pantallas conectadas.""" with lock: state["meta"]["monitors"] = list_monitors() socketio.emit("stage_rescan") broadcast() @socketio.on("reacquire_audio") def on_reacquire_audio(): """Pide al escenario que vuelva a conectar la captura del microfono.""" socketio.emit("stage_reacquire") def pactl_default(kind): """Nombre del sink o source por defecto ('sink' | 'source'), o None.""" try: out = subprocess.check_output(["pactl", f"get-default-{kind}"], text=True, timeout=5).strip() except (OSError, subprocess.SubprocessError): return None return out or None # Aplicaciones cuyos streams de captura enrutamos: el navegador de las # visuales (Chromium/Firefox) y projectM (SDL), que siempre abre la # entrada por defecto y no se puede configurar de otra forma. CAPTURE_APPS = ("chrom", "firefox", "projectm") def capture_streams(): """Indices de los streams de captura de las visuales, leidos de pactl.""" try: out = subprocess.check_output(["pactl", "list", "source-outputs"], text=True, timeout=5) except (OSError, subprocess.SubprocessError): return [] indices, current = [], None for line in out.splitlines(): line = line.strip() if line.startswith("Source Output #"): current = line.split("#", 1)[1] elif current and ("application.process.binary" in line or "application.name" in line): if any(app in line.lower() for app in CAPTURE_APPS): indices.append(current) current = None return indices def route_captures(target, quiet=False, attempts=10): """Mueve los streams de captura al 'monitor' de la salida por defecto (target='monitor': lo que suena en el equipo) o los devuelve a la entrada por defecto (target='mic'). Hace falta porque Chromium no lista los monitores de PulseAudio/PipeWire al enumerar dispositivos y projectM (SDL) captura siempre la entrada por defecto: la unica via fiable es mover sus streams con pactl. Requiere que las visuales corran en esta misma maquina (kiosko o modo portatil).""" if target == "monitor": sink = pactl_default("sink") dest = sink + ".monitor" if sink else None else: dest = pactl_default("source") if not dest: notify("warn", "pactl no pudo averiguar la salida/entrada por " "defecto del sistema. Revisa el audio con 'pactl " "info'; mientras tanto usa el microfono.") return # Los streams tardan un instante en aparecer tras abrir la captura. streams = [] for _ in range(attempts): streams = capture_streams() if streams: break time.sleep(0.3) if not streams: if not quiet: notify("warn", "No hay ningun stream de captura que mover (ni " "navegador ni projectM). Recarga el escenario o " "reinicia FOSFENO y vuelve a intentarlo.") return moved = all( subprocess.run(["pactl", "move-source-output", idx, dest], check=False).returncode == 0 for idx in streams) if not moved: notify("warn", f"pactl no pudo mover la captura a '{dest}'. Si la " "salida acaba de cambiar (p. ej. bluetooth), pulsa " "'Aplicar entrada' otra vez; si sigue fallando, " "revisa 'pactl list sources short'.") elif target == "monitor" and not quiet: notify("info", f"Audio del sistema conectado ({dest}).") @socketio.on("audio_route") def on_audio_route(data): """El escenario pide enrutar la captura: 'monitor' (lo que suena en el equipo) o 'mic' (entrada por defecto).""" target = (data or {}).get("target") if target in ("monitor", "mic"): route_captures(target) @socketio.on("stage_status") def on_stage_status(data): if isinstance(data, dict): with lock: if "label" in data: state["status"]["label"] = data["label"] if "bpm" in data: state["audio"]["bpm"] = data["bpm"] socketio.emit("status", {"label": state["status"]["label"], "bpm": state["audio"]["bpm"]}) @socketio.on("stage_notify") def on_stage_notify(data): """El escenario (navegador) nos comunica un error para mostrarlo en el panel.""" data = data or {} notify(data.get("level", "error"), data.get("message", "Error en el escenario de visuales.")) @socketio.on("rescan_videos") def on_rescan_videos(): with lock: state["meta"]["videos"] = list_videos() broadcast() @socketio.on("system") def on_system(data): action = (data or {}).get("action") if action not in ("reboot", "shutdown"): return # En modo portatil estos botones apagarian el ordenador de trabajo: # solo tienen sentido en la Raspberry del kiosko. if NO_KIOSK: notify("warn", "Modo portatil: el boton de reiniciar/apagar es " "para la Raspberry. En este equipo hazlo a mano.") return if action == "reboot": subprocess.Popen(["sudo", "reboot"]) else: subprocess.Popen(["sudo", "poweroff"]) # -------------------------------------------------------------------------- # Arranque # -------------------------------------------------------------------------- def setup_audio(): """Marca el microfono USB como fuente de audio por defecto (best-effort).""" # En modo portatil no hay micro USB dedicado y la fuente se elige en el # panel (micro o "audio del navegador"). Evitamos forzar nada y el aviso # de "no hay micro USB", que aqui solo confundiria. if NO_KIOSK: return match = CFG.get("audio", {}).get("matchSource", "usb").lower() try: out = subprocess.check_output(["pactl", "list", "short", "sources"], text=True, timeout=5) except (OSError, subprocess.SubprocessError): return for line in out.splitlines(): cols = line.split("\t") if len(cols) >= 2 and match in cols[1].lower() \ and "monitor" not in cols[1].lower(): subprocess.run(["pactl", "set-default-source", cols[1]], check=False) print("[FOSFENO] Microfono por defecto:", cols[1]) return notify("warn", "No se detecto ningun microfono USB. Conecta uno y " "reinicia, o elige la entrada a mano en el panel.") def watchdog(): """Reabre Chromium si se cierra (salvo cuando projectM esta activo).""" while True: time.sleep(10) if refresh_network(): broadcast() if not NO_KIOSK and state["engine"] != "projectm" \ and not is_running(procs["chromium"]): notify("warn", "El navegador de las visuales se cerro. " "Reabriendolo automaticamente.") start_chromium() def check_install(): """Avisa al panel si la instalacion esta incompleta.""" needed = ["socket.io.min.js", "butterchurn.min.js", "butterchurn-presets.min.js", "hydra-synth.js"] missing = [f for f in needed if not (WEB / "lib" / f).exists()] if missing: notify("error", "Faltan librerias web (" + ", ".join(missing) + "). Vuelve a ejecutar install.sh.") if not (WEB / "lib" / "codemirror" / "codemirror.js").exists(): notify("warn", "Falta CodeMirror: el editor de codigo no funcionara. " "Vuelve a ejecutar install.sh.") def main(): binary = CFG.get("projectm", {}).get("binary", "projectMSDL") state["projectm"]["available"] = shutil.which(binary) is not None state["meta"]["projectmPresets"] = list_projectm_presets() state["meta"]["monitors"] = list_monitors() print(f"[FOSFENO] projectM nativo: " f"{'disponible' if state['projectm']['available'] else 'no instalado'}") print(f"[FOSFENO] Videos encontrados: {len(state['meta']['videos'])}") check_install() setup_audio() refresh_network() # En modo portatil el script 'fosfeno' abre el navegador; el servidor no. if not NO_KIOSK: threading.Timer(3.0, start_chromium).start() threading.Thread(target=watchdog, daemon=True).start() threading.Thread(target=watch_projectm_log, daemon=True).start() net = state["network"] suffix = "" if PORT == 80 else f":{PORT}" print("[FOSFENO] ===================================================") print("[FOSFENO] Panel de control disponible en:") print(f"[FOSFENO] http://{net['ip']}{suffix}/") print(f"[FOSFENO] http://{net['hostname']}.local{suffix}/") print("[FOSFENO] ===================================================") socketio.run(app, host=HOST, port=PORT, allow_unsafe_werkzeug=True) if __name__ == "__main__": main()