hacklab/FOSFENO
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
FOSFENO/backend/server.py
hacklab f017246f0e Capas independientes, projectM en el mapping y catalogo de la galeria 
Mapping
- Los clips salian del reves: mapper.js activaba UNPACK_FLIP_Y_WEBGL, pero el
  modelo va con origen arriba-izquierda y texImage2D ya sube la primera fila
  en t=0, asi que el flip la mandaba al final. Comprobado renderizando el
  mapper real en Chromium headless con una fuente mitad roja / mitad azul.
- projectM se puede usar YA en una capa: los .milk de data/presets-projectm se
  traducen a Butterchurn en el navegador al elegirlos (milkdrop-preset-
  converter). La capa MilkDrop gana 'biblioteca' (butter | projectm) sin
  cambiar su firma, para no recrear el contexto WebGL al cambiar de una a otra.
  Medido: 100/100 presets de una muestra convierten, 84/84 de los que llevan
  shaders warp/comp; ~7 ms por preset.
- Cada capa tiene su pestana arriba y su propia vista, y '+ CAPA' crea una y
  entra en ella: con varias capas, ir y volver a MAPPING no era viable.
- Dos capas MilkDrop sin preset ya no salen identicas (cogian el indice 0):
  cada instancia elige uno al azar y lo escribe en el estado.
- Una capa nueva ya no nace en la fuente "motor", que es el lienzo del motor
  activo y hacia que dos capas ensenaran lo mismo.

Galeria de visuales
- scripts/catalogar-visuales.py: ficha de cada clip (pelicula, personajes,
  duracion) y, sobre todo, si es una silueta de verdad y cuanta figura tiene.
  Distingue silueta de corte crudo por el negro puro del fondo: 0,63-0,90
  frente a 0,04-0,06, sin zona gris.
- El panel lista los clips agrupados por pelicula, con nombre legible y aviso
  de los que casi no tienen figura, en vez del nombre del archivo.
- Soporte de siluetas con canal alfa (.webm VP9): transparencia de verdad, sin
  recorte por luminancia. El shader del mapper ya la respeta.

Documentacion
- docs/visuales.md nuevo; pendiente.md al dia con lo hecho y lo que queda.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-02 14:56:01 +02:00

1081 lines
41 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
FOSFENO - Motor de visuales audio-reactivas para Raspberry Pi OS.
Un unico proceso que:
- sirve el PANEL de control (http://<ip>/)
- sirve el ESCENARIO (http://<ip>/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"))
# Clave compartida para la red local. Vacia (por defecto) = abierto, como
# siempre: en casa o en un ensayo no molesta. Si se pone (config.json ->
# auth.token, o la variable FOSFENO_TOKEN), hace falta para conectarse al
# panel y para subir archivos, y el codigo QR del proyector ya la lleva: el
# que escanea entra, el que solo conoce la IP no. Recomendado en local
# publico, donde cualquiera en la wifi podria apagar el equipo o meter
# codigo en las visuales.
TOKEN = os.environ.get("FOSFENO_TOKEN") or CFG.get("auth", {}).get("token", "")
app = Flask(__name__, static_folder=None)
# Tope por archivo subido. Con 4 GB, cualquiera en la red podia llenar la
# tarjeta de la Raspberry en tres subidas; con esto siguen cabiendo clips
# largos de sobra.
app.config["MAX_CONTENT_LENGTH"] = 512 * 1024 ** 2 # 512 MB
# cors_allowed_origins=None => solo el mismo origen, calculado del Host de
# cada peticion (asi vale igual por IP, por fosfeno.local o por localhost).
# Con el comodin "*", una web cualquiera abierta en un movil de la sala podia
# conectarse al socket y mandar ordenes.
socketio = SocketIO(app, async_mode="threading", cors_allowed_origins=None)
# --------------------------------------------------------------------------
# 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 load_visuales():
"""Fichas de la galeria: de que peli sale cada clip, quien aparece y como
de llena esta la silueta. Lo genera scripts/catalogar-visuales.py; si no
esta, el panel sigue funcionando con los nombres de archivo pelados.
Devuelve {archivo: ficha} y solo de lo que sigue en disco, para que un
catalogo viejo no ensene clips borrados."""
ruta = DATA / "visuales.json"
datos = load_json(ruta, {}) if ruta.is_file() else {}
hay = set(list_videos())
return {c["archivo"]: c for c in datos.get("clips", [])
if isinstance(c, dict) and c.get("archivo") in hay}
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": dict(load_json(DATA / "mapping.json",
{"enabled": False, "surfaces": [], "masks": []}),
# 'edit' (los tiradores encima de las propias visuales) es
# un modo de montaje, no un ajuste: arranca siempre
# apagado para que el proyector salga limpio, sin marcos
# de colores ni circulos en las esquinas.
edit=False),
"meta": {
"butterchurnPresets": [],
"audioDevices": [],
"cameraDevices": [],
"videos": list_videos(),
"visuales": load_visuales(),
"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
# El escenario tambien necesita la clave para hablar con el servidor
url = f"http://localhost:{PORT}/stage" + (f"?k={TOKEN}" if TOKEN else "")
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 dentro_de(base, *partes):
"""Ruta compuesta solo si no se sale de 'base'. Categoria, visual y
variante llegan del panel, y con '../..' se podria apuntar a cualquier
carpeta del disco."""
base = Path(base).resolve()
try:
p = base.joinpath(*partes).resolve()
except (OSError, ValueError):
return None
return p if str(p).startswith(str(base) + os.sep) else None
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"):
cat = dentro_de(base, pm["category"])
if cat and cat.is_dir():
path = cat
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 = dentro_de(base, pm["category"], pm["preset"])
if vis and 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 = dentro_de(base, pm["category"], pm["preset"],
pm["variante"] + ".milk")
if origen and 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/<path:fname>")
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/<path:fname>")
def stage_files(fname):
return send_from_directory(str(WEB / "stage"), fname)
@app.route("/lib/<path:fname>")
def lib_files(fname):
return send_from_directory(str(WEB / "lib"), fname)
@app.route("/data/<path:fname>")
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)."""
# Con clave puesta, subir tambien la pide: si no, cualquiera en la wifi
# puede llenar la tarjeta de la Raspberry.
if TOKEN and request.args.get("k") != TOKEN:
return {"ok": False, "error": "No autorizado."}, 403
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()
state["meta"]["visuales"] = load_visuales()
notify("info", f"'{name}' anadido a la galeria del mezclador.")
broadcast()
return {"ok": True, "name": name}
# --------------------------------------------------------------------------
# Eventos WebSocket
# --------------------------------------------------------------------------
@socketio.on("connect")
def on_connect(auth=None):
# Sin la clave no se conecta: ni ordenes, ni codigo en las visuales, ni
# apagar la maquina. Con TOKEN vacio entra todo el mundo, como siempre.
if TOKEN and (auth or {}).get("token") != TOKEN:
return False
socketio.emit("state", state)
return None
@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()
# Claves que cada motor acepta en update_settings. Sin esta lista, cualquiera
# en la red puede meter claves inventadas o cadenas kilometricas en el estado,
# que ademas se difunde entero a todos los clientes en cada cambio.
CLAVES_AJUSTES = {
"butterchurn": {"preset", "shuffle", "interval", "intervalMode", "blendTime"},
"mixer": {"source", "fondo", "camOn", "cameraId", "video", "mix",
"blendMode", "hue", "saturate", "contrast", "brightness",
"colorama", "posterize", "pixelate", "kaleid", "rotate",
"feedback", "invert", "beatPulse"},
"audio": {"device", "bpm", "source"},
"projectm": {"category", "preset", "duration", "shuffle", "monitor",
"variante"},
}
def patch_saneado(engine, patch):
"""Deja solo claves conocidas con valores simples y de tamano razonable."""
permitidas = CLAVES_AJUSTES.get(engine, set())
limpio = {}
for clave, valor in patch.items():
if clave not in permitidas:
continue
if isinstance(valor, str) and len(valor) > 512:
continue
if isinstance(valor, (str, int, float, bool)):
limpio[clave] = valor
return limpio
@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):
patch = patch_saneado(engine, patch)
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()
_map_timer = None
def save_mapping():
"""Guarda el mapping con un segundo de retardo. Arrastrar un punto emite
~16 veces por segundo y no queremos escribir la tarjeta SD 16 veces por
segundo: se reprograma y solo escribe cuando el usuario suelta."""
global _map_timer
if _map_timer:
_map_timer.cancel()
_map_timer = threading.Timer(1.0, _write_mapping)
_map_timer.daemon = True
_map_timer.start()
def _write_mapping():
try:
with lock:
datos = json.dumps(state["mapping"], ensure_ascii=False, indent=2)
with open(DATA / "mapping.json", "w", encoding="utf-8") as fh:
fh.write(datos)
except OSError as e:
notify("warn", f"No se pudo guardar el mapping: {e}")
def capas_saneadas(lista, clave_geo):
"""Descarta lo que no tenga forma de capa/mascara y limita cuantas se
aceptan. Sin esto, cualquiera en la red puede dejar mapping.json lleno de
basura (y persiste entre reinicios)."""
limpio = []
for it in lista[:64]:
if isinstance(it, dict) and isinstance(it.get("id"), str) \
and isinstance(it.get(clave_geo), list):
limpio.append(it)
return limpio
@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"] = capas_saneadas(data["surfaces"], "points")
if "masks" in data and isinstance(data["masks"], list):
m["masks"] = capas_saneadas(data["masks"], "corners")
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(data=None):
"""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(data=None):
"""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(data=None):
with lock:
state["meta"]["videos"] = list_videos()
state["meta"]["visuales"] = load_visuales()
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()
Reference in a new issue View git blame Copy permalink