Jules
d719f4c100
- Teste get_summary() pour verifier que le DSLR repond - Reconnecte apres 2 echecs consecutifs - Fallback webcam si DSLR plus disponible Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
566 lines
17 KiB
Python
566 lines
17 KiB
Python
import asyncio
|
|
import base64
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from contextlib import asynccontextmanager
|
|
|
|
import cv2
|
|
|
|
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, UploadFile, File
|
|
from fastapi.responses import FileResponse, JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from backend.config import (
|
|
RACINE, DOSSIER_PHOTOS, DOSSIER_EXPORTS, DOSSIER_OVERLAYS, DOSSIER_FONDS, DOSSIER_ANIMATIONS,
|
|
charger_config, sauvegarder_config, mettre_a_jour_config,
|
|
)
|
|
from backend.camera import camera, GPHOTO2_DISPONIBLE
|
|
try:
|
|
import gphoto2 as gp
|
|
except ImportError:
|
|
gp = None
|
|
from backend.gallery import lister_photos, compter_photos, supprimer_photo, vider_galerie
|
|
from backend.effects import appliquer_filtre, appliquer_overlay, chroma_key, lister_overlays, lister_fonds, FILTRES
|
|
from backend.collage import creer_strip, creer_collage, creer_impression_strip
|
|
from backend.destinations import distribuer_photo, detecter_usb, compteur_restant, reset_compteur, recuperer_booth_password
|
|
from backend.printer import lister_imprimantes, imprimer
|
|
from backend.mailer import envoyer_photo
|
|
from backend.qrcode_gen import generer_qr, qr_galerie
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(message)s")
|
|
log = logging.getLogger("photobooth")
|
|
|
|
# Clients WebSocket connectes
|
|
clients_ws: list[WebSocket] = []
|
|
|
|
|
|
capture_en_cours = False
|
|
_dslr_erreurs = 0
|
|
|
|
async def surveiller_dslr():
|
|
"""Verifie periodiquement si un DSLR est branche et bascule dessus."""
|
|
global _dslr_erreurs
|
|
while True:
|
|
await asyncio.sleep(5)
|
|
if capture_en_cours:
|
|
continue
|
|
try:
|
|
if GPHOTO2_DISPONIBLE:
|
|
dslrs = gp.Camera.autodetect()
|
|
if camera.mode == "gphoto2":
|
|
# Verifier que le DSLR repond encore
|
|
try:
|
|
camera.camera.get_summary()
|
|
_dslr_erreurs = 0
|
|
except Exception:
|
|
_dslr_erreurs += 1
|
|
if _dslr_erreurs >= 2:
|
|
log.warning("DSLR ne repond plus, reconnexion...")
|
|
_dslr_erreurs = 0
|
|
camera.deconnecter()
|
|
if len(dslrs) > 0:
|
|
camera.connecter(source="gphoto2")
|
|
else:
|
|
camera.connecter() # fallback webcam
|
|
elif len(dslrs) > 0:
|
|
log.info(f"DSLR detecte : {dslrs[0][0]}, bascule automatique")
|
|
camera.connecter(source="gphoto2")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
"""Demarrage et arret de l'application."""
|
|
log.info("Demarrage du photobooth")
|
|
camera.connecter()
|
|
task = asyncio.create_task(surveiller_dslr())
|
|
yield
|
|
task.cancel()
|
|
log.info("Arret du photobooth")
|
|
camera.deconnecter()
|
|
|
|
|
|
app = FastAPI(title="Photobooth", lifespan=lifespan)
|
|
|
|
# Servir les fichiers statiques
|
|
app.mount("/assets", StaticFiles(directory=str(RACINE / "frontend" / "assets")), name="assets")
|
|
app.mount("/css", StaticFiles(directory=str(RACINE / "frontend" / "css")), name="css")
|
|
app.mount("/js", StaticFiles(directory=str(RACINE / "frontend" / "js")), name="js")
|
|
app.mount("/data", StaticFiles(directory=str(RACINE / "data")), name="data")
|
|
|
|
|
|
# --- Pages ---
|
|
|
|
@app.get("/")
|
|
async def page_principale():
|
|
return FileResponse(str(RACINE / "frontend" / "index.html"))
|
|
|
|
|
|
# --- API Config ---
|
|
|
|
@app.get("/api/config")
|
|
async def api_config():
|
|
return charger_config()
|
|
|
|
|
|
@app.post("/api/config")
|
|
async def api_config_update(modifications: dict):
|
|
config = mettre_a_jour_config(modifications)
|
|
await diffuser_ws({"type": "config_maj", "config": config})
|
|
return config
|
|
|
|
|
|
# --- API Camera ---
|
|
|
|
@app.post("/api/capturer")
|
|
async def api_capturer():
|
|
global capture_en_cours
|
|
# Verifier le compteur
|
|
etat = compteur_restant()
|
|
if etat["actif"] and etat["restantes"] <= 0:
|
|
return JSONResponse({"erreur": "Limite de photos atteinte"}, status_code=403)
|
|
|
|
capture_en_cours = True
|
|
try:
|
|
chemin = camera.capturer()
|
|
finally:
|
|
capture_en_cours = False
|
|
if chemin is None:
|
|
return JSONResponse({"erreur": "Echec capture"}, status_code=500)
|
|
nom = chemin.name
|
|
|
|
# Distribuer vers les destinations
|
|
distribuer_photo(chemin, imprimee=False)
|
|
|
|
await diffuser_ws({"type": "photo_capturee", "nom": nom})
|
|
return {"nom": nom, "chemin": f"/data/photos/{nom}"}
|
|
|
|
|
|
@app.get("/api/preview")
|
|
async def api_preview():
|
|
donnees = camera.preview()
|
|
if donnees is None:
|
|
return JSONResponse({"erreur": "Preview indisponible"}, status_code=500)
|
|
b64 = base64.b64encode(donnees).decode("ascii")
|
|
return {"image": f"data:image/jpeg;base64,{b64}"}
|
|
|
|
|
|
@app.get("/api/camera/statut")
|
|
async def api_camera_statut():
|
|
return {
|
|
"connectee": camera.connectee,
|
|
"mode": camera.mode,
|
|
"appareils": camera.lister_appareils(),
|
|
}
|
|
|
|
|
|
@app.post("/api/camera/reconnecter")
|
|
async def api_camera_reconnecter(body: dict = {}):
|
|
camera.deconnecter()
|
|
source = body.get("source")
|
|
ok = camera.connecter(source=source)
|
|
return {"connectee": ok, "mode": camera.mode}
|
|
|
|
|
|
# --- API Effets ---
|
|
|
|
@app.get("/api/filtres")
|
|
async def api_filtres():
|
|
return FILTRES
|
|
|
|
|
|
@app.post("/api/filtre")
|
|
async def api_appliquer_filtre(donnees: dict):
|
|
nom_photo = donnees.get("photo", "")
|
|
filtre = donnees.get("filtre", "original")
|
|
chemin = DOSSIER_PHOTOS / nom_photo
|
|
if not chemin.exists():
|
|
return JSONResponse({"erreur": "Photo introuvable"}, status_code=404)
|
|
if filtre == "original":
|
|
return {"nom": nom_photo, "chemin": f"/data/photos/{nom_photo}"}
|
|
chemin_export = appliquer_filtre(chemin, filtre)
|
|
return {"nom": chemin_export.name, "chemin": f"/data/exports/{chemin_export.name}"}
|
|
|
|
|
|
@app.get("/api/overlays")
|
|
async def api_overlays():
|
|
return lister_overlays()
|
|
|
|
|
|
@app.post("/api/overlay")
|
|
async def api_appliquer_overlay(donnees: dict):
|
|
nom_photo = donnees.get("photo", "")
|
|
nom_overlay = donnees.get("overlay", "")
|
|
chemin = DOSSIER_PHOTOS / nom_photo
|
|
if not chemin.exists():
|
|
return JSONResponse({"erreur": "Photo introuvable"}, status_code=404)
|
|
chemin_export = appliquer_overlay(chemin, nom_overlay)
|
|
return {"nom": chemin_export.name, "chemin": f"/data/exports/{chemin_export.name}"}
|
|
|
|
|
|
@app.post("/api/chroma")
|
|
async def api_chroma_key(donnees: dict):
|
|
nom_photo = donnees.get("photo", "")
|
|
nom_fond = donnees.get("fond")
|
|
chemin = DOSSIER_PHOTOS / nom_photo
|
|
if not chemin.exists():
|
|
return JSONResponse({"erreur": "Photo introuvable"}, status_code=404)
|
|
config = charger_config()
|
|
conf_chroma = config.get("chroma_key", {})
|
|
chemin_export = chroma_key(
|
|
chemin, nom_fond,
|
|
couleur_cle=conf_chroma.get("couleur", "#00ff00"),
|
|
tolerance=conf_chroma.get("tolerance", 40),
|
|
)
|
|
return {"nom": chemin_export.name, "chemin": f"/data/exports/{chemin_export.name}"}
|
|
|
|
|
|
@app.get("/api/fonds")
|
|
async def api_fonds():
|
|
return lister_fonds()
|
|
|
|
|
|
@app.delete("/api/fonds/{nom}")
|
|
async def api_supprimer_fond(nom: str):
|
|
chemin = DOSSIER_FONDS / nom
|
|
if chemin.exists() and chemin.parent == DOSSIER_FONDS:
|
|
chemin.unlink()
|
|
return {"succes": True}
|
|
return JSONResponse({"erreur": "Fichier introuvable"}, status_code=404)
|
|
|
|
|
|
# --- API Collage ---
|
|
|
|
@app.post("/api/strip")
|
|
async def api_strip(donnees: dict):
|
|
noms = donnees.get("photos", [])
|
|
chemins = [DOSSIER_PHOTOS / n for n in noms]
|
|
for c in chemins:
|
|
if not c.exists():
|
|
return JSONResponse({"erreur": f"Photo introuvable : {c.name}"}, status_code=404)
|
|
chemin_strip = creer_strip(chemins)
|
|
# Creer aussi la page d'impression (2 bandes sur 10x15 paysage)
|
|
chemin_print = creer_impression_strip(chemin_strip)
|
|
return {
|
|
"nom": chemin_strip.name,
|
|
"chemin": f"/data/exports/{chemin_strip.name}",
|
|
"impression": chemin_print.name,
|
|
"chemin_impression": f"/data/exports/{chemin_print.name}",
|
|
}
|
|
|
|
|
|
@app.post("/api/collage")
|
|
async def api_collage(donnees: dict):
|
|
noms = donnees.get("photos", [])
|
|
colonnes = donnees.get("colonnes", 2)
|
|
chemins = [DOSSIER_PHOTOS / n for n in noms]
|
|
for c in chemins:
|
|
if not c.exists():
|
|
return JSONResponse({"erreur": f"Photo introuvable : {c.name}"}, status_code=404)
|
|
chemin = creer_collage(chemins, colonnes=colonnes)
|
|
return {"nom": chemin.name, "chemin": f"/data/exports/{chemin.name}"}
|
|
|
|
|
|
|
|
# --- API Impression ---
|
|
|
|
@app.get("/api/imprimantes")
|
|
async def api_imprimantes():
|
|
return lister_imprimantes()
|
|
|
|
|
|
@app.post("/api/imprimer")
|
|
async def api_imprimer(donnees: dict):
|
|
nom = donnees.get("photo", "")
|
|
copies = donnees.get("copies", 1)
|
|
# Limiter au max configure
|
|
config = charger_config()
|
|
copies_max = config.get("impression", {}).get("copies_max", 5)
|
|
copies = max(1, min(copies, copies_max))
|
|
# Chercher dans exports d'abord, puis photos
|
|
chemin = DOSSIER_EXPORTS / nom
|
|
if not chemin.exists():
|
|
chemin = DOSSIER_PHOTOS / nom
|
|
if not chemin.exists():
|
|
return JSONResponse({"erreur": "Photo introuvable"}, status_code=404)
|
|
ok = imprimer(chemin, copies=copies)
|
|
if ok:
|
|
distribuer_photo(chemin, imprimee=True)
|
|
return {"succes": ok}
|
|
|
|
|
|
# --- API Booth (galerie live) ---
|
|
|
|
@app.get("/api/booth/info")
|
|
async def api_booth_info():
|
|
return recuperer_booth_password()
|
|
|
|
|
|
# --- API Email ---
|
|
|
|
@app.post("/api/email")
|
|
async def api_email(donnees: dict):
|
|
email_dest = donnees.get("email", "")
|
|
nom = donnees.get("photo", "")
|
|
if not email_dest:
|
|
return JSONResponse({"erreur": "Email requis"}, status_code=400)
|
|
chemin = DOSSIER_EXPORTS / nom
|
|
if not chemin.exists():
|
|
chemin = DOSSIER_PHOTOS / nom
|
|
if not chemin.exists():
|
|
return JSONResponse({"erreur": "Photo introuvable"}, status_code=404)
|
|
ok = envoyer_photo(email_dest, chemin)
|
|
return {"succes": ok}
|
|
|
|
|
|
# --- API QR Code ---
|
|
|
|
@app.get("/api/qr")
|
|
async def api_qr(url: str = ""):
|
|
"""Genere un QR code pour une URL arbitraire."""
|
|
if not url:
|
|
return JSONResponse({"erreur": "URL requise"}, status_code=400)
|
|
import qrcode
|
|
import io
|
|
qr = qrcode.make(url, box_size=6, border=2)
|
|
buf = io.BytesIO()
|
|
qr.save(buf, format="PNG")
|
|
buf.seek(0)
|
|
from fastapi.responses import StreamingResponse
|
|
return StreamingResponse(buf, media_type="image/png")
|
|
|
|
|
|
@app.get("/api/qr/galerie")
|
|
async def api_qr_galerie():
|
|
chemin = qr_galerie()
|
|
if chemin is None:
|
|
return JSONResponse({"erreur": "URL galerie non configuree"}, status_code=400)
|
|
return {"chemin": f"/data/exports/{chemin.name}"}
|
|
|
|
|
|
# --- API Galerie ---
|
|
|
|
@app.get("/api/galerie")
|
|
async def api_galerie():
|
|
return lister_photos("exports")
|
|
|
|
|
|
@app.get("/api/galerie/compteur")
|
|
async def api_compteur():
|
|
return compter_photos()
|
|
|
|
|
|
@app.delete("/api/galerie/{nom}")
|
|
async def api_supprimer_photo(nom: str):
|
|
ok = supprimer_photo(nom)
|
|
return {"succes": ok}
|
|
|
|
|
|
@app.post("/api/galerie/vider")
|
|
async def api_vider_galerie():
|
|
vider_galerie()
|
|
return {"succes": True}
|
|
|
|
|
|
# --- API Upload overlay/fond ---
|
|
|
|
@app.post("/api/upload/overlay")
|
|
async def api_upload_overlay(fichier: UploadFile = File(...)):
|
|
chemin = DOSSIER_OVERLAYS / fichier.filename
|
|
with open(chemin, "wb") as f:
|
|
f.write(await fichier.read())
|
|
return {"nom": fichier.filename}
|
|
|
|
|
|
@app.post("/api/upload/fond")
|
|
async def api_upload_fond(fichier: UploadFile = File(...)):
|
|
chemin = DOSSIER_FONDS / fichier.filename
|
|
with open(chemin, "wb") as f:
|
|
f.write(await fichier.read())
|
|
return {"nom": fichier.filename}
|
|
|
|
|
|
# --- API Compteur ---
|
|
|
|
@app.get("/api/compteur")
|
|
async def api_compteur_etat():
|
|
return compteur_restant()
|
|
|
|
|
|
@app.post("/api/compteur/reset")
|
|
async def api_compteur_reset():
|
|
reset_compteur()
|
|
return compteur_restant()
|
|
|
|
|
|
# --- API Destinations ---
|
|
|
|
@app.get("/api/usb/detecter")
|
|
async def api_detecter_usb():
|
|
return detecter_usb()
|
|
|
|
|
|
# --- API Cadres actifs ---
|
|
|
|
@app.get("/api/cadres")
|
|
async def api_cadres():
|
|
config = charger_config()
|
|
tous = lister_overlays()
|
|
actifs = config.get("cadres", {}).get("actifs", [])
|
|
return {"tous": tous, "actifs": actifs}
|
|
|
|
|
|
@app.post("/api/cadres")
|
|
async def api_cadres_update(donnees: dict):
|
|
actifs = donnees.get("actifs", [])
|
|
mettre_a_jour_config({"cadres": {"actifs": actifs}})
|
|
return {"actifs": actifs}
|
|
|
|
|
|
# --- API Animations compte a rebours ---
|
|
|
|
ANIMATIONS_CAR = {
|
|
"classique": "Classique (chiffres simples)",
|
|
"explosion": "Explosion (chiffres qui eclatent)",
|
|
"rebond": "Rebond (chiffres qui rebondissent)",
|
|
"rotation": "Rotation 3D",
|
|
"fondu": "Fondu enchaine",
|
|
"emoji": "Emoji fun",
|
|
"shake": "Tremblement",
|
|
}
|
|
|
|
@app.get("/api/animations")
|
|
async def api_animations():
|
|
return ANIMATIONS_CAR
|
|
|
|
|
|
@app.get("/api/animations/custom")
|
|
async def api_animations_custom():
|
|
"""Liste les animations personnalisees (GIF/video)."""
|
|
extensions = {".gif", ".mp4", ".webm"}
|
|
fichiers = []
|
|
if DOSSIER_ANIMATIONS.exists():
|
|
for f in sorted(DOSSIER_ANIMATIONS.iterdir()):
|
|
if f.suffix.lower() in extensions:
|
|
fichiers.append(f.name)
|
|
config = charger_config()
|
|
actives = config.get("animations_custom", {}).get("actives", [])
|
|
return {"tous": fichiers, "actives": actives}
|
|
|
|
|
|
@app.post("/api/animations/custom")
|
|
async def api_animations_custom_update(donnees: dict):
|
|
actives = donnees.get("actives", [])
|
|
mettre_a_jour_config({"animations_custom": {"actives": actives}})
|
|
return {"actives": actives}
|
|
|
|
|
|
@app.post("/api/upload/animation")
|
|
async def api_upload_animation(fichier: UploadFile = File(...)):
|
|
chemin = DOSSIER_ANIMATIONS / fichier.filename
|
|
with open(chemin, "wb") as f:
|
|
f.write(await fichier.read())
|
|
return {"nom": fichier.filename}
|
|
|
|
|
|
@app.delete("/api/animations/custom/{nom}")
|
|
async def api_supprimer_animation(nom: str):
|
|
chemin = DOSSIER_ANIMATIONS / nom
|
|
if chemin.exists() and chemin.parent == DOSSIER_ANIMATIONS:
|
|
chemin.unlink()
|
|
return {"succes": True}
|
|
return JSONResponse({"erreur": "Fichier introuvable"}, status_code=404)
|
|
|
|
|
|
# --- API Systeme ---
|
|
|
|
@app.post("/api/systeme/redemarrer")
|
|
async def api_redemarrer():
|
|
import subprocess
|
|
log.warning("Redemarrage systeme demande")
|
|
subprocess.Popen(["sudo", "reboot"])
|
|
return {"succes": True}
|
|
|
|
|
|
@app.post("/api/systeme/eteindre")
|
|
async def api_eteindre():
|
|
import subprocess
|
|
log.warning("Extinction systeme demandee")
|
|
subprocess.Popen(["sudo", "shutdown", "-h", "now"])
|
|
return {"succes": True}
|
|
|
|
|
|
@app.post("/api/systeme/quitter-navigateur")
|
|
async def api_quitter_navigateur():
|
|
import subprocess
|
|
log.warning("Fermeture navigateur demandee")
|
|
subprocess.Popen(["pkill", "-f", "firefox"])
|
|
subprocess.Popen(["pkill", "-f", "chromium"])
|
|
return {"succes": True}
|
|
|
|
|
|
# --- WebSocket ---
|
|
|
|
@app.websocket("/ws")
|
|
async def websocket_endpoint(ws: WebSocket):
|
|
await ws.accept()
|
|
clients_ws.append(ws)
|
|
log.info(f"Client WS connecte ({len(clients_ws)} clients)")
|
|
try:
|
|
while True:
|
|
data = await ws.receive_text()
|
|
msg = json.loads(data)
|
|
await traiter_message_ws(msg, ws)
|
|
except WebSocketDisconnect:
|
|
clients_ws.remove(ws)
|
|
log.info(f"Client WS deconnecte ({len(clients_ws)} clients)")
|
|
|
|
|
|
async def traiter_message_ws(msg: dict, ws: WebSocket):
|
|
"""Traite les messages WebSocket entrants."""
|
|
type_msg = msg.get("type", "")
|
|
|
|
if type_msg == "ping":
|
|
await ws.send_json({"type": "pong"})
|
|
elif type_msg == "preview":
|
|
# Pendant une capture DSLR, utiliser directement la webcam fallback
|
|
donnees = None
|
|
if capture_en_cours and camera._preview_webcam_fallback:
|
|
ret, frame = camera._preview_webcam_fallback.read()
|
|
if ret:
|
|
_, buf = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 70])
|
|
donnees = buf.tobytes()
|
|
else:
|
|
donnees = camera.preview()
|
|
if donnees:
|
|
b64 = base64.b64encode(donnees).decode("ascii")
|
|
await ws.send_json({"type": "preview", "image": f"data:image/jpeg;base64,{b64}"})
|
|
|
|
|
|
async def diffuser_ws(message: dict):
|
|
"""Envoie un message a tous les clients WebSocket."""
|
|
deconnectes = []
|
|
for ws in clients_ws:
|
|
try:
|
|
await ws.send_json(message)
|
|
except Exception:
|
|
deconnectes.append(ws)
|
|
for ws in deconnectes:
|
|
clients_ws.remove(ws)
|
|
|
|
|
|
# Point d'entree
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
config = charger_config()
|
|
conf_srv = config.get("serveur", {})
|
|
uvicorn.run(
|
|
"backend.main:app",
|
|
host=conf_srv.get("host", "0.0.0.0"),
|
|
port=conf_srv.get("port", 8080),
|
|
reload=False,
|
|
log_level="info",
|
|
)
|