Jules
e4e5ee05d4
Quand gphoto2 détecte le DSLR mais que camera.init() timeout répétitivement (USB stall après kill -9 ou déconnexion brutale), on tente un reset USB ioctl (USBDEVFS_RESET sur /dev/bus/usb/<bus>/<dev>) toutes les 3 tentatives (~15s). Évite de rester bloqué indéfiniment sur "Timeout reading from or writing to the port". Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1109 lines
39 KiB
Python
1109 lines
39 KiB
Python
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}"})
|
|
|
|
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}")
|
|
|
|
|
|
async def surveiller_dslr():
|
|
"""Surveille le DSLR en continu : detecte les deconnexions et reconnecte automatiquement."""
|
|
global _dslr_erreurs
|
|
_echecs_connexion = 0
|
|
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
|
|
_echecs_connexion = 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")
|
|
_echecs_connexion = 0
|
|
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")
|
|
_echecs_connexion = 0
|
|
await diffuser_ws({"type": "camera_ok"})
|
|
else:
|
|
_echecs_connexion += 1
|
|
log.warning(f"Echec connexion DSLR ({_echecs_connexion})")
|
|
# Après 3 échecs consécutifs (~15s), reset USB pour débloquer un port stall
|
|
if _echecs_connexion % 3 == 0:
|
|
log.warning("Reset USB Canon tentative de deblocage...")
|
|
_usb_reset_canon()
|
|
await asyncio.sleep(3)
|
|
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("<h1>Code invalide</h1>", 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'<div class="photo"><img src="/data/photos/session_{code}/{p.name}" loading="lazy" onclick="agrandir(this)"></div>\n'
|
|
if not items:
|
|
items = '<p class="vide">Les photos apparaîtront ici après chaque prise 📷</p>'
|
|
html = f"""<!DOCTYPE html>
|
|
<html lang="fr"><head>
|
|
<meta charset="UTF-8"><meta name="viewport" content="width=device-width,initial-scale=1">
|
|
<title>{nom_event} — Galerie</title>
|
|
<style>
|
|
*{{margin:0;padding:0;box-sizing:border-box}}
|
|
body{{background:#0a0a1a;color:#fff;font-family:system-ui,sans-serif;min-height:100vh}}
|
|
header{{background:#111;padding:1rem;text-align:center;border-bottom:2px solid #e91e63}}
|
|
header h1{{font-size:1.4rem;color:#e91e63}}
|
|
header p{{font-size:.85rem;color:#aaa;margin-top:.3rem}}
|
|
.grille{{display:grid;grid-template-columns:repeat(auto-fill,minmax(160px,1fr));gap:6px;padding:8px}}
|
|
.photo img{{width:100%;aspect-ratio:3/2;object-fit:cover;border-radius:6px;cursor:pointer;transition:transform .2s}}
|
|
.photo img:active{{transform:scale(.97)}}
|
|
.vide{{text-align:center;padding:3rem;color:#666;font-size:1.1rem}}
|
|
.lightbox{{display:none;position:fixed;inset:0;background:#000d;align-items:center;justify-content:center;z-index:99}}
|
|
.lightbox.open{{display:flex}}
|
|
.lightbox img{{max-width:95vw;max-height:95vh;border-radius:8px}}
|
|
.lightbox::after{{content:"✕";position:absolute;top:1rem;right:1.2rem;font-size:2rem;cursor:pointer;color:#fff}}
|
|
</style></head><body>
|
|
<header><h1>📷 {nom_event}</h1><p>Code session : {code}</p></header>
|
|
<div class="grille">{items}</div>
|
|
<div class="lightbox" id="lb" onclick="this.classList.remove('open')"><img id="lb-img" src=""></div>
|
|
<script>
|
|
function agrandir(el){{document.getElementById('lb-img').src=el.src;document.getElementById('lb').classList.add('open')}}
|
|
// Rafraîchissement auto toutes les 15s
|
|
setTimeout(()=>location.reload(), 15000)
|
|
</script>
|
|
</body></html>"""
|
|
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, _preview_actif
|
|
# 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 (race condition HTTP avant preview_stop)
|
|
log.info("Capture déclenchée — preview figé, thread en attente de lock")
|
|
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
|
|
log.info(f"Capture terminée : {chemin}")
|
|
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",
|
|
)
|