import asyncio import base64 import json import logging import os 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, consommables_etat, reset_consommables from backend.printer import lister_imprimantes, imprimer from backend.mailer import envoyer_photo, charger_emails_historique, effacer_emails_historique, ajouter_au_spool, taille_spool, tache_spool_demarrage from backend.qrcode_gen import generer_qr, qr_galerie from backend import relais from backend.evenements import ( lister_evenements, creer_evenement, obtenir_evenement, modifier_evenement, supprimer_evenement, activer_evenement, terminer_evenement, evenement_est_termine, lister_cadres_event, set_cadre_event, cadre_actif_pour_impression, ) from backend.evenements import DOSSIER_EVENEMENTS from backend.eclairage import analyser_frame as analyser_eclairage, dernier_resultat as dernier_eclairage from backend.wifi import wifi_status, wifi_scan, wifi_connect, wifi_saved_list, wifi_forget, wifi_get_password 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 = 40 # 40 x 50ms = 2s avant de declarer une erreur (Canon LiveView peut être lent à démarrer) _derniere_keepalive = 0.0 # --- Suivi activité Canon pour diagnostic et gestion veille --- _canon_stats = { "derniere_capture": 0.0, "derniere_preview": 0.0, "dernier_keepalive_ok": 0.0, "dernier_keepalive_fail": 0.0, "keepalive_ok": 0, "keepalive_fail": 0, "preview_ok": 0, "preview_fail": 0, "captures": 0, "connexions": 0, "deconnexions": 0, "power_cycles": 0, "demarrage": time.time(), } _CANON_VEILLE_MINUTES = 30 # couper le Canon après X min sans activité utilisateur _canon_en_veille = False _reveil_en_cours = False def _thread_preview(): """Thread de fond : capture les frames DSLR. Actif seulement quand _preview_actif=True (ecran capture). Sinon keepalive leger toutes les 30s pour empecher le Canon de s'eteindre. Gère la mise en veille Canon après inactivité prolongée.""" global _derniere_frame_preview, _derniere_keepalive, _canon_en_veille _dernier_log_keepalive = 0.0 while True: try: if capture_en_cours or camera.mode != "gphoto2" or not camera.connectee: time.sleep(0.2) continue # Vérifier inactivité prolongée → couper Canon via relais now = time.time() derniere_activite = max( _canon_stats["derniere_capture"], _canon_stats["derniere_preview"], ) if derniere_activite > 0 and not _preview_actif and not _canon_en_veille and not _reveil_en_cours: inactif_min = (now - derniere_activite) / 60 if inactif_min >= _CANON_VEILLE_MINUTES and relais.est_connecte(): log.info(f"Canon inactif depuis {inactif_min:.0f} min — mise en veille relais") _canon_en_veille = True camera.deconnecter() relais.activer("canon") relais.activer("canon_usb") _canon_stats["power_cycles"] += 1 continue if _canon_en_veille: time.sleep(2) continue if not _preview_actif: if now - _derniere_keepalive > 30: ok = camera.keepalive() _derniere_keepalive = now if ok: _canon_stats["keepalive_ok"] += 1 _canon_stats["keepalive_fail_consecutifs"] = 0 _canon_stats["dernier_keepalive_ok"] = now if now - _dernier_log_keepalive > 300: log.info(f"Canon keepalive OK (total: {_canon_stats['keepalive_ok']}, fail: {_canon_stats['keepalive_fail']})") _dernier_log_keepalive = now else: _canon_stats["keepalive_fail"] += 1 _canon_stats["keepalive_fail_consecutifs"] = _canon_stats.get("keepalive_fail_consecutifs", 0) + 1 _canon_stats["dernier_keepalive_fail"] = now consec = _canon_stats["keepalive_fail_consecutifs"] log.warning(f"Canon keepalive FAIL #{_canon_stats['keepalive_fail']} (consec: {consec})") if consec >= 5: log.error(f"Canon PTP freeze detecte — {consec} keepalive FAIL consecutifs, deconnexion forcee") _canon_stats["keepalive_fail_consecutifs"] = 0 camera.deconnecter() time.sleep(1) continue try: donnees = camera.preview() with _preview_lock: _derniere_frame_preview = donnees _derniere_keepalive = now if donnees: _canon_stats["preview_ok"] += 1 _canon_stats["derniere_preview"] = now else: _canon_stats["preview_fail"] += 1 except Exception: with _preview_lock: _derniere_frame_preview = None _canon_stats["preview_fail"] += 1 if now - _derniere_keepalive > 10: camera.keepalive() _derniere_keepalive = now time.sleep(0.05) # ~20 fps max except Exception as e: log.error(f"_thread_preview crash: {e}") time.sleep(1) 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: log.warning(f"camera_erreur envoyé : preview None depuis {_preview_none_count * 50}ms") 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}"}) if not capture_en_cours: eclairage = await loop.run_in_executor(None, analyser_eclairage, donnees) if eclairage is not None: await diffuser_ws({"type": "eclairage", **eclairage}) _gerer_eclairage_auto(eclairage) _proj_etat = {"proj1": False, "proj2": False} _veille_eclairage = False _nb_proj_capture = 2 _preview_eclairage_init = False def _gerer_eclairage_auto(eclairage: dict): """Preview : 1 projecteur ON pour estimer le besoin. Le score détermine combien pour la capture.""" global _nb_proj_capture, _preview_eclairage_init if _veille_eclairage or capture_en_cours: return config = charger_config() rcfg = config.get("relais", {}) if not rcfg.get("eclairage_auto") or not relais.est_connecte(): return if not _preview_eclairage_init: relais.activer("projecteur_gauche") _proj_etat["proj1"] = True _proj_etat["proj2"] = False _preview_eclairage_init = True score = eclairage.get("score", -1) if score < 0: return seuil_2 = rcfg.get("seuil_proj2", 30) if score < seuil_2: _nb_proj_capture = 2 else: _nb_proj_capture = 1 def _usb_reset_canon(): """Reset USB ioctl du Canon EOS (vendor 04a9) pour débloquer un port stall.""" import fcntl, glob USBDEVFS_RESET = 0x5514 try: for f in glob.glob("/sys/bus/usb/devices/*/idVendor"): if open(f).read().strip() == "04a9": d = __import__("os.path", fromlist=["dirname"]).dirname(f) bus = open(d + "/busnum").read().strip().zfill(3) dev = open(d + "/devnum").read().strip().zfill(3) path = f"/dev/bus/usb/{bus}/{dev}" with open(path, "wb") as fh: fcntl.ioctl(fh, USBDEVFS_RESET, 0) log.info(f"USB reset Canon: {path}") return except Exception as e: log.debug(f"_usb_reset_canon: {e}") def _ptp_reset_canon(): """Envoie un PTP USB Device Reset (class request 0x66) + Clear Halt endpoints. Cette commande atteint le firmware PTP même quand le Canon est figé en I/O error, car elle utilise le control transfer USB (endpoint 0) et non les bulk endpoints.""" try: import usb.core, usb.util, struct dev = usb.core.find(idVendor=0x04a9) if not dev: log.debug("_ptp_reset_canon: Canon non trouvé en USB") return False for cfg in dev: for intf in cfg: try: if dev.is_kernel_driver_active(intf.bInterfaceNumber): dev.detach_kernel_driver(intf.bInterfaceNumber) except Exception: pass dev.set_configuration() cfg = dev.get_active_configuration() intf = cfg[(0, 0)] ep_out = usb.util.find_descriptor(intf, custom_match=lambda e: usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_OUT) ep_in = usb.util.find_descriptor(intf, custom_match=lambda e: usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_IN and e.bmAttributes == 2) dev.ctrl_transfer(0x21, 0x66, 0, 0, None, timeout=5000) log.info("PTP USB Device Reset (0x66) envoyé") time.sleep(1) if ep_out: try: dev.clear_halt(ep_out) except Exception: pass if ep_in: try: dev.clear_halt(ep_in) except Exception: pass try: status = dev.ctrl_transfer(0xA1, 0x67, 0, 0, 12, timeout=5000) code = struct.unpack_from(" unbind/rebind -> uhubctl power cycle -> backoff long.""" global _dslr_erreurs _echecs_connexion = 0 while True: if _echecs_connexion >= 10: await asyncio.sleep(60) else: await asyncio.sleep(5) if capture_en_cours or _reveil_en_cours: continue try: if not GPHOTO2_DISPONIBLE: continue if camera.mode == "gphoto2" and camera.connectee: if not _preview_actif and not capture_en_cours: try: with camera._gp_lock: camera.camera.get_config() except Exception: _dslr_erreurs += 1 if _dslr_erreurs >= 3: log.warning(f"Canon get_config echoue {_dslr_erreurs} fois — reconnexion forcee") _dslr_erreurs = 0 _echecs_connexion += 1 await _reconnecter_dslr_avec_reset(_echecs_connexion) continue if camera.preview_dslr_ok: _dslr_erreurs = 0 _echecs_connexion = 0 else: _dslr_erreurs += 1 if _dslr_erreurs >= 3: log.warning(f"camera_erreur : DSLR preview KO depuis {_dslr_erreurs * 5}s, reconnexion forcee...") _dslr_erreurs = 0 _echecs_connexion += 1 await _reconnecter_dslr_avec_reset(_echecs_connexion) else: try: dslrs = gp.Camera.autodetect() except Exception: dslrs = [] if len(dslrs) > 0: _echecs_connexion += 1 log.info(f"DSLR detecte : {dslrs[0][0]}, connexion automatique... (tentative {_echecs_connexion})") if _echecs_connexion <= 2: await asyncio.sleep(5) else: _usb_reset_canon() await asyncio.sleep(3) await _reconnecter_dslr_avec_reset(_echecs_connexion) else: if _echecs_connexion > 0: _echecs_connexion += 1 if relais.est_connecte() and _echecs_connexion % 10 == 0: log.warning(f"Canon absent USB (tentative {_echecs_connexion}) — power cycle relais...") await asyncio.get_event_loop().run_in_executor(None, relais.power_cycle_canon, 10.0) await asyncio.sleep(25) else: log.info(f"DSLR non detecte en USB (tentative {_echecs_connexion}), rebind...") await asyncio.get_event_loop().run_in_executor(None, _usb_rebind_canon) elif _canon_sur_usb(): log.info("Canon detecte sur USB sans echec precedent — tentative connexion...") _echecs_connexion = 1 camera.connecter(source="gphoto2") if camera.connectee: log.info("Canon reconnecte automatiquement") _canon_stats["connexions"] += 1 _canon_stats["derniere_preview"] = time.time() _canon_stats["derniere_capture"] = time.time() _appliquer_config_camera() _echecs_connexion = 0 await diffuser_ws({"type": "camera_ok"}) except Exception as e: log.debug(f"surveiller_dslr: {e}") async def _reconnecter_dslr_avec_reset(echecs: int): """Reconnexion DSLR avec escalade progressive : 1 : simple reconnexion 2 : PTP class reset (0x66) + ioctl reset 3-5 : unbind/rebind USB 6-9 : PTP reset + unbind/rebind (plus agressif) 10+ : relay hard reset (si câblé) + backoff long 20+ : dormant (5 min entre tentatives) """ camera.deconnecter() _canon_stats["deconnexions"] += 1 import gc; gc.collect() await diffuser_ws({"type": "camera_erreur", "message": "Appareil photo deconnecte"}) if echecs >= 20: log.info(f"Mode dormant (echec #{echecs}) — attente 5 min avant prochaine tentative") await diffuser_ws({"type": "camera_erreur", "message": "Appareil photo injoignable — redémarrez-le physiquement"}) await asyncio.sleep(300) elif echecs >= 10 and echecs % 5 == 0 and relais.est_connecte(): log.warning(f"Hard reset Canon (echec #{echecs}) — trappe + coupure alim 30s") await diffuser_ws({"type": "camera_erreur", "message": "Reset complet appareil photo..."}) await asyncio.get_event_loop().run_in_executor(None, relais.hard_reset_canon, 30.0) await asyncio.sleep(20) elif echecs >= 6: log.info(f"PTP reset + rebind Canon (echec #{echecs})...") await asyncio.get_event_loop().run_in_executor(None, _ptp_reset_canon) await asyncio.sleep(2) await asyncio.get_event_loop().run_in_executor(None, _usb_rebind_canon) elif echecs >= 3: log.info(f"USB rebind Canon (echec #{echecs})...") ok = await asyncio.get_event_loop().run_in_executor(None, _usb_rebind_canon) if not ok: _usb_reset_canon() await asyncio.sleep(4) elif echecs >= 2: log.info(f"PTP reset + USB reset Canon (echec #{echecs})...") await asyncio.get_event_loop().run_in_executor(None, _ptp_reset_canon) await asyncio.sleep(1) _usb_reset_canon() await asyncio.sleep(3) else: await asyncio.sleep(2) camera.connecter(source="gphoto2") if camera.connectee: log.info("DSLR reconnecte avec succes") _canon_stats["connexions"] += 1 _canon_stats["derniere_preview"] = time.time() _canon_stats["derniere_capture"] = time.time() _appliquer_config_camera() await diffuser_ws({"type": "camera_ok"}) else: log.warning(f"Echec reconnexion DSLR (tentative {echecs})") def _appliquer_config_camera(): """Applique les paramètres caméra persistants (flash, etc.) après connexion.""" cfg = charger_config() flash_integre = cfg.get("camera", {}).get("flash_integre", True) camera.configurer_flash(flash_integre) async def _watchdog_systemd(): """Notifie systemd que le service est vivant + surveille la santé.""" try: import socket addr = os.environ.get("NOTIFY_SOCKET") if not addr: return sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) if addr[0] == "@": addr = "\0" + addr[1:] sock.sendto(b"READY=1", addr) while True: await asyncio.sleep(10) sock.sendto(b"WATCHDOG=1", addr) except Exception: pass def _diagnostiquer_erreur_imprimante(nom: str) -> dict: """Parse les logs CUPS pour identifier le type d'erreur imprimante.""" import subprocess try: r = subprocess.run( ["sudo", "tail", "-20", "/var/log/cups/error_log"], capture_output=True, text=True, timeout=5 ) logs = r.stdout.lower() if "ribbon" in logs and ("end" in logs or "count" in logs): return { "situation": "changement_rouleau", "titre": "Ruban termine", "message": "Le ruban d'impression est termine. Il faut le remplacer.", "etapes": [ "Ouvrez le capot de l'imprimante", "Retirez la cassette ruban usagee", "Inserez une cassette ruban neuve (sens indique par la fleche)", "Refermez le capot jusqu'au clic", ], "code": "05/02/02", } if "paper" in logs and ("end" in logs or "empty" in logs or "out" in logs): return { "situation": "changement_rouleau", "titre": "Papier termine", "message": "Le papier est termine. Il faut recharger le bac.", "etapes": [ "Ouvrez le capot de l'imprimante", "Retirez le bac papier vide", "Chargez un nouveau rouleau de papier (face brillante vers le haut)", "Refermez le capot jusqu'au clic", ], "code": "paper_end", } if "jam" in logs or "bourrage" in logs: return { "situation": "bourrage", "titre": "Bourrage papier", "message": "Un bourrage papier a ete detecte.", "etapes": [ "Ouvrez le capot de l'imprimante", "Retirez delicatement le papier coince (ne pas tirer fort)", "Verifiez qu'il ne reste pas de morceaux", "Refermez le capot jusqu'au clic", ], "code": "jam", } if "media" in logs and "mismatch" in logs: return { "situation": "depannage_imprimante", "titre": "Mauvais format", "message": "Le format papier ne correspond pas a la cassette inseree.", "etapes": [ "Ouvrez le capot de l'imprimante", "Verifiez que la cassette correspond au format (10x15 ou 15x20)", "Retirez et reinserez la cassette correctement", "Refermez le capot", ], "code": "media_mismatch", } if "offline" in logs or "not connected" in logs: return { "situation": "depannage_imprimante", "titre": "Imprimante deconnectee", "message": "L'imprimante n'est pas detectee.", "etapes": [ "Verifiez que l'imprimante est allumee (voyant vert)", "Verifiez le cable USB entre l'imprimante et la borne", "Eteignez et rallumez l'imprimante", "Si le probleme persiste, redemarrez la borne", ], "code": "offline", } except Exception: pass return { "situation": "depannage_imprimante", "titre": "Probleme imprimante", "message": f"L'imprimante {nom} ne fonctionne pas correctement.", "etapes": [ "Verifiez que l'imprimante est allumee", "Verifiez le cable USB", "Ouvrez le capot et verifiez papier et ruban", "Eteignez et rallumez l'imprimante", ], "code": "unknown", } async def _surveiller_imprimante(): """Surveille l'imprimante CUPS en continu : réactive si disabled, notifie les clients.""" import subprocess _derniere_alerte = 0 _dernier_code = None while True: await asyncio.sleep(30) try: config = charger_config() if not config.get("fonctionnalites", {}).get("impression", False): continue nom = config.get("impression", {}).get("imprimante") or "Mitsubishi" r = subprocess.run(["lpstat", "-p", nom], capture_output=True, text=True, timeout=5) out = r.stdout.lower() if "disabled" in out or "stopped" in out: log.warning(f"Imprimante {nom} désactivée — réactivation automatique") subprocess.run(["sudo", "cupsenable", nom], capture_output=True, timeout=5) await asyncio.sleep(2) r2 = subprocess.run(["lpstat", "-p", nom], capture_output=True, text=True, timeout=5) if "disabled" not in r2.stdout.lower() and "stopped" not in r2.stdout.lower(): log.info(f"Imprimante {nom} réactivée avec succès") _dernier_code = None await diffuser_ws({"type": "imprimante_ok"}) else: now = time.time() diag = _diagnostiquer_erreur_imprimante(nom) if now - _derniere_alerte > 300 or diag["code"] != _dernier_code: _derniere_alerte = now _dernier_code = diag["code"] await diffuser_ws({"type": "imprimante_erreur", **diag}) else: if _dernier_code is not None: _dernier_code = None await diffuser_ws({"type": "imprimante_ok"}) except Exception as e: log.debug(f"_surveiller_imprimante: {e}") def _reveiller_canon(): """Réveil Canon depuis veille relais : restaure alimentation, attend boot, reconnecte.""" global _reveil_en_cours _reveil_en_cours = True try: log.info("Réveil Canon — restauration alimentation relais") relais.desactiver("canon") time.sleep(15) relais.desactiver("canon_usb") time.sleep(25) if _canon_sur_usb(): log.info("Canon visible USB après réveil — connexion...") camera.connecter(source="gphoto2") if camera.connectee: _appliquer_config_camera() _canon_stats["derniere_preview"] = time.time() _canon_stats["derniere_capture"] = time.time() log.info("Canon reconnecté après réveil") _canon_stats["connexions"] += 1 else: log.warning("Canon visible USB mais connexion gphoto2 échouée") else: log.warning("Canon absent USB après réveil relais — vérifier branchement physique") finally: _reveil_en_cours = False def _canon_sur_usb() -> bool: """Vérifie si un Canon (vendor 04a9) est visible sur le bus USB.""" import glob for f in glob.glob("/sys/bus/usb/devices/*/idVendor"): try: if open(f).read().strip() == "04a9": return True except Exception: pass return False async def _sequence_boot_canon(): """Séquençage propre du Canon au démarrage via relais. Coupe l'alimentation, attend la vidange, restaure alim puis USB. Donne au Canon le temps de booter avant toute tentative gphoto2.""" if not relais.est_connecte(): log.info("Pas de relais — skip sequence boot Canon") return if _canon_sur_usb(): log.info("Canon déjà visible USB — skip power cycle") return log.info("Canon absent USB — séquence boot relais (power cycle 10s + attente 25s)") loop = asyncio.get_event_loop() def _do_power_cycle(): relais.activer("canon") relais.activer("canon_usb") time.sleep(10) relais.desactiver("canon") time.sleep(15) relais.desactiver("canon_usb") await loop.run_in_executor(None, _do_power_cycle) log.info("Canon alimenté — attente boot 25s...") await asyncio.sleep(25) if _canon_sur_usb(): log.info("Canon visible USB après power cycle relais") else: log.warning("Canon toujours absent USB après power cycle — problème physique probable") async def _connecter_camera_bg(): """Connexion caméra en arrière-plan — ne bloque pas le serveur. Si relais dispo et Canon absent USB, fait un power cycle propre d'abord.""" if GPHOTO2_DISPONIBLE: await _sequence_boot_canon() for _tentative in range(5): if camera.connecter(): _appliquer_config_camera() log.info("Camera connectee en arriere-plan") return if _tentative < 2: log.info(f"Connexion camera echouee (tentative {_tentative+1}/5), attente 5s...") await asyncio.sleep(5) else: log.warning(f"Connexion camera echouee (tentative {_tentative+1}/5), USB reset...") _usb_reset_canon() import gc; gc.collect() await asyncio.sleep(3) log.warning("Camera non connectee au demarrage — surveiller_dslr prendra le relais") @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() relais.connecter() # Thread de capture preview (bloquant, tourne en parallele) t = threading.Thread(target=_thread_preview, daemon=True) t.start() task_camera = asyncio.create_task(_connecter_camera_bg()) task_dslr = asyncio.create_task(surveiller_dslr()) task_push = asyncio.create_task(_pusher_preview()) task_spool = asyncio.create_task(tache_spool_demarrage()) task_watchdog = asyncio.create_task(_watchdog_systemd()) task_imprimante = asyncio.create_task(_surveiller_imprimante()) yield task_dslr.cancel() task_push.cancel() task_spool.cancel() task_watchdog.cancel() task_imprimante.cancel() relais.deconnecter() 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("/sounds", StaticFiles(directory=str(RACINE / "frontend" / "sounds")), name="sounds") 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("/api/health") async def api_health(): return { "status": "ok", "camera": camera.mode if camera.connectee else "disconnected", "clients": len(clients_ws), "uptime": int(time.time() - _start_time), } _start_time = time.time() @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}) # Appliquer immédiatement les paramètres caméra si modifiés if "camera" in modifications and camera.connectee: _appliquer_config_camera() return config # --- API Camera --- @app.post("/api/capturer") async def api_capturer(): global capture_en_cours, _preview_actif, _canon_en_veille if _canon_en_veille and not _reveil_en_cours: log.info("Canon en veille lors de la capture — réveil automatique") _canon_en_veille = False await asyncio.get_event_loop().run_in_executor(None, _reveiller_canon) if _reveil_en_cours: log.info("Canon en cours de réveil — attente max 60s") for _ in range(60): if not _reveil_en_cours: break await asyncio.sleep(1) if not camera.connectee: return JSONResponse({"erreur": "Appareil photo en cours de démarrage — réessayez"}, status_code=503) capture_en_cours = True _preview_actif = False _canon_stats["captures"] += 1 _canon_stats["derniere_capture"] = time.time() config = charger_config() _flash = config.get("relais", {}).get("eclairage_auto") and relais.est_connecte() if _flash: relais.projecteurs(True) _proj_etat["proj1"] = True _proj_etat["proj2"] = True log.info("Capture déclenchée — projecteurs déjà ON depuis countdown") await diffuser_ws({"type": "shutter"}) try: loop = asyncio.get_event_loop() try: chemin = await asyncio.wait_for( loop.run_in_executor(None, camera.capturer), timeout=20.0 ) except asyncio.TimeoutError: log.error("TIMEOUT capture DSLR après 20s — déconnexion forcée") camera.deconnecter() await diffuser_ws({"type": "camera_erreur", "message": "Timeout capture"}) return JSONResponse({"erreur": "Timeout capture — vérifiez le DSLR"}, status_code=500) finally: capture_en_cours = False if _flash: relais.projecteurs(False) _proj_etat["proj1"] = False _proj_etat["proj2"] = False global _preview_eclairage_init _preview_eclairage_init = False log.info(f"Capture terminée : {chemin}") if chemin is None: if camera.mode == "gphoto2" and not camera.preview_dslr_ok: log.warning("Capture echouee + preview KO → reconnexion DSLR imminente") await diffuser_ws({"type": "camera_erreur", "message": "Erreur capture — reconnexion en cours"}) 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.post("/api/log-js") async def api_log_js(request: Request): data = await request.json() log.warning(f"[JS] {data.get('msg', '?')}") return {"ok": True} @app.get("/api/camera/statut") async def api_camera_statut(): return { "connectee": camera.connectee, "mode": camera.mode, "reveil_en_cours": _reveil_en_cours, "appareils": camera.lister_appareils(), } @app.get("/api/eclairage") async def api_eclairage(): resultat = dernier_eclairage() if resultat is None: return {"score": -1, "action": "no_data", "detail": "Aucune analyse disponible"} return resultat # --- API Relais --- @app.get("/api/relais") async def api_relais_status(): return relais.etat_complet() @app.post("/api/relais/{nom}/on") async def api_relais_on(nom: str): if nom == "projecteurs": relais.projecteurs(True) return {"ok": True} if not relais.activer(nom): return JSONResponse({"erreur": f"relais '{nom}' introuvable ou non connecté"}, 400) return {"ok": True} @app.post("/api/relais/{nom}/off") async def api_relais_off(nom: str): if nom == "projecteurs": relais.projecteurs(False) return {"ok": True} if not relais.desactiver(nom): return JSONResponse({"erreur": f"relais '{nom}' introuvable ou non connecté"}, 400) return {"ok": True} @app.post("/api/relais/canon/power-cycle") async def api_power_cycle_canon(req: Request): body = await req.json() if req.headers.get("content-type", "").startswith("application/json") else {} duree = body.get("duree", 3.0) loop = asyncio.get_event_loop() await loop.run_in_executor(None, relais.power_cycle_canon, duree) return {"ok": True} @app.post("/api/relais/canon/hard-reset") async def api_hard_reset_canon(req: Request): body = await req.json() if req.headers.get("content-type", "").startswith("application/json") else {} duree = body.get("duree", 3.0) loop = asyncio.get_event_loop() await loop.run_in_executor(None, relais.hard_reset_canon, duree) return {"ok": True} @app.get("/api/relais/config") async def api_relais_config(): config = charger_config() return config.get("relais", { "eclairage_auto": False, "seuil_proj1": 45, "seuil_proj2": 30, "seuil_off": 65, }) @app.post("/api/relais/config") async def api_relais_config_save(req: Request): body = await req.json() mettre_a_jour_config({"relais": body}) return {"ok": True} @app.post("/api/relais/veille") async def api_relais_veille(): global _veille_eclairage _veille_eclairage = True if relais.est_connecte(): relais.projecteurs(False) _proj_etat["proj1"] = False _proj_etat["proj2"] = False log.info("Eclairage en veille") return {"ok": True} @app.post("/api/relais/reveil") async def api_relais_reveil(): global _veille_eclairage, _canon_en_veille _veille_eclairage = False if relais.est_connecte(): config = charger_config() rcfg = config.get("relais", {}) if rcfg.get("eclairage_auto"): relais.activer("projecteur_gauche") _proj_etat["proj1"] = True _proj_etat["proj2"] = False if _canon_en_veille: log.info("Réveil Canon déclenché par reveil éclairage (HTTP)") _canon_en_veille = False asyncio.get_event_loop().run_in_executor(None, _reveiller_canon) log.info("Eclairage reveille (1 projecteur)") return {"ok": True} @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: expo = {} with camera._gp_lock: cfg = camera.camera.get_config() for nom in ["autoexposuremode", "iso", "shutterspeed", "aperture", "meteringmode", "whitebalance"]: try: w = cfg.get_child_by_name(nom) expo[nom] = w.get_value() except Exception: pass info["exposition"] = expo mode = expo.get("autoexposuremode", "") if mode in ("Flash Off", "Auto", "Night Portrait", "Landscape", "Portrait", "Sports"): info["avertissement"] = f"Canon en mode scène '{mode}' — vitesse/ISO auto non modifiables. Tourner le dial sur M ou Av." except Exception as e: info["exposition"] = f"ERREUR : {e}" return info @app.get("/api/canon/stats") async def api_canon_stats(): """Stats diagnostic Canon : keepalive, preview, connexions, veille.""" uptime = time.time() - _canon_stats["demarrage"] stats = dict(_canon_stats) stats["uptime_min"] = round(uptime / 60, 1) stats["canon_en_veille"] = _canon_en_veille stats["canon_sur_usb"] = _canon_sur_usb() stats["preview_actif"] = _preview_actif stats["veille_apres_min"] = _CANON_VEILLE_MINUTES derniere_activite = max(stats["derniere_capture"], stats["derniere_preview"]) if derniere_activite > 0: stats["inactif_min"] = round((time.time() - derniere_activite) / 60, 1) else: stats["inactif_min"] = None return stats @app.post("/api/camera/reconnecter") async def api_camera_reconnecter(body: dict = {}): camera.deconnecter() import gc; gc.collect() source = body.get("source") # PTP reset avant reconnexion pour débloquer un éventuel freeze await asyncio.get_event_loop().run_in_executor(None, _ptp_reset_canon) await asyncio.sleep(1) _usb_reset_canon() await asyncio.sleep(2) ok = camera.connecter(source=source) if ok: _appliquer_config_camera() 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") or "Mitsubishi" try: r = subprocess.run(["lpstat", "-p", nom], capture_output=True, text=True, timeout=5) statut_ligne = r.stdout.strip() except Exception: statut_ligne = "Impossible de lire le statut CUPS" try: r2 = subprocess.run( ["sudo", "grep", f"\\[{nom}\\]\\|Job.*cancel\\|media.*match\\|jam\\|paper", "/var/log/cups/error_log"], capture_output=True, text=True, timeout=5 ) 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") or "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") or "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") or "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): if evenement_est_termine(): return JSONResponse({"succes": False, "erreur": "evenement_termine", "message": "L'evenement est termine — impression desactivee"}, status_code=403) nom = donnees.get("photo", "") copies = donnees.get("copies", 1) config = charger_config() copies_max = config.get("impression", {}).get("copies_max", 5) copies = max(1, min(copies, copies_max)) 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 try: resultat = imprimer(chemin, copies=copies, cadre_override=cadre_override, format_papier=format_papier) except Exception as e: log.error(f"Erreur impression inattendue : {e}") return {"succes": False, "erreur": "erreur_impression", "message": f"Erreur interne : {e}"} if resultat.get("succes"): distribuer_photo(chemin, imprimee=True, copies=copies, format_papier=format_papier) return resultat # --- API Consommables --- @app.get("/api/consommables") async def api_consommables(): return consommables_etat() @app.post("/api/consommables/reset") async def api_reset_consommables(request: Request): body = await request.json() quoi = body.get("quoi", "tout") reset_consommables(quoi) return consommables_etat() @app.post("/api/consommables/capacite") async def api_consommables_capacite(request: Request): body = await request.json() updates = {} if "papier_capacite" in body: updates["papier_capacite"] = int(body["papier_capacite"]) if "ruban_capacite" in body: updates["ruban_capacite"] = int(body["ruban_capacite"]) if updates: config = charger_config() conso = config.get("consommables", {}) conso.update(updates) mettre_a_jour_config({"consommables": conso}) return consommables_etat() # --- 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) if ok: return {"succes": True} ajouter_au_spool(email_dest, str(chemin)) _sauvegarder_email_historique(email_dest) return {"succes": True, "spool": True, "en_attente": taille_spool()} @app.get("/api/emails/historique") async def api_emails_historique(): return {"emails": charger_emails_historique()} @app.delete("/api/emails/historique") async def api_emails_historique_supprimer(): effacer_emails_historique() return {"succes": True} # --- 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}"} @app.post("/api/booth/imprimer-qr") async def api_booth_imprimer_qr(): """Genere et imprime un flyer QR pour la galerie booth.""" config = charger_config() booth = config.get("booth", {}) event = config.get("evenement", {}) if not booth.get("actif") or not booth.get("url"): return JSONResponse({"erreur": "Booth non configure"}, status_code=400) info = recuperer_booth_password() password = info.get("password", "") event_id = booth.get("event_id", "default") nom_event = event.get("nom", "Photobooth") url_galerie = f"{booth['url']}/{event_id}?p={password}" from PIL import Image, ImageDraw, ImageFont import qrcode W, H = 1200, 1800 # 10x15 portrait @ 300dpi img = Image.new("RGB", (W, H), "#0a0a1a") draw = ImageDraw.Draw(img) try: font_big = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 72) font_med = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 42) font_small = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 32) except OSError: font_big = ImageFont.load_default() font_med = font_big font_small = font_big # Titre evenement draw.text((W // 2, 120), "📸", font=font_big, anchor="mt", fill="#e91e63") draw.text((W // 2, 220), nom_event, font=font_big, anchor="mt", fill="#ffffff") # Ligne decorative draw.line([(150, 320), (W - 150, 320)], fill="#e91e63", width=3) # Texte d'invitation draw.text((W // 2, 400), "Retrouvez toutes les photos", font=font_med, anchor="mt", fill="#cccccc") draw.text((W // 2, 460), "en direct sur la galerie !", font=font_med, anchor="mt", fill="#cccccc") # QR code qr = qrcode.QRCode(version=None, error_correction=qrcode.constants.ERROR_CORRECT_M, box_size=12, border=3) qr.add_data(url_galerie) qr.make(fit=True) qr_img = qr.make_image(fill_color="#ffffff", back_color="#0a0a1a").convert("RGB") qr_size = 700 qr_img = qr_img.resize((qr_size, qr_size), Image.NEAREST) img.paste(qr_img, ((W - qr_size) // 2, 540)) # Instruction draw.text((W // 2, 1300), "Scannez ce QR code", font=font_med, anchor="mt", fill="#ffffff") draw.text((W // 2, 1360), "avec votre téléphone", font=font_med, anchor="mt", fill="#ffffff") # Ligne decorative bas draw.line([(150, 1460), (W - 150, 1460)], fill="#e91e63", width=3) draw.text((W // 2, 1520), "Les photos apparaissent en temps réel", font=font_small, anchor="mt", fill="#888888") draw.text((W // 2, 1570), "Téléchargez-les directement !", font=font_small, anchor="mt", fill="#888888") # Sauvegarder chemin = DOSSIER_EXPORTS / "flyer_qr_booth.jpg" img.save(str(chemin), "JPEG", quality=95) # Imprimer ok = imprimer(str(chemin), copies=1) return {"succes": ok, "chemin": f"/data/exports/flyer_qr_booth.jpg"} # --- API Evenements --- @app.get("/api/evenements") async def api_lister_evenements(): return lister_evenements() @app.post("/api/evenements") async def api_creer_evenement(donnees: dict): nom = donnees.get("nom", "").strip() if not nom: return JSONResponse({"erreur": "Nom requis"}, status_code=400) event = creer_evenement( nom, event_id=donnees.get("event_id"), theme=donnees.get("theme", "base"), couleur_primaire=donnees.get("couleur_primaire", "#e91e63"), couleur_secondaire=donnees.get("couleur_secondaire", "#ffffff"), media_accueil=donnees.get("media_accueil"), date_fin=donnees.get("date_fin"), ) return event @app.get("/api/evenements/{event_id}") async def api_obtenir_evenement(event_id: str): event = obtenir_evenement(event_id) if not event: return JSONResponse({"erreur": "Evenement introuvable"}, status_code=404) return event def _event_id_actif() -> str | None: return charger_config().get("evenement", {}).get("event_id") @app.put("/api/evenements/{event_id}") async def api_modifier_evenement(event_id: str, donnees: dict): event = modifier_evenement(event_id, donnees) if not event: return JSONResponse({"erreur": "Evenement introuvable"}, status_code=404) if event_id == _event_id_actif(): await diffuser_ws({"type": "config_maj", "config": charger_config()}) return event @app.delete("/api/evenements/{event_id}") async def api_supprimer_evenement(event_id: str): etait_actif = event_id == _event_id_actif() ok = supprimer_evenement(event_id) if not ok: return JSONResponse({"erreur": "Evenement introuvable"}, status_code=404) if etait_actif: await diffuser_ws({"type": "config_maj", "config": charger_config()}) return {"succes": True} @app.post("/api/evenements/{event_id}/activer") async def api_activer_evenement(event_id: str): event = activer_evenement(event_id) if not event: return JSONResponse({"erreur": "Evenement introuvable"}, status_code=404) await diffuser_ws({"type": "config_maj", "config": charger_config()}) return event @app.post("/api/evenements/{event_id}/terminer") async def api_terminer_evenement(event_id: str): event = terminer_evenement(event_id) if not event: return JSONResponse({"erreur": "Evenement introuvable"}, status_code=404) await diffuser_ws({"type": "config_maj", "config": charger_config()}) return event @app.get("/api/evenements/{event_id}/cadres/{format_papier}") async def api_cadres_event(event_id: str, format_papier: str): cadres = lister_cadres_event(event_id, format_papier) event = obtenir_evenement(event_id) conf = (event or {}).get("cadres", {}).get(format_papier, {}) return { "disponibles": cadres, "mode": conf.get("mode", "aucun"), "cadre": conf.get("cadre"), } @app.post("/api/evenements/{event_id}/cadres/{format_papier}") async def api_set_cadre_event(event_id: str, format_papier: str, donnees: dict): mode = donnees.get("mode", "aucun") cadre = donnees.get("cadre") set_cadre_event(event_id, format_papier, mode, cadre) if event_id == _event_id_actif(): await diffuser_ws({"type": "config_maj", "config": charger_config()}) return {"succes": True} @app.post("/api/evenements/{event_id}/upload-cadre/{format_papier}") async def api_upload_cadre_event(event_id: str, 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 acceptes"}, status_code=400) dossier = DOSSIER_EVENEMENTS / event_id / "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/evenements/{event_id}/cadres/{format_papier}/{nom}") async def api_supprimer_cadre_event(event_id: str, format_papier: str, nom: str): chemin = DOSSIER_EVENEMENTS / event_id / "cadres" / format_papier / nom if not chemin.exists(): return JSONResponse({"erreur": "Fichier introuvable"}, status_code=404) chemin.unlink() return {"succes": True} # --- 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) result = {"format": format_papier, "disponibles": lister_cadres(format_papier), "actif": actif} event_id = config.get("evenement", {}).get("event_id") if event_id: ev_cadres = lister_cadres_event(event_id, format_papier) ev_cadre, ev_mode = cadre_actif_pour_impression(format_papier) result["event_cadres"] = ev_cadres result["event_mode"] = ev_mode result["event_cadre"] = ev_cadre result["event_id"] = event_id return result @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 Videos didactiques (multi-etapes) --- DOSSIER_VIDEOS = RACINE / "data" / "videos" DOSSIER_VIDEOS.mkdir(parents=True, exist_ok=True) FICHIER_VIDEOS_CONFIG = DOSSIER_VIDEOS / "config.json" SITUATIONS_DEFAUT = [ {"id": "montage", "label": "Montage / Mise en service"}, {"id": "demontage", "label": "Demontage"}, {"id": "camera_hs", "label": "Appareil photo ne repond pas"}, {"id": "bourrage", "label": "Bourrage imprimante"}, {"id": "changement_rouleau", "label": "Changement rouleau / ruban"}, {"id": "depannage_imprimante", "label": "Depannage imprimante"}, {"id": "wifi", "label": "Probleme WiFi"}, ] def charger_videos_config(): if FICHIER_VIDEOS_CONFIG.exists(): with open(FICHIER_VIDEOS_CONFIG) as f: return json.load(f) return {} def sauvegarder_videos_config(cfg): with open(FICHIER_VIDEOS_CONFIG, "w") as f: json.dump(cfg, f, indent=2, ensure_ascii=False) def _lister_etapes(situation: str): dossier = DOSSIER_VIDEOS / situation if not dossier.is_dir(): return [] fichiers = sorted( [f for f in dossier.iterdir() if f.is_file() and f.suffix.lower() in ('.mp4', '.webm', '.mov')], key=lambda f: f.name ) return [{"index": i, "fichier": f.name} for i, f in enumerate(fichiers)] def _migrer_anciens_videos(): for f in DOSSIER_VIDEOS.iterdir(): if f.is_file() and f.suffix.lower() in ('.mp4', '.webm', '.mov'): situation = f.stem dossier = DOSSIER_VIDEOS / situation dossier.mkdir(exist_ok=True) dest = dossier / f"etape_00{f.suffix}" f.rename(dest) _migrer_anciens_videos() @app.get("/api/videos") async def api_videos(): cfg = charger_videos_config() result = {} for sit in SITUATIONS_DEFAUT: sid = sit["id"] etapes = _lister_etapes(sid) sit_cfg = cfg.get(sid, {}) result[sid] = { "label": sit.get("label", sid), "actif": sit_cfg.get("actif", False), "etapes": etapes, "labels_etapes": sit_cfg.get("labels_etapes", {}), } for sid in sorted(cfg.keys()): if sid not in result: etapes = _lister_etapes(sid) result[sid] = { "label": cfg[sid].get("label", sid), "actif": cfg[sid].get("actif", False), "etapes": etapes, "labels_etapes": cfg[sid].get("labels_etapes", {}), } return {"situations": result} @app.post("/api/videos/{situation}/toggle") async def api_toggle_situation(situation: str, req: Request): body = await req.json() cfg = charger_videos_config() if situation not in cfg: cfg[situation] = {} cfg[situation]["actif"] = bool(body.get("actif", False)) sauvegarder_videos_config(cfg) return {"ok": True} @app.post("/api/videos/{situation}/upload") async def api_upload_etape(situation: str, video: UploadFile = File(...)): dossier = DOSSIER_VIDEOS / situation dossier.mkdir(parents=True, exist_ok=True) existantes = sorted(dossier.glob("etape_*")) prochain = len(existantes) ext = Path(video.filename).suffix.lower() or '.mp4' dest = dossier / f"etape_{prochain:02d}{ext}" with open(dest, "wb") as f: while chunk := await video.read(1024 * 1024): f.write(chunk) return {"ok": True, "fichier": dest.name, "index": prochain} @app.delete("/api/videos/{situation}/{index}") async def api_delete_etape(situation: str, index: int): dossier = DOSSIER_VIDEOS / situation if not dossier.is_dir(): return JSONResponse({"erreur": "situation introuvable"}, 404) fichiers = sorted( [f for f in dossier.iterdir() if f.is_file() and f.name.startswith("etape_")], key=lambda f: f.name ) if index < 0 or index >= len(fichiers): return JSONResponse({"erreur": "index invalide"}, 400) fichiers[index].unlink() for i, f in enumerate(sorted( [f for f in dossier.iterdir() if f.is_file() and f.name.startswith("etape_")], key=lambda f: f.name )): nouveau = dossier / f"etape_{i:02d}{f.suffix}" if f != nouveau: f.rename(nouveau) cfg = charger_videos_config() sit_cfg = cfg.get(situation, {}) old_labels = sit_cfg.get("labels_etapes", {}) new_labels = {} for k, v in old_labels.items(): ki = int(k) if ki < index: new_labels[str(ki)] = v elif ki > index: new_labels[str(ki - 1)] = v if old_labels != new_labels: sit_cfg["labels_etapes"] = new_labels cfg[situation] = sit_cfg sauvegarder_videos_config(cfg) return {"ok": True} @app.post("/api/videos/{situation}/reorder") async def api_reorder_etapes(situation: str, req: Request): body = await req.json() old_idx = body.get("de") new_idx = body.get("vers") dossier = DOSSIER_VIDEOS / situation if not dossier.is_dir(): return JSONResponse({"erreur": "situation introuvable"}, 404) fichiers = sorted( [f for f in dossier.iterdir() if f.is_file() and f.name.startswith("etape_")], key=lambda f: f.name ) if old_idx is None or new_idx is None or old_idx < 0 or old_idx >= len(fichiers): return JSONResponse({"erreur": "index invalide"}, 400) moved = fichiers.pop(old_idx) fichiers.insert(new_idx, moved) tmp_names = [] for i, f in enumerate(fichiers): tmp = dossier / f"_tmp_{i:02d}{f.suffix}" f.rename(tmp) tmp_names.append(tmp) for i, tmp in enumerate(tmp_names): ext = tmp.suffix final = dossier / f"etape_{i:02d}{ext}" tmp.rename(final) return {"ok": True} @app.post("/api/videos/{situation}/label") async def api_label_etape(situation: str, req: Request): body = await req.json() index = body.get("index") label = body.get("label", "") cfg = charger_videos_config() if situation not in cfg: cfg[situation] = {} labels = cfg[situation].setdefault("labels_etapes", {}) labels[str(index)] = label sauvegarder_videos_config(cfg) return {"ok": True} @app.get("/api/videos/{situation}/{index}/stream") async def api_stream_etape(situation: str, index: int): dossier = DOSSIER_VIDEOS / situation if not dossier.is_dir(): return JSONResponse({"erreur": "situation introuvable"}, 404) fichiers = sorted( [f for f in dossier.iterdir() if f.is_file() and f.name.startswith("etape_")], key=lambda f: f.name ) if index < 0 or index >= len(fichiers): return JSONResponse({"erreur": "index invalide"}, 404) f = fichiers[index] media = "video/mp4" if f.suffix == ".mp4" else "video/webm" return FileResponse(f, media_type=media) @app.post("/api/videos/situation/creer") async def api_creer_situation(req: Request): body = await req.json() sid = body.get("id", "").strip().lower().replace(" ", "_") label = body.get("label", sid) if not sid: return JSONResponse({"erreur": "id requis"}, 400) cfg = charger_videos_config() if sid not in cfg: cfg[sid] = {"label": label, "actif": False, "labels_etapes": {}} sauvegarder_videos_config(cfg) dossier = DOSSIER_VIDEOS / sid dossier.mkdir(parents=True, exist_ok=True) return {"ok": True} # --- API Surprise --- DOSSIER_SURPRISE = RACINE / "data" / "surprise" DOSSIER_SURPRISE.mkdir(parents=True, exist_ok=True) @app.post("/api/surprise/upload") async def api_upload_surprise(media: UploadFile = File(...)): for old in DOSSIER_SURPRISE.iterdir(): old.unlink() ext = Path(media.filename).suffix.lower() or '.jpg' dest = DOSSIER_SURPRISE / f"surprise{ext}" with open(dest, "wb") as f: while chunk := await media.read(1024 * 1024): f.write(chunk) fichier_type = "video" if ext in ('.mp4', '.webm', '.mov') else "photo" mettre_a_jour_config({"surprise": {"fichier": dest.name, "type": fichier_type}}) return {"ok": True, "fichier": dest.name, "type": fichier_type} @app.delete("/api/surprise/media") async def api_delete_surprise(): for f in DOSSIER_SURPRISE.iterdir(): f.unlink() mettre_a_jour_config({"surprise": {"fichier": None}}) return {"ok": True} @app.get("/api/surprise/media") async def api_get_surprise(): for f in DOSSIER_SURPRISE.iterdir(): if f.is_file(): ext = f.suffix.lower() if ext in ('.mp4', '.webm', '.mov'): return FileResponse(f, media_type="video/mp4") return FileResponse(f, media_type="image/jpeg") return JSONResponse({"erreur": "aucun media"}, 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} @app.post("/api/systeme/redemarrer-backend") async def api_redemarrer_backend(): """Tue le processus backend — le kiosk-session.sh le relance automatiquement.""" log.warning("Redemarrage backend demande via API") import signal asyncio.get_event_loop().call_later(0.5, lambda: os.kill(os.getpid(), signal.SIGTERM)) return {"succes": True, "message": "Backend redémarre dans 1 seconde..."} @app.get("/api/systeme/logs") async def api_systeme_logs(n: int = 80): """Retourne les dernières N lignes du log backend.""" n = min(n, 500) log_file = Path("/tmp/photobooth-backend.log") if not log_file.exists(): return {"lignes": [], "fichier": str(log_file), "existe": False} try: with open(log_file, "rb") as f: f.seek(0, 2) taille = f.tell() bloc = min(taille, n * 200) f.seek(max(0, taille - bloc)) data = f.read().decode("utf-8", errors="replace") lignes = data.splitlines()[-n:] return {"lignes": lignes, "fichier": str(log_file), "existe": True} except Exception as e: return {"lignes": [f"Erreur lecture log : {e}"], "fichier": str(log_file), "existe": True} @app.get("/api/systeme/sante") async def api_systeme_sante(): """Bilan de santé complet : camera, imprimante, relais, USB, système.""" import subprocess, shutil # Camera cam_ok = camera.connectee and camera.mode == "gphoto2" cam_status = "ok" if cam_ok else ("erreur" if camera.mode == "erreur" else camera.mode) # Imprimante config = charger_config() nom_imp = config.get("impression", {}).get("imprimante") or "Mitsubishi" imp_status = "inconnu" try: r = subprocess.run(["lpstat", "-p", nom_imp], capture_output=True, text=True, timeout=5) out = r.stdout.lower() if "disabled" in out or "stopped" in out: imp_status = "arretee" elif "idle" in out or "inactive" in out: imp_status = "prete" elif "printing" in out: imp_status = "impression" else: imp_status = "inconnu" except Exception: imp_status = "erreur" # Jobs en file jobs_en_attente = 0 try: r = subprocess.run(["lpstat", "-o", nom_imp], capture_output=True, text=True, timeout=5) jobs_en_attente = len([l for l in r.stdout.splitlines() if l.strip()]) except Exception: pass # Relais rel_ok = relais.est_connecte() # USB Canon détecté canon_usb = False try: import glob for f in glob.glob("/sys/bus/usb/devices/*/idVendor"): if open(f).read().strip() == "04a9": canon_usb = True break except Exception: pass # USB imprimante détectée (Mitsubishi vendor 06d3) imp_usb = False try: import glob for f in glob.glob("/sys/bus/usb/devices/*/idVendor"): if open(f).read().strip() == "06d3": imp_usb = True break except Exception: pass # RAM & disque ram_pct = 0 try: with open('/proc/meminfo') as f: lignes = {l.split(':')[0]: l.split(':')[1].strip() for l in f.readlines()} total = int(lignes.get('MemTotal', '1 kB').split()[0]) libre = int(lignes.get('MemAvailable', '0 kB').split()[0]) ram_pct = round((total - libre) / total * 100) except Exception: pass disque_pct = 0 try: total, used, _ = shutil.disk_usage("/") disque_pct = round(used / total * 100) except Exception: pass # Uptime uptime = "" try: with open('/proc/uptime') as f: secs = int(float(f.read().split()[0])) h, rem = divmod(secs, 3600) m = rem // 60 uptime = f"{h}h{m:02d}" except Exception: pass # Temperature temp = None try: with open('/sys/class/thermal/thermal_zone0/temp') as f: temp = round(int(f.read().strip()) / 1000, 1) except Exception: pass return { "camera": {"status": cam_status, "mode": camera.mode, "usb": canon_usb, "preview_ok": camera.preview_dslr_ok}, "imprimante": {"status": imp_status, "nom": nom_imp, "usb": imp_usb, "jobs": jobs_en_attente}, "relais": {"connecte": rel_ok}, "systeme": {"ram_pct": ram_pct, "disque_pct": disque_pct, "uptime": uptime, "temp": temp}, } @app.post("/api/imprimante/reactiver") async def api_reactiver_imprimante(): """Réactive l'imprimante si elle est en erreur/stoppée.""" import subprocess nom = charger_config().get("impression", {}).get("imprimante") or "Mitsubishi" try: subprocess.run(["sudo", "cupsenable", nom], capture_output=True, timeout=5) subprocess.run(["sudo", "accept", nom], capture_output=True, timeout=5) log.info(f"Imprimante {nom} réactivée manuellement") return {"succes": True, "message": f"{nom} réactivée"} except Exception as e: return {"succes": False, "message": str(e)} @app.get("/api/systeme/usb") async def api_systeme_usb(): """Liste les périphériques USB connectés.""" import subprocess try: r = subprocess.run(["lsusb"], capture_output=True, text=True, timeout=5) peripheriques = [] for l in r.stdout.splitlines(): peripheriques.append(l.strip()) return {"peripheriques": peripheriques} except Exception: return {"peripheriques": ["lsusb non disponible"]} # --- API WiFi --- @app.get("/api/wifi/status") async def api_wifi_status(): loop = asyncio.get_event_loop() return await loop.run_in_executor(None, wifi_status) @app.get("/api/wifi/scan") async def api_wifi_scan(): loop = asyncio.get_event_loop() return await loop.run_in_executor(None, wifi_scan) @app.post("/api/wifi/connect") async def api_wifi_connect(donnees: dict): ssid = donnees.get("ssid", "") password = donnees.get("password") if not ssid: return JSONResponse({"erreur": "SSID requis"}, status_code=400) loop = asyncio.get_event_loop() return await loop.run_in_executor(None, wifi_connect, ssid, password) @app.get("/api/wifi/saved") async def api_wifi_saved(): loop = asyncio.get_event_loop() return await loop.run_in_executor(None, wifi_saved_list) @app.delete("/api/wifi/saved/{ssid}") async def api_wifi_forget(ssid: str): loop = asyncio.get_event_loop() return await loop.run_in_executor(None, wifi_forget, ssid) @app.get("/api/wifi/password/{ssid}") async def api_wifi_password(ssid: str): mdp = wifi_get_password(ssid) return {"ssid": ssid, "password": mdp} # --- 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, _canon_en_veille _preview_actif = True if _canon_en_veille: log.info("WS: preview_start — réveil Canon depuis veille relais") _canon_en_veille = False await asyncio.get_event_loop().run_in_executor(None, _reveiller_canon) else: log.info("WS: preview_start recu") elif type_msg == "prepare_capture": config = charger_config() if config.get("relais", {}).get("eclairage_auto") and relais.est_connecte(): relais.projecteurs(True) _proj_etat["proj1"] = True _proj_etat["proj2"] = True log.info("WS: prepare_capture — 2 projecteurs ON (pré-éclairage countdown)") elif type_msg == "preview_stop": _preview_actif = False global _preview_eclairage_init _preview_eclairage_init = False if not capture_en_cours and relais.est_connecte() and (_proj_etat["proj1"] or _proj_etat["proj2"]): relais.projecteurs(False) _proj_etat["proj1"] = False _proj_etat["proj2"] = False log.info("WS: preview_stop recu" + (" (projecteurs maintenus pour capture)" if capture_en_cours else "")) 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( app, host=conf_srv.get("host", "0.0.0.0"), port=conf_srv.get("port", 80), log_level="info", )