import asyncio import base64 import json import logging import threading import time from pathlib import Path from contextlib import asynccontextmanager import cv2 from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect, UploadFile, File from fastapi.responses import FileResponse, JSONResponse, HTMLResponse from fastapi.staticfiles import StaticFiles from backend.config import ( RACINE, DOSSIER_PHOTOS, DOSSIER_EXPORTS, DOSSIER_OVERLAYS, DOSSIER_FONDS, DOSSIER_ANIMATIONS, DOSSIER_CADRES, FORMATS_CADRES, 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, lister_cadres, FILTRES from backend.collage import creer_strip, creer_collage 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 # --- Preview push : thread de fond qui capture en continu --- _derniere_frame_preview: bytes | None = None # derniere frame JPEG capturee _preview_actif = False # True quand au moins un client veut le preview _preview_lock = threading.Lock() _loop_principal: asyncio.AbstractEventLoop | None = None # loop asyncio principal _preview_none_count = 0 _PREVIEW_GRACE_FRAMES = 15 # ~500ms avant de declarer une erreur def _thread_preview(): """Thread de fond : capture les frames DSLR en continu. Tourne TOUJOURS quand le DSLR est connecte pour maintenir le miroir leve.""" global _derniere_frame_preview while True: if capture_en_cours or camera.mode != "gphoto2" or not camera.connectee: time.sleep(0.05) continue try: donnees = camera.preview() with _preview_lock: _derniere_frame_preview = donnees except Exception: with _preview_lock: _derniere_frame_preview = None time.sleep(0.033) # ~30 fps max async def _pusher_preview(): """Tache asyncio : pousse la derniere frame a tous les clients WS abonnes.""" global _preview_actif, _preview_none_count while True: await asyncio.sleep(0.05) # 20 fps max envoye aux clients if not _preview_actif: _preview_none_count = 0 continue if not clients_ws: _preview_actif = False continue with _preview_lock: donnees = _derniere_frame_preview if donnees is None: _preview_none_count += 1 # Grace period : attendre plusieurs frames avant de signaler l'erreur # Ne pas signaler pendant une capture (race condition HTTP/WS) if camera.mode == "gphoto2" and _preview_none_count > _PREVIEW_GRACE_FRAMES and not capture_en_cours: await diffuser_ws({"type": "camera_erreur", "message": "DSLR ne repond pas au preview"}) _preview_actif = False _preview_none_count = 0 continue _preview_none_count = 0 loop = asyncio.get_event_loop() b64 = await loop.run_in_executor(None, lambda d=donnees: base64.b64encode(d).decode("ascii")) await diffuser_ws({"type": "preview", "image": f"data:image/jpeg;base64,{b64}"}) async def surveiller_dslr(): """Surveille le DSLR en continu : detecte les deconnexions et reconnecte automatiquement.""" global _dslr_erreurs while True: await asyncio.sleep(5) if capture_en_cours: continue try: if not GPHOTO2_DISPONIBLE: continue if camera.mode == "gphoto2" and camera.connectee: # Le thread preview capture en permanence — preview_dslr_ok reflète la santé réelle if camera.preview_dslr_ok: _dslr_erreurs = 0 else: _dslr_erreurs += 1 if _dslr_erreurs >= 3: # 3 x 5s = 15s sans preview valide log.warning("DSLR ne repond plus (preview KO depuis 15s), reconnexion forcee...") _dslr_erreurs = 0 camera.deconnecter() await diffuser_ws({"type": "camera_erreur", "message": "Appareil photo deconnecte"}) await asyncio.sleep(2) # Laisser le bus USB se reinitialiser camera.connecter(source="gphoto2") if camera.connectee: log.info("DSLR reconnecte avec succes") await diffuser_ws({"type": "camera_ok"}) else: log.warning("Echec reconnexion DSLR — nouvelle tentative dans 5s") else: # Non connecte : chercher et connecter un DSLR disponible dslrs = gp.Camera.autodetect() if len(dslrs) > 0: log.info(f"DSLR detecte : {dslrs[0][0]}, connexion automatique...") camera.connecter(source="gphoto2") if camera.connectee: log.info("DSLR connecte avec succes") await diffuser_ws({"type": "camera_ok"}) else: log.warning("Echec connexion DSLR detecte") except Exception as e: log.debug(f"surveiller_dslr: {e}") @asynccontextmanager async def lifespan(app: FastAPI): """Demarrage et arret de l'application.""" global _loop_principal log.info("Demarrage du photobooth") _loop_principal = asyncio.get_event_loop() camera.connecter() # Thread de capture preview (bloquant, tourne en parallele) t = threading.Thread(target=_thread_preview, daemon=True) t.start() task_dslr = asyncio.create_task(surveiller_dslr()) task_push = asyncio.create_task(_pusher_preview()) yield task_dslr.cancel() task_push.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("/admin", include_in_schema=False) async def page_admin(): return FileResponse(str(RACINE / "frontend" / "index.html")) @app.get("/") async def page_principale(request: Request): from fastapi.responses import RedirectResponse host = request.client.host if request.client else "" if host not in ("127.0.0.1", "::1", "localhost"): return RedirectResponse(url="/admin", status_code=302) return FileResponse(str(RACINE / "frontend" / "index.html")) @app.post("/api/sessions/{code}/upload") async def api_session_upload(code: str, fichier: UploadFile = File(...)): """Upload d'une photo dans une session identifiée par code à 9 chiffres.""" if not code.isdigit() or len(code) != 9: return JSONResponse({"erreur": "Code invalide"}, status_code=400) dossier = DOSSIER_PHOTOS / f"session_{code}" dossier.mkdir(parents=True, exist_ok=True) dest = dossier / fichier.filename dest.write_bytes(await fichier.read()) return {"ok": True, "fichier": fichier.filename} @app.get("/s/{code}", response_class=HTMLResponse) async def galerie_session(code: str): """Galerie mobile pour une session photo (QR code).""" if not code.isdigit() or len(code) != 9: return HTMLResponse("

Code invalide

", status_code=404) dossier = DOSSIER_PHOTOS / f"session_{code}" cfg = charger_config() nom_event = cfg.get("evenement", {}).get("nom", "Photobooth") photos = sorted(dossier.glob("*.jpg"), key=lambda p: p.stat().st_mtime, reverse=True) if dossier.exists() else [] items = "" for p in photos: items += f'
\n' if not items: items = '

Les photos apparaîtront ici après chaque prise 📷

' html = f""" {nom_event} — Galerie

📷 {nom_event}

Code session : {code}

{items}
""" return HTMLResponse(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 _preview_actif = False # Eviter faux camera_erreur si race condition HTTP avant preview_stop try: loop = asyncio.get_event_loop() chemin = await loop.run_in_executor(None, 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/systeme/info") async def api_systeme_info(): import socket import subprocess import shutil # Interfaces reseau avec leurs IPs interfaces = [] try: import netifaces for iface in netifaces.interfaces(): addrs = netifaces.ifaddresses(iface) ipv4 = addrs.get(netifaces.AF_INET, []) ipv6 = addrs.get(netifaces.AF_INET6, []) ips = [a['addr'] for a in ipv4 if a.get('addr') != '127.0.0.1'] ips += [a['addr'].split('%')[0] for a in ipv6 if not a.get('addr', '').startswith('::1')] if ips: interfaces.append({"nom": iface, "ips": ips}) except ImportError: # Fallback sans netifaces try: out = subprocess.check_output(["ip", "-o", "addr", "show"], text=True) for ligne in out.splitlines(): parts = ligne.split() if len(parts) >= 4: iface = parts[1] ip = parts[3].split('/')[0] if ip not in ('127.0.0.1', '::1') and not ip.startswith('fe80'): existing = next((x for x in interfaces if x['nom'] == iface), None) if existing: existing['ips'].append(ip) else: interfaces.append({"nom": iface, "ips": [ip]}) except Exception: pass # Hostname hostname = socket.gethostname() # Uptime uptime_str = "" try: with open('/proc/uptime') as f: secs = float(f.read().split()[0]) h, rem = divmod(int(secs), 3600) m = rem // 60 uptime_str = f"{h}h {m}min" except Exception: pass # Espace disque (partition racine) disque = {} try: total, used, free = shutil.disk_usage("/") disque = { "total": round(total / 1e9, 1), "utilise": round(used / 1e9, 1), "libre": round(free / 1e9, 1), } except Exception: pass # Memoire RAM ram = {} try: with open('/proc/meminfo') as f: lignes = {l.split(':')[0]: l.split(':')[1].strip() for l in f.readlines()} total_kb = int(lignes.get('MemTotal', '0 kB').split()[0]) libre_kb = int(lignes.get('MemAvailable', '0 kB').split()[0]) utilise_kb = total_kb - libre_kb ram = { "total": round(total_kb / 1e6, 1), "utilise": round(utilise_kb / 1e6, 1), "libre": round(libre_kb / 1e6, 1), } except Exception: pass # Temperature CPU (RPi / Linux) temp = None try: with open('/sys/class/thermal/thermal_zone0/temp') as f: temp = round(int(f.read().strip()) / 1000, 1) except Exception: pass # Version Python et app import sys python_version = f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}" return { "hostname": hostname, "interfaces": interfaces, "uptime": uptime_str, "disque": disque, "ram": ram, "temperature_cpu": temp, "python": python_version, "camera_mode": camera.mode, "camera_connectee": camera.connectee, } @app.get("/api/camera/statut") async def api_camera_statut(): return { "connectee": camera.connectee, "mode": camera.mode, "appareils": camera.lister_appareils(), } @app.get("/api/camera/debug") async def api_camera_debug(): """Diagnostic complet de la caméra (preview, viewfinder, erreurs gphoto2).""" info = { "mode": camera.mode, "connectee": camera.connectee, "preview_actif": _preview_actif, "derniere_frame_ok": _derniere_frame_preview is not None, "preview_none_count": _preview_none_count, "capture_en_cours": capture_en_cours, } if camera.mode == "gphoto2" and camera.connectee: loop = asyncio.get_event_loop() try: frame = await loop.run_in_executor(None, camera._preview_gphoto2) info["test_preview"] = "ok" if frame else "retourne None" info["test_preview_taille"] = len(frame) if frame else 0 except Exception as e: info["test_preview"] = f"ERREUR : {e}" try: cfg_names = [] with camera._gp_lock: cfg = camera.camera.get_config() for i in range(cfg.count_children()): child = cfg.get_child(i) cfg_names.append(child.get_name()) info["config_widgets_top"] = cfg_names except Exception as e: info["config_widgets_top"] = f"ERREUR : {e}" return info @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", []) source = donnees.get("source", "photos") dossier_src = DOSSIER_EXPORTS if source == "exports" else DOSSIER_PHOTOS chemins = [] for n in noms: c = dossier_src / n if not c.exists(): c = DOSSIER_EXPORTS / n # fallback if not c.exists(): return JSONResponse({"erreur": f"Photo introuvable : {n}"}, status_code=404) chemins.append(c) chemin_strip = creer_strip(chemins) return { "nom": chemin_strip.name, "chemin": f"/data/exports/{chemin_strip.name}", # La strip elle-même est envoyée avec format 10x15-2up : # le K60 duplique et coupe automatiquement via -div2 "impression": chemin_strip.name, "format_impression": "10x15-2up", } @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.get("/api/imprimante/statut-detail") async def api_statut_imprimante(): """Retourne le statut détaillé CUPS + dernière erreur du log.""" import subprocess config = charger_config() nom = config.get("impression", {}).get("imprimante", "Mitsubishi") # Statut lpstat r = subprocess.run(["lpstat", "-p", nom], capture_output=True, text=True) statut_ligne = r.stdout.strip() # Dernière erreur dans error_log try: r2 = subprocess.run( ["sudo", "grep", f"\\[{nom}\\]\\|Job.*cancel\\|media.*match\\|jam\\|paper", "/var/log/cups/error_log"], capture_output=True, text=True ) lignes = [l for l in r2.stdout.splitlines() if "error_log" not in l] derniere_erreur = lignes[-1] if lignes else "" except Exception: derniere_erreur = "" return { "statut": statut_ligne, "derniere_erreur": derniere_erreur, "imprimante": nom, } @app.post("/api/imprimante/evacuer") async def api_evacuer_bourrage(): """Annule tous les jobs, réactive l'imprimante, retourne le dernier message d'erreur.""" import subprocess config = charger_config() nom = config.get("impression", {}).get("imprimante", "Mitsubishi") subprocess.run(["sudo", "cancel", "-a", nom], capture_output=True) subprocess.run(["sudo", "cupsenable", nom], capture_output=True) # Lire la dernière erreur CUPS pour informer l'utilisateur try: r = subprocess.run( ["sudo", "tail", "-50", "/var/log/cups/error_log"], capture_output=True, text=True ) erreurs = [l for l in r.stdout.splitlines() if any(k in l.lower() for k in ["media", "match", "jam", "paper", "cancel", "error"]) and "createprofile" not in l.lower()] derniere = erreurs[-1] if erreurs else "" except Exception: derniere = "" if "media does not match" in derniere.lower(): msg = ("Format papier incorrect — vérifiez que le format configuré " "correspond à la cassette dans l'imprimante") elif "jam" in derniere.lower(): msg = "Bourrage papier — retirez le papier bloqué physiquement" elif derniere: msg = f"Jobs annulés. Dernière erreur : {derniere.split(']')[-1].strip()}" else: msg = f"Jobs annulés, {nom} réactivée" return {"succes": True, "message": msg} @app.post("/api/imprimante/couper") async def api_couper_papier(): """Déclenche la coupe en envoyant un micro-job blanc (la K60 coupe après chaque impression).""" import subprocess, tempfile, os from PIL import Image config = charger_config() nom = config.get("impression", {}).get("imprimante", "Mitsubishi") try: # Image blanche 1x15cm (la plus petite possible pour la K60) img = Image.new("RGB", (600, 1800), (255, 255, 255)) tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) img.save(tmp.name, "JPEG", quality=95, dpi=(300, 300)) tmp.close() r = subprocess.run( ["lp", "-d", nom, "-n", "1", "-o", "PageSize=w288h432-div2", "-o", "StpiShrinkOutput=Crop", tmp.name], capture_output=True, text=True, timeout=15 ) os.unlink(tmp.name) if r.returncode == 0: return {"succes": True, "message": "Coupe lancée — le papier va être coupé"} return {"succes": False, "message": f"Erreur : {r.stderr.strip() or r.stdout.strip()}"} except Exception as e: return {"succes": False, "message": str(e)} @app.post("/api/imprimante/reset-usb") async def api_reset_usb(): """Réinitialise la connexion USB de l'imprimante (unbind/rebind driver).""" import subprocess, glob try: # Trouver le device USB Mitsubishi K60 (VID 06d3) devs = glob.glob("/sys/bus/usb/devices/*/idVendor") path = None for f in devs: if open(f).read().strip() == "06d3": path = f.replace("/idVendor", "") break if not path: return {"succes": False, "message": "Imprimante USB non trouvée — vérifiez la connexion"} dev_id = path.split("/")[-1] # Annuler les jobs d'abord nom = charger_config().get("impression", {}).get("imprimante", "Mitsubishi") subprocess.run(["sudo", "cancel", "-a", nom], capture_output=True) # Unbind/rebind USB subprocess.run(["sudo", "sh", "-c", f"echo '{dev_id}' > /sys/bus/usb/drivers/usb/unbind"], capture_output=True) await asyncio.sleep(1) subprocess.run(["sudo", "sh", "-c", f"echo '{dev_id}' > /sys/bus/usb/drivers/usb/bind"], capture_output=True) await asyncio.sleep(2) subprocess.run(["sudo", "cupsenable", nom], capture_output=True) return {"succes": True, "message": "Imprimante réinitialisée — prête dans quelques secondes"} except Exception as e: return {"succes": False, "message": str(e)} @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) cadre_override = donnees.get("cadre") format_papier = donnees.get("format_papier") or None resultat = imprimer(chemin, copies=copies, cadre_override=cadre_override, format_papier=format_papier) if resultat.get("succes"): distribuer_photo(chemin, imprimee=True) return resultat # --- 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 (overlays photo) --- @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 Cadres impression (par format) --- @app.get("/api/cadres-impression/{format_papier}") async def api_cadres_impression_list(format_papier: str): if format_papier not in FORMATS_CADRES: return JSONResponse({"erreur": "Format inconnu"}, status_code=400) config = charger_config() actif = config.get("impression", {}).get("cadres_actifs", {}).get(format_papier) return {"format": format_papier, "disponibles": lister_cadres(format_papier), "actif": actif} @app.post("/api/cadres-impression/{format_papier}/actif") async def api_cadres_impression_set(format_papier: str, donnees: dict): if format_papier not in FORMATS_CADRES: return JSONResponse({"erreur": "Format inconnu"}, status_code=400) nom = donnees.get("nom") # None = désactiver config = charger_config() cadres_actifs = config.get("impression", {}).get("cadres_actifs", {}) cadres_actifs[format_papier] = nom mettre_a_jour_config({"impression": {"cadres_actifs": cadres_actifs}}) return {"format": format_papier, "actif": nom} @app.post("/api/upload/cadre/{format_papier}") async def api_upload_cadre(format_papier: str, fichier: UploadFile = File(...)): if format_papier not in FORMATS_CADRES: return JSONResponse({"erreur": "Format inconnu"}, status_code=400) if not fichier.filename.lower().endswith(".png"): return JSONResponse({"erreur": "Seuls les PNG sont acceptés"}, status_code=400) dossier = DOSSIER_CADRES / format_papier dossier.mkdir(parents=True, exist_ok=True) chemin = dossier / fichier.filename with open(chemin, "wb") as f: f.write(await fichier.read()) return {"nom": fichier.filename, "format": format_papier} @app.delete("/api/cadres-impression/{format_papier}/{nom}") async def api_supprimer_cadre(format_papier: str, nom: str): if format_papier not in FORMATS_CADRES: return JSONResponse({"erreur": "Format inconnu"}, status_code=400) chemin = DOSSIER_CADRES / format_papier / nom if not chemin.exists() or chemin.parent != DOSSIER_CADRES / format_papier: return JSONResponse({"erreur": "Fichier introuvable"}, status_code=404) chemin.unlink() # Désactiver si c'était le cadre actif config = charger_config() cadres_actifs = config.get("impression", {}).get("cadres_actifs", {}) if cadres_actifs.get(format_papier) == nom: cadres_actifs[format_papier] = None mettre_a_jour_config({"impression": {"cadres_actifs": cadres_actifs}}) return {"succes": True} # --- 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 Photostation --- FLAG_PHOTOSTATION = Path("/tmp/photostation_running") PHOTOSTATION_DIR = Path("/opt/photostation") @app.get("/api/photostation/disponible") async def api_photostation_disponible(): disponible = (PHOTOSTATION_DIR / "src" / "main.py").exists() return {"disponible": disponible} @app.post("/api/photostation/lancer") async def api_photostation_lancer(): if not (PHOTOSTATION_DIR / "src" / "main.py").exists(): return JSONResponse({"erreur": "Photostation non installée"}, status_code=404) asyncio.create_task(_lancer_photostation()) return {"succes": True} async def _lancer_photostation(): FLAG_PHOTOSTATION.touch() await asyncio.sleep(1) # Fermer Chromium (la boucle kiosk attend le flag avant de le relancer) await asyncio.create_subprocess_exec("pkill", "chromium") await asyncio.sleep(2) import subprocess as _sp env = {**__import__("os").environ, "DISPLAY": ":0"} # Récupérer les variables de session X/dbus depuis la session LightDM try: r = _sp.run(["grep", "-z", "DISPLAY\|XAUTHORITY\|DBUS_SESSION", f"/proc/{__import__('os').getpid()}/environ"], capture_output=True) for kv in r.stdout.split(b"\x00"): if b"=" in kv: k, v = kv.decode(errors="ignore").split("=", 1) if k in ("DISPLAY", "XAUTHORITY", "DBUS_SESSION_BUS_ADDRESS"): env[k] = v except Exception: pass log_path = "/tmp/photostation.log" cmd = (f"source {PHOTOSTATION_DIR}/venv/bin/activate && " f"python {PHOTOSTATION_DIR}/src/main.py --kiosk >> {log_path} 2>&1") proc = await asyncio.create_subprocess_exec("bash", "-c", cmd, env=env) await proc.wait() FLAG_PHOTOSTATION.unlink(missing_ok=True) # --- 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_start": global _preview_actif _preview_actif = True elif type_msg == "preview_stop": _preview_actif = False 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", 80), reload=False, log_level="info", )