82 lines
2.4 KiB
Python
82 lines
2.4 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()
|
|
|
|
DEFAULT_COUNTDOWN = 5.0
|
|
|
|
|
|
@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
|
|
|
|
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=DEFAULT_COUNTDOWN)
|
|
if request.app.state.config.leds.auto_brightness:
|
|
pb = request.app.state.photobooth_service
|
|
asyncio.create_task(_analyze_ambient_lux(led, pb))
|
|
|
|
case "capture":
|
|
led.play("capture")
|
|
|
|
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_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)
|