105 lines
3.5 KiB
Python
105 lines
3.5 KiB
Python
"""Webhooks recus de photobooth-app via plugin_commander."""
|
|
|
|
import asyncio
|
|
import logging
|
|
|
|
from fastapi import APIRouter, Request, Query
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/webhook/photobooth")
|
|
async def photobooth_webhook(
|
|
request: Request,
|
|
event_key: str = Query(default=""),
|
|
mediaitem_type: str = Query(default=""),
|
|
):
|
|
led = request.app.state.led_service
|
|
btn = request.app.state.button_service
|
|
ws = request.app.state.ws_manager
|
|
cfg = request.app.state.config.leds
|
|
|
|
logger.info("Webhook photobooth: event=%s type=%s", event_key, mediaitem_type)
|
|
await ws.broadcast({"type": "photobooth_event", "event": event_key, "media_type": mediaitem_type})
|
|
|
|
match event_key:
|
|
case "counting":
|
|
led.play("countdown", countdown_duration=cfg.countdown_duration)
|
|
|
|
# Annuler le pre-flash précédent si toujours en attente
|
|
prev = getattr(request.app.state, "pre_flash_task", None)
|
|
if prev and not prev.done():
|
|
prev.cancel()
|
|
|
|
# Déclencher le flash X secondes avant la fin du countdown
|
|
delay = max(0.0, cfg.countdown_duration - cfg.pre_flash_advance)
|
|
task = asyncio.create_task(_delayed_flash(led, delay))
|
|
request.app.state.pre_flash_task = task
|
|
logger.info(
|
|
"Pre-flash schedulé dans %.2fs (countdown=%.1fs advance=%.1fs)",
|
|
delay, cfg.countdown_duration, cfg.pre_flash_advance,
|
|
)
|
|
|
|
if cfg.auto_brightness:
|
|
pb = request.app.state.photobooth_service
|
|
asyncio.create_task(_analyze_ambient_lux(led, pb))
|
|
|
|
case "capture":
|
|
# Flash déjà lancé par le pre-flash depuis counting ; ne pas redémarrer
|
|
logger.debug("Event capture recu (pre-flash déjà actif)")
|
|
|
|
case "captured":
|
|
led.play("captured")
|
|
|
|
case "finished":
|
|
led.play("finished")
|
|
asyncio.create_task(_delayed_relay_on(btn, 2.5))
|
|
event_svc = getattr(request.app.state, "event_service", None)
|
|
if event_svc:
|
|
slug = request.app.state.config.event.slug
|
|
asyncio.create_task(event_svc.increment(slug, "photos_taken"))
|
|
|
|
case "start":
|
|
led.play("idle")
|
|
|
|
case "stop":
|
|
led.play("off")
|
|
|
|
case _:
|
|
logger.debug("Evenement inconnu: %s", event_key)
|
|
|
|
return {"ok": True, "event": event_key}
|
|
|
|
|
|
async def _delayed_flash(led, delay: float):
|
|
"""Attend `delay` secondes puis déclenche l'effet flash."""
|
|
if delay > 0:
|
|
await asyncio.sleep(delay)
|
|
led.play("capture")
|
|
logger.info("Pre-flash déclenché (délai=%.2fs)", delay)
|
|
|
|
|
|
async def _delayed_relay_on(btn, delay: float):
|
|
await asyncio.sleep(delay)
|
|
if btn:
|
|
btn.relay_on()
|
|
|
|
|
|
async def _analyze_ambient_lux(led, pb):
|
|
"""Capture une frame du liveview et calcule la luminosite moyenne."""
|
|
try:
|
|
snapshot = await pb.get_liveview_snapshot()
|
|
if not snapshot:
|
|
logger.debug("Analyse lux: aucun snapshot disponible")
|
|
return
|
|
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)
|
|
logger.info("Luminosite ambiante: %.1f / 255", avg)
|
|
except Exception as e:
|
|
logger.warning("Analyse lux echouee: %s", e)
|