photobooth/backend/camera.py

import io
import logging
import subprocess
import time
from datetime import datetime
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

    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.contexte = gp.Context()
                    self.camera = gp.Camera()
                    self.camera.init(self.contexte)
                    self.connectee = True
                    self.mode = "gphoto2"
                    log.info("Camera DSLR connectee")
                    return True
                except gp.GPhoto2Error as e:
                    log.warning(f"Pas de DSLR : {e}")

        # Fallback webcam si pas de source specifique
        if source is None:
            webcams = lister_webcams()
            if webcams:
                return self._connecter_webcam(webcams[0]["index"])

        # Fallback simulation
        log.info("Mode simulation active")
        self.connectee = True
        self.mode = "simulation"
        return True

    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(self.contexte)
            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:
        try:
            chemin_camera = self.camera.capture(gp.GP_CAPTURE_IMAGE, self.contexte)
            fichier_camera = self.camera.file_get(
                chemin_camera.folder, chemin_camera.name, gp.GP_FILE_TYPE_NORMAL, self.contexte
            )
            fichier_camera.save(str(chemin_dest))
            log.info(f"Photo capturee (DSLR) : {chemin_dest}")
            return chemin_dest
        except gp.GPhoto2Error as e:
            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":
            return self._preview_gphoto2()
        else:
            return self._preview_simulation()

    def _preview_gphoto2(self) -> bytes | None:
        try:
            fichier = self.camera.capture_preview(self.contexte)
            donnees = fichier.get_data_and_size()
            return bytes(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:
                ctx = self.contexte or gp.Context()
                dslrs = gp.Camera.autodetect(ctx)
                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()