jules/photobooth
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Robustesse DSLR + éclairage visage + fix pellicule cadre imposé

Browse Source 
- camera.py: fix fuite FD USB sur échec reconnexion (gp.Camera non libéré),
  cleanup PTP session avant init, catch toutes exceptions preview/capture,
  deconnecter() avec timeout thread + force del pour libérer le port USB
- main.py: thread preview anti-crash (outer try/except), reconnexion DSLR
  même si preview actif, notification camera_erreur après échec capture,
  intégration module éclairage (analyse luminosité visage via WS)
- printer.py: mode pellicule (-2up) ignoré par le cadre imposé événement
- eclairage.py: nouveau module analyse luminosité visage (Haar + HSV)
- camera.js: indicateur éclairage temps réel (score/100 + visages)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Jules
2026-06-25 19:11:10 +02:00
parent 182042350d
commit 48f54ab70a
5 changed files with 219 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
backend/camera.py
View File

@@ -79,19 +79,31 @@ class Camera:
if source == "gphoto2" or source is None:
if GPHOTO2_DISPONIBLE:
cam = None
try:
self.camera = gp.Camera()
self.camera.init()
# Desactiver la mise en veille auto du DSLR (sinon il se deconnecte tout seul)
cam = gp.Camera()
try:
cam.exit()
except Exception:
pass
cam.init()
self.camera = cam
self._desactiver_veille_init()
# Viewfinder activé AVANT de rendre la caméra visible au thread preview
self._activer_viewfinder_init()
self.connectee = True
self.mode = "gphoto2"
self.preview_dslr_ok = True
log.info("Camera DSLR connectee (LiveView actif)")
return True
except gp.GPhoto2Error as e:
except Exception as e:
log.warning(f"Pas de DSLR : {e}")
if cam is not None and cam is not self.camera:
try:
cam.exit()
except Exception:
pass
del cam
self.camera = None
# Pas de DSLR disponible : mode erreur, pas de fallback
log.warning("Aucun DSLR disponible, camera non connectee")
@@ -116,25 +128,30 @@ class Camera:
def deconnecter(self, timeout: float = 3.0):
"""Deconnecte l'appareil photo avec timeout pour eviter le blocage PTP."""
if self.camera and GPHOTO2_DISPONIBLE:
cam_ref = self.camera
ctx_ref = self.contexte
if cam_ref and GPHOTO2_DISPONIBLE:
def _exit():
try:
self.camera.exit()
cam_ref.exit()
except Exception:
pass
t = threading.Thread(target=_exit, daemon=True)
t.start()
t.join(timeout=timeout)
if t.is_alive():
log.warning("camera.exit() bloque — abandon (le FD sera libere au GC)")
log.warning("camera.exit() bloque — force del camera/contexte pour liberer le FD USB")
if self.webcam:
self.webcam.release()
self.camera = None
self.contexte = None
del cam_ref
del ctx_ref
self.webcam = None
self.webcam_index = -1
self.connectee = False
self.mode = "simulation"
self.preview_dslr_ok = False
log.info("Camera deconnectee")
def capturer(self) -> Path | None:
@@ -160,13 +177,16 @@ class Camera:
for attempt in range(3):
try:
with self._gp_lock:
# Désactiver LiveView : miroir abaissé → AF phase-détection Canon disponible
if not self.camera:
log.error("Camera None pendant capture")
self.preview_dslr_ok = False
return None
try:
cfg = self.camera.get_config()
vf = cfg.get_child_by_name("viewfinder")
vf.set_value(0)
self.camera.set_config(cfg)
time.sleep(0.2) # Laisser le miroir descendre
time.sleep(0.2)
except Exception:
pass
log.info(f"Déclenchement capture DSLR (tentative {attempt+1}/3)")
@@ -177,8 +197,8 @@ class Camera:
)
tmp_path = str(chemin_dest) + ".tmp"
fichier_camera.save(tmp_path)
self.activer_viewfinder() # Miroir relevé immédiatement après la prise
self._chauffer_liveview() # Absorbe la stabilisation LiveView ici (spinner visible), pas sur le countdown suivant
self.activer_viewfinder()
self._chauffer_liveview()
from PIL import ImageOps as PILImageOps
img = PILImageOps.exif_transpose(PILImage.open(tmp_path))
if img.width > 4000:
@@ -194,6 +214,11 @@ class Camera:
time.sleep(0.5)
continue
log.error(f"Erreur capture DSLR : {e}")
self.preview_dslr_ok = False
return None
except Exception as e:
log.error(f"Erreur fatale capture DSLR : {e}")
self.preview_dslr_ok = False
return None
def _chauffer_liveview(self, tentatives: int = 8, pause: float = 0.15):
@@ -308,6 +333,8 @@ class Camera:
def _preview_gphoto2(self) -> bytes | None:
try:
with self._gp_lock:
if not self.camera:
return None
fichier = self.camera.capture_preview()
donnees = bytes(fichier.get_data_and_size())
img = cv2.imdecode(np.frombuffer(donnees, dtype=np.uint8), cv2.IMREAD_COLOR)
@@ -318,7 +345,7 @@ class Camera:
_, buf = cv2.imencode(".jpg", img, [cv2.IMWRITE_JPEG_QUALITY, 75])
return buf.tobytes()
return donnees
except gp.GPhoto2Error as e:
except Exception as e:
log.error(f"Erreur preview DSLR : {e}")
return None

114
backend/eclairage.py Normal file
View File

@@ -0,0 +1,114 @@
"""Analyse de luminosité sur les visages détectés dans le preview DSLR.
Produit une note d'éclairage (0-100) et une recommandation (allumer/ok/sombre).
Échantillonne toutes les ~3 secondes pour ne pas charger le CPU.
"""
import logging
import time
from pathlib import Path
import cv2
import numpy as np
log = logging.getLogger("photobooth.eclairage")
_CASCADE_CANDIDATES = [
"/usr/share/opencv4/haarcascades/haarcascade_frontalface_default.xml",
"/usr/share/opencv/haarcascades/haarcascade_frontalface_default.xml",
"/usr/local/share/opencv4/haarcascades/haarcascade_frontalface_default.xml",
]
try:
_CASCADE_PATH = cv2.data.haarcascades + "haarcascade_frontalface_default.xml"
except AttributeError:
_CASCADE_PATH = next((p for p in _CASCADE_CANDIDATES if Path(p).exists()), _CASCADE_CANDIDATES[0])
_face_cascade = cv2.CascadeClassifier(_CASCADE_PATH)
SEUIL_SOMBRE = 45
SEUIL_CORRECT = 65
_dernier_analyse: float = 0
_INTERVALLE = 3.0
_dernier_resultat: dict | None = None
def analyser_frame(jpeg_data: bytes) -> dict | None:
"""Analyse une frame JPEG du preview.
Retourne None si l'intervalle n'est pas écoulé (throttle).
Sinon retourne:
{
"score": int 0-100,
"action": "allumer" | "attention" | "ok",
"visages": int,
"detail": str,
}
"""
global _dernier_analyse, _dernier_resultat
now = time.monotonic()
if now - _dernier_analyse < _INTERVALLE:
return None
_dernier_analyse = now
try:
arr = np.frombuffer(jpeg_data, dtype=np.uint8)
img = cv2.imdecode(arr, cv2.IMREAD_COLOR)
if img is None:
return None
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
faces = _face_cascade.detectMultiScale(
gray, scaleFactor=1.2, minNeighbors=4, minSize=(60, 60)
)
if len(faces) == 0:
_dernier_resultat = {
"score": -1,
"action": "no_face",
"visages": 0,
"detail": "Aucun visage détecté",
}
return _dernier_resultat
hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
scores = []
for (x, y, w, h) in faces:
pad_x = int(w * 0.1)
pad_y = int(h * 0.1)
x1 = max(0, x + pad_x)
y1 = max(0, y + pad_y)
x2 = min(img.shape[1], x + w - pad_x)
y2 = min(img.shape[0], y + h - pad_y)
roi_v = hsv[y1:y2, x1:x2, 2]
score = int(np.mean(roi_v) / 2.55)
scores.append(score)
score_moyen = int(np.mean(scores))
if score_moyen < SEUIL_SOMBRE:
action = "allumer"
detail = f"Visages sombres ({score_moyen}/100)"
elif score_moyen < SEUIL_CORRECT:
action = "attention"
detail = f"Éclairage limite ({score_moyen}/100)"
else:
action = "ok"
detail = f"Éclairage correct ({score_moyen}/100)"
_dernier_resultat = {
"score": score_moyen,
"action": action,
"visages": len(faces),
"detail": detail,
}
return _dernier_resultat
except Exception as e:
log.debug(f"Erreur analyse éclairage : {e}")
return None
def dernier_resultat() -> dict | None:
return _dernier_resultat

42
backend/main.py
View File

@@ -36,6 +36,7 @@ from backend.evenements import (
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
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(message)s")
log = logging.getLogger("photobooth")
@@ -63,17 +64,21 @@ def _thread_preview():
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
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
except Exception as e:
log.error(f"_thread_preview crash: {e}")
time.sleep(1)
async def _pusher_preview():
@@ -103,6 +108,10 @@ async def _pusher_preview():
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})
def _usb_reset_canon():
"""Reset USB ioctl du Canon EOS (vendor 04a9) pour débloquer un port stall."""
@@ -206,7 +215,7 @@ async def surveiller_dslr():
_echecs_connexion = 0
else:
_dslr_erreurs += 1
if _dslr_erreurs >= 3 and not _preview_actif:
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
@@ -409,6 +418,9 @@ async def api_capturer():
capture_en_cours = 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
@@ -547,6 +559,14 @@ async def api_camera_statut():
}
@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
@app.get("/api/camera/debug")
async def api_camera_debug():
"""Diagnostic complet de la caméra (preview, viewfinder, erreurs gphoto2)."""

5
backend/printer.py
View File

@@ -258,14 +258,15 @@ def imprimer(
largeur, hauteur = fmt["px"]
fmt_base = format_papier.replace("-2up", "")
cadre = cadre_override if cadre_override is not None else conf_imp.get("cadres_actifs", {}).get(fmt_base)
is_strip = "-2up" in format_papier
cadre = cadre_override if cadre_override is not None else (None if is_strip else conf_imp.get("cadres_actifs", {}).get(fmt_base))
orientation = conf_imp.get("orientations", {}).get(fmt_base, "portrait")
if orientation == "paysage" and hauteur > largeur:
largeur, hauteur = hauteur, largeur
event_id = config.get("evenement", {}).get("event_id")
if event_id and cadre_override is None:
if event_id and cadre_override is None and not is_strip:
from backend.evenements import cadre_actif_pour_impression
ev_cadre, ev_mode = cadre_actif_pour_impression(fmt_base)
if ev_mode == "impose" and ev_cadre: