Jules
1364b62297
- _thread_preview tourne TOUJOURS quand DSLR connecté (plus de condition _preview_actif) → capture_preview() appelé en continu → miroir reste levé - Grace period 15 frames (~500ms) avant de signaler camera_erreur : évite la race condition qui tuait le preview avant la 1ère frame - _preview_gphoto2 protégé par _gp_lock (accès concurrent safe) - surveiller_dslr ne tourne plus quand le DSLR est en mode gphoto2 connecté Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
305 lines
11 KiB
Python
305 lines
11 KiB
Python
import io
|
|
import logging
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from datetime import datetime
|
|
|
|
import numpy as np
|
|
from pathlib import Path
|
|
|
|
import cv2
|
|
|
|
from backend.config import DOSSIER_PHOTOS, charger_config
|
|
|
|
log = logging.getLogger("photobooth.camera")
|
|
|
|
# Essayer d'importer gphoto2, sinon mode simulation
|
|
try:
|
|
import gphoto2 as gp
|
|
GPHOTO2_DISPONIBLE = True
|
|
except ImportError:
|
|
GPHOTO2_DISPONIBLE = False
|
|
log.warning("python-gphoto2 non installe")
|
|
|
|
|
|
def lister_webcams() -> list[dict]:
|
|
"""Liste les webcams disponibles via v4l2."""
|
|
webcams = []
|
|
try:
|
|
result = subprocess.run(
|
|
["v4l2-ctl", "--list-devices"],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
if result.returncode == 0:
|
|
lines = result.stdout.strip().split("\n")
|
|
nom = ""
|
|
for line in lines:
|
|
if not line.startswith("\t"):
|
|
nom = line.strip().rstrip(":")
|
|
elif "/dev/video" in line:
|
|
dev = line.strip()
|
|
# Ne garder que les devices principaux (pas metadata)
|
|
idx = int(dev.replace("/dev/video", ""))
|
|
cap = cv2.VideoCapture(idx)
|
|
if cap.isOpened():
|
|
cap.release()
|
|
webcams.append({"nom": nom, "device": dev, "index": idx})
|
|
nom = "" # Eviter les doublons
|
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
|
# v4l2-ctl non installe, fallback
|
|
for i in range(4):
|
|
cap = cv2.VideoCapture(i)
|
|
if cap.isOpened():
|
|
cap.release()
|
|
webcams.append({"nom": f"Camera {i}", "device": f"/dev/video{i}", "index": i})
|
|
return webcams
|
|
|
|
|
|
class Camera:
|
|
"""Controle camera : DSLR (gphoto2), webcam (OpenCV) ou simulation."""
|
|
|
|
def __init__(self):
|
|
self.camera = None
|
|
self.contexte = None
|
|
self.connectee = False
|
|
self.mode = "simulation" # "gphoto2", "webcam", "simulation"
|
|
self.webcam = None
|
|
self.webcam_index = -1
|
|
self.preview_dslr_ok = True # True si le dernier preview DSLR a reussi
|
|
self._gp_lock = threading.Lock()
|
|
|
|
def connecter(self, source=None) -> bool:
|
|
"""Connecte la camera. source: 'gphoto2', 'webcam:0', 'webcam:2', etc."""
|
|
self.deconnecter()
|
|
|
|
if source and source.startswith("webcam:"):
|
|
idx = int(source.split(":")[1])
|
|
return self._connecter_webcam(idx)
|
|
|
|
if source == "gphoto2" or source is None:
|
|
if GPHOTO2_DISPONIBLE:
|
|
try:
|
|
self.camera = gp.Camera()
|
|
self.camera.init()
|
|
self.connectee = True
|
|
self.mode = "gphoto2"
|
|
log.info("Camera DSLR connectee")
|
|
self.activer_viewfinder()
|
|
return True
|
|
except gp.GPhoto2Error as e:
|
|
log.warning(f"Pas de DSLR : {e}")
|
|
|
|
# Pas de DSLR disponible : mode erreur, pas de fallback
|
|
log.warning("Aucun DSLR disponible, camera non connectee")
|
|
self.connectee = False
|
|
self.mode = "erreur"
|
|
return False
|
|
|
|
def _connecter_webcam(self, index: int) -> bool:
|
|
"""Connecte une webcam via OpenCV."""
|
|
cap = cv2.VideoCapture(index)
|
|
if cap.isOpened():
|
|
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1920)
|
|
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 1080)
|
|
self.webcam = cap
|
|
self.webcam_index = index
|
|
self.connectee = True
|
|
self.mode = "webcam"
|
|
log.info(f"Webcam connectee : /dev/video{index}")
|
|
return True
|
|
log.error(f"Impossible d'ouvrir /dev/video{index}")
|
|
return False
|
|
|
|
def deconnecter(self):
|
|
"""Deconnecte l'appareil photo."""
|
|
if self.camera and GPHOTO2_DISPONIBLE:
|
|
try:
|
|
self.camera.exit()
|
|
except gp.GPhoto2Error:
|
|
pass
|
|
if self.webcam:
|
|
self.webcam.release()
|
|
self.camera = None
|
|
self.contexte = None
|
|
self.webcam = None
|
|
self.webcam_index = -1
|
|
self.connectee = False
|
|
self.mode = "simulation"
|
|
log.info("Camera deconnectee")
|
|
|
|
def capturer(self) -> Path | None:
|
|
"""Capture une photo et retourne le chemin du fichier."""
|
|
if not self.connectee:
|
|
log.error("Camera non connectee")
|
|
return None
|
|
|
|
horodatage = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
|
|
nom_fichier = f"photo_{horodatage}.jpg"
|
|
chemin_dest = DOSSIER_PHOTOS / nom_fichier
|
|
|
|
if self.mode == "webcam":
|
|
return self._capture_webcam(chemin_dest)
|
|
elif self.mode == "gphoto2":
|
|
return self._capture_gphoto2(chemin_dest)
|
|
else:
|
|
return self._capture_simulation(chemin_dest)
|
|
|
|
def _capture_gphoto2(self, chemin_dest: Path) -> Path | None:
|
|
from PIL import Image as PILImage
|
|
|
|
for attempt in range(3):
|
|
try:
|
|
with self._gp_lock:
|
|
chemin_camera = self.camera.capture(gp.GP_CAPTURE_IMAGE)
|
|
fichier_camera = gp.CameraFile()
|
|
self.camera.file_get(
|
|
chemin_camera.folder, chemin_camera.name, gp.GP_FILE_TYPE_NORMAL, fichier_camera
|
|
)
|
|
tmp_path = str(chemin_dest) + ".tmp"
|
|
fichier_camera.save(tmp_path)
|
|
self.activer_viewfinder() # Miroir relevé immédiatement après la prise
|
|
img = PILImage.open(tmp_path)
|
|
if img.width > 4000:
|
|
ratio = 4000 / img.width
|
|
img = img.resize((4000, int(img.height * ratio)), PILImage.LANCZOS)
|
|
img.save(str(chemin_dest), "JPEG", quality=92)
|
|
Path(tmp_path).unlink(missing_ok=True)
|
|
log.info(f"Photo capturee (DSLR) : {chemin_dest} ({img.width}x{img.height})")
|
|
return chemin_dest
|
|
except gp.GPhoto2Error as e:
|
|
if "I/O in progress" in str(e) and attempt < 2:
|
|
log.warning(f"DSLR occupe, retry {attempt + 1}/3...")
|
|
time.sleep(0.5)
|
|
continue
|
|
log.error(f"Erreur capture DSLR : {e}")
|
|
return None
|
|
|
|
def _capture_webcam(self, chemin_dest: Path) -> Path | None:
|
|
if not self.webcam or not self.webcam.isOpened():
|
|
return None
|
|
ret, frame = self.webcam.read()
|
|
if ret:
|
|
cv2.imwrite(str(chemin_dest), frame, [cv2.IMWRITE_JPEG_QUALITY, 95])
|
|
log.info(f"Photo capturee (webcam) : {chemin_dest}")
|
|
return chemin_dest
|
|
log.error("Erreur capture webcam")
|
|
return None
|
|
|
|
def preview(self) -> bytes | None:
|
|
"""Capture un apercu live (preview) en JPEG."""
|
|
if not self.connectee:
|
|
return None
|
|
|
|
if self.mode == "webcam":
|
|
return self._preview_webcam()
|
|
elif self.mode == "gphoto2":
|
|
data = self._preview_gphoto2()
|
|
self.preview_dslr_ok = data is not None
|
|
return data # None si le DSLR ne repond pas, pas de fallback
|
|
else:
|
|
return None # Pas de simulation ni webcam en dehors du mode dedie
|
|
|
|
def activer_viewfinder(self):
|
|
"""Active le LiveView (miroir levé en permanence) sur les Canon."""
|
|
if self.mode != "gphoto2" or not self.connectee:
|
|
return
|
|
try:
|
|
with self._gp_lock:
|
|
cfg = self.camera.get_config()
|
|
vf = cfg.get_child_by_name("viewfinder")
|
|
vf.set_value(1)
|
|
self.camera.set_config(cfg)
|
|
log.info("Viewfinder activé")
|
|
except Exception as e:
|
|
log.warning(f"Viewfinder non supporté : {e}")
|
|
|
|
def _preview_gphoto2(self) -> bytes | None:
|
|
try:
|
|
with self._gp_lock:
|
|
fichier = self.camera.capture_preview()
|
|
donnees = bytes(fichier.get_data_and_size())
|
|
img = cv2.imdecode(np.frombuffer(donnees, dtype=np.uint8), cv2.IMREAD_COLOR)
|
|
if img is not None:
|
|
h, w = img.shape[:2]
|
|
if w > 960:
|
|
img = cv2.resize(img, (960, int(h * 960 / w)), interpolation=cv2.INTER_LINEAR)
|
|
_, buf = cv2.imencode(".jpg", img, [cv2.IMWRITE_JPEG_QUALITY, 75])
|
|
return buf.tobytes()
|
|
return donnees
|
|
except gp.GPhoto2Error as e:
|
|
log.error(f"Erreur preview DSLR : {e}")
|
|
return None
|
|
|
|
def _preview_webcam(self) -> bytes | None:
|
|
if not self.webcam or not self.webcam.isOpened():
|
|
return None
|
|
ret, frame = self.webcam.read()
|
|
if ret:
|
|
_, buf = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 70])
|
|
return buf.tobytes()
|
|
return None
|
|
|
|
def lister_appareils(self) -> list[dict]:
|
|
"""Liste tous les appareils disponibles (DSLR + webcams)."""
|
|
appareils = []
|
|
|
|
# Webcams
|
|
for wc in lister_webcams():
|
|
appareils.append({
|
|
"nom": wc["nom"],
|
|
"source": f"webcam:{wc['index']}",
|
|
"type": "webcam"
|
|
})
|
|
|
|
# DSLR via gphoto2
|
|
if GPHOTO2_DISPONIBLE:
|
|
try:
|
|
dslrs = gp.Camera.autodetect()
|
|
for nom, port in dslrs:
|
|
appareils.append({
|
|
"nom": f"{nom} ({port})",
|
|
"source": "gphoto2",
|
|
"type": "dslr"
|
|
})
|
|
except gp.GPhoto2Error:
|
|
pass
|
|
|
|
if not appareils:
|
|
appareils.append({
|
|
"nom": "Simulation",
|
|
"source": "simulation",
|
|
"type": "simulation"
|
|
})
|
|
|
|
return appareils
|
|
|
|
def _capture_simulation(self, chemin: Path) -> Path:
|
|
"""Genere une image de test en mode simulation."""
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
|
|
img = Image.new("RGB", (1920, 1280), color=(40, 40, 40))
|
|
draw = ImageDraw.Draw(img)
|
|
texte = f"SIMULATION\n{datetime.now().strftime('%H:%M:%S')}"
|
|
draw.text((960, 640), texte, fill=(255, 255, 255), anchor="mm")
|
|
draw.rectangle([50, 50, 1870, 1230], outline=(233, 30, 99), width=4)
|
|
img.save(chemin, "JPEG", quality=95)
|
|
log.info(f"Photo simulation : {chemin}")
|
|
return chemin
|
|
|
|
def _preview_simulation(self) -> bytes:
|
|
"""Genere un apercu de test en mode simulation."""
|
|
from PIL import Image, ImageDraw
|
|
|
|
img = Image.new("RGB", (640, 480), color=(30, 30, 30))
|
|
draw = ImageDraw.Draw(img)
|
|
draw.text((320, 240), "PREVIEW", fill=(200, 200, 200), anchor="mm")
|
|
draw.ellipse([300, 220, 340, 260], outline=(233, 30, 99), width=2)
|
|
buf = io.BytesIO()
|
|
img.save(buf, "JPEG", quality=70)
|
|
return buf.getvalue()
|
|
|
|
|
|
# Instance globale
|
|
camera = Camera()
|