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()