222 lines
7.5 KiB
Python
222 lines
7.5 KiB
Python
"""API de contrôle des LEDs WS2812b."""
|
||
|
||
import asyncio
|
||
import logging
|
||
from fastapi import APIRouter, Request, Query
|
||
from fastapi.responses import JSONResponse
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = APIRouter()
|
||
|
||
|
||
@router.get("/leds/status")
|
||
async def led_status(request: Request):
|
||
led = request.app.state.led_service
|
||
return {"effect": led.current_effect}
|
||
|
||
|
||
@router.post("/leds/effect")
|
||
async def set_effect(
|
||
request: Request,
|
||
effect: str = Query(..., description="idle | countdown | capture | captured | finished | printing | error | disabled | off"),
|
||
):
|
||
"""Force un effet LED (admin uniquement, pas de vérification auth ici — à protéger via Zoraxy)."""
|
||
valid = ("idle", "countdown", "capture", "captured", "finished", "printing", "error", "disabled", "off")
|
||
if effect not in valid:
|
||
return JSONResponse({"error": f"Effet invalide. Valeurs: {valid}"}, status_code=400)
|
||
|
||
led = request.app.state.led_service
|
||
ws = request.app.state.ws_manager
|
||
led.play(effect)
|
||
await ws.broadcast({"type": "led_effect", "effect": effect})
|
||
return {"ok": True, "effect": effect}
|
||
|
||
|
||
@router.post("/leds/color")
|
||
async def set_color(
|
||
request: Request,
|
||
r: int = Query(default=0, ge=0, le=255),
|
||
g: int = Query(default=0, ge=0, le=255),
|
||
b: int = Query(default=0, ge=0, le=255),
|
||
):
|
||
"""Couleur fixe immédiate sur tout l'anneau."""
|
||
led = request.app.state.led_service
|
||
led.set_color(r, g, b)
|
||
return {"ok": True, "color": [r, g, b]}
|
||
|
||
|
||
@router.post("/leds/off")
|
||
async def leds_off(request: Request):
|
||
led = request.app.state.led_service
|
||
led.play("off")
|
||
return {"ok": True}
|
||
|
||
|
||
@router.get("/leds/brightness")
|
||
async def get_brightness(request: Request):
|
||
cfg = request.app.state.config.leds
|
||
return {"brightness": cfg.brightness, "auto_brightness": cfg.auto_brightness}
|
||
|
||
|
||
@router.put("/leds/brightness")
|
||
async def set_brightness(
|
||
request: Request,
|
||
value: int = Query(..., ge=0, le=255),
|
||
save: bool = Query(default=True),
|
||
):
|
||
"""Regle la luminosite globale (0-255) et l'applique immediatement."""
|
||
led = request.app.state.led_service
|
||
led.set_brightness(value)
|
||
if save:
|
||
config_svc = request.app.state.config_service
|
||
config_svc.save_led_brightness(value)
|
||
return {"ok": True, "brightness": value}
|
||
|
||
|
||
@router.put("/leds/auto-brightness")
|
||
async def set_auto_brightness(
|
||
request: Request,
|
||
enabled: bool = Query(...),
|
||
save: bool = Query(default=True),
|
||
):
|
||
"""Active ou desactive la luminosite automatique du flash selon l'ambiance."""
|
||
cfg = request.app.state.config.leds
|
||
cfg.auto_brightness = enabled
|
||
if save:
|
||
config_svc = request.app.state.config_service
|
||
config_svc.save_led_auto_brightness(enabled)
|
||
return {"ok": True, "auto_brightness": enabled}
|
||
|
||
|
||
@router.get("/leds/effects")
|
||
async def get_effects(request: Request):
|
||
"""Retourne la configuration de tous les effets LED."""
|
||
cfg = request.app.state.config.leds
|
||
return {
|
||
name: {
|
||
"color": effect.get("color", [0, 0, 0]),
|
||
"flash_duration": effect.get("flash_duration", 0.1),
|
||
"flashes": effect.get("flashes", 2),
|
||
"speed": effect.get("speed", 0.05),
|
||
"duration": effect.get("duration", 2.0),
|
||
}
|
||
for name, effect in cfg.effects.items()
|
||
}
|
||
|
||
|
||
@router.get("/leds/ambient-lux")
|
||
async def get_ambient_lux(request: Request):
|
||
"""Retourne la luminosité ambiante mesurée (dernière valeur connue)."""
|
||
led = request.app.state.led_service
|
||
lux = led._ambient_lux
|
||
return {
|
||
"ambient_lux": lux,
|
||
"known": lux >= 0,
|
||
"auto_brightness": request.app.state.config.leds.auto_brightness,
|
||
"description": (
|
||
"sombre (flash plein)" if lux < 50 and lux >= 0 else
|
||
"luminosité moyenne (flash adapté)" if lux < 180 and lux >= 0 else
|
||
"clair (flash réduit)" if lux >= 180 else
|
||
"non mesuré"
|
||
),
|
||
}
|
||
|
||
|
||
@router.post("/leds/measure-ambient")
|
||
async def measure_ambient_lux(request: Request):
|
||
"""Déclenche une mesure de luminosité ambiante via snapshot liveview.
|
||
Utile pour diagnostiquer si le flux liveview est accessible depuis JH-Photomaton.
|
||
"""
|
||
pb = request.app.state.photobooth_service
|
||
led = request.app.state.led_service
|
||
|
||
snapshot = await pb.get_liveview_snapshot()
|
||
if not snapshot:
|
||
return JSONResponse({
|
||
"ok": False,
|
||
"error": "Aucun endpoint liveview accessible sur photobooth-app. "
|
||
"Vérifiez que photobooth-app est démarré et que le flux vidéo est actif.",
|
||
}, status_code=503)
|
||
|
||
try:
|
||
from PIL import Image
|
||
import io
|
||
img = Image.open(io.BytesIO(snapshot)).convert("L").resize((80, 60))
|
||
pixels = list(img.getdata())
|
||
avg = sum(pixels) / len(pixels)
|
||
led.set_ambient_lux(avg)
|
||
return {
|
||
"ok": True,
|
||
"ambient_lux": round(avg, 1),
|
||
"description": (
|
||
"sombre → flash plein (255)" if avg < 50 else
|
||
f"moyen → flash {int(255 - (avg - 50) / 130 * 135)}" if avg < 180 else
|
||
"clair → flash réduit (80)"
|
||
),
|
||
}
|
||
except Exception as e:
|
||
return JSONResponse({"ok": False, "error": str(e)}, status_code=500)
|
||
|
||
|
||
@router.get("/leds/flash-timing")
|
||
async def get_flash_timing(request: Request):
|
||
"""Retourne les paramètres de timing du flash."""
|
||
cfg = request.app.state.config.leds
|
||
return {
|
||
"pre_flash_advance": cfg.pre_flash_advance,
|
||
"countdown_duration": cfg.countdown_duration,
|
||
}
|
||
|
||
|
||
@router.put("/leds/flash-timing")
|
||
async def set_flash_timing(
|
||
request: Request,
|
||
pre_flash_advance: float = Query(..., ge=0.0, le=3.0, description="Secondes avant fin countdown pour démarrer le flash"),
|
||
save: bool = Query(default=True),
|
||
):
|
||
"""Règle combien de secondes avant la fin du countdown le flash démarre (0–3s)."""
|
||
cfg = request.app.state.config.leds
|
||
cfg.pre_flash_advance = pre_flash_advance
|
||
if save:
|
||
config_svc = request.app.state.config_service
|
||
with open(config_svc.path, encoding="utf-8") as f:
|
||
import yaml
|
||
raw = yaml.safe_load(f) or {}
|
||
raw.setdefault("leds", {})["pre_flash_advance"] = pre_flash_advance
|
||
with open(config_svc.path, "w", encoding="utf-8") as f:
|
||
yaml.dump(raw, f, allow_unicode=True, default_flow_style=False)
|
||
return {"ok": True, "pre_flash_advance": pre_flash_advance}
|
||
|
||
|
||
@router.put("/leds/effect/{effect_name}")
|
||
async def update_effect(
|
||
request: Request,
|
||
effect_name: str,
|
||
r: int = Query(default=None, ge=0, le=255),
|
||
g: int = Query(default=None, ge=0, le=255),
|
||
b: int = Query(default=None, ge=0, le=255),
|
||
flash_duration: float = Query(default=None, ge=0.05, le=2.0),
|
||
flashes: int = Query(default=None, ge=1, le=5),
|
||
save: bool = Query(default=True),
|
||
):
|
||
"""Met a jour la configuration d'un effet LED (couleur, timings).
|
||
|
||
Exemple : PUT /api/leds/effect/capture?r=255&g=200&b=80&flash_duration=0.3&flashes=2
|
||
"""
|
||
cfg = request.app.state.config.leds
|
||
config_svc = request.app.state.config_service
|
||
|
||
effect_raw = cfg.effects.setdefault(effect_name, {})
|
||
|
||
if r is not None and g is not None and b is not None:
|
||
effect_raw["color"] = [r, g, b]
|
||
if flash_duration is not None:
|
||
effect_raw["flash_duration"] = flash_duration
|
||
if flashes is not None:
|
||
effect_raw["flashes"] = flashes
|
||
|
||
if save:
|
||
config_svc.save_led_effect(effect_name, effect_raw)
|
||
|
||
return {"ok": True, "effect": effect_name, "config": effect_raw}
|