Jules
f49d5511ef
La coupe physique du K60 n'est jamais parfaitement centree, donc une marge noire trop fine fait perdre la pellicule d'un cote du strip dedouble. Marge de coupe (perforations) elargie de 34 a 55px (~4.6mm) de chaque cote. Ajoute aussi un warmup LiveView (_chauffer_liveview) juste apres la remontee du miroir post-capture : consomme les ~1-2s de stabilisation pendant le spinner de traitement plutot que de laisser le preview du countdown suivant figé sur l'ancienne photo.
390 lines
15 KiB
Python
390 lines
15 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()
|
|
# Desactiver la mise en veille auto du DSLR (sinon il se deconnecte tout seul)
|
|
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"
|
|
log.info("Camera DSLR connectee (LiveView actif)")
|
|
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:
|
|
# Désactiver LiveView : miroir abaissé → AF phase-détection Canon disponible
|
|
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
|
|
except Exception:
|
|
pass
|
|
log.info(f"Déclenchement capture DSLR (tentative {attempt+1}/3)")
|
|
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
|
|
self._chauffer_liveview() # Absorbe la stabilisation LiveView ici (spinner visible), pas sur le countdown suivant
|
|
from PIL import ImageOps as PILImageOps
|
|
img = PILImageOps.exif_transpose(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 _chauffer_liveview(self, tentatives: int = 8, pause: float = 0.15):
|
|
"""Consomme les premieres frames LiveView juste apres la remontee du miroir
|
|
(souvent en erreur ~1-2s le temps que l'AF/capteur se stabilise), pendant
|
|
que le spinner de traitement est affiche — pour que le preview du prochain
|
|
countdown soit deja a jour au lieu de rester figé sur l'ancienne photo."""
|
|
for _ in range(tentatives):
|
|
try:
|
|
with self._gp_lock:
|
|
self.camera.capture_preview()
|
|
return
|
|
except Exception:
|
|
time.sleep(pause)
|
|
|
|
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 configurer_flash(self, actif: bool):
|
|
"""Active ou désactive le flash intégré Canon.
|
|
Essaie plusieurs noms de widgets selon le modèle."""
|
|
if not self.camera or self.mode != "gphoto2":
|
|
return
|
|
# strobofiring : "0"=inhibé(off), "2"=auto — Canon EOS 60D (widget radio, valeurs string)
|
|
# flashmode : "On"/"Off" — autres Canon
|
|
candidats = [
|
|
("strobofiring", "2", "0"), # (widget, valeur_on, valeur_off)
|
|
("flashmode", "On", "Off"),
|
|
]
|
|
try:
|
|
with self._gp_lock:
|
|
cfg = self.camera.get_config()
|
|
for nom, val_on, val_off in candidats:
|
|
try:
|
|
widget = cfg.get_child_by_name(nom)
|
|
widget.set_value(val_on if actif else val_off)
|
|
self.camera.set_config(cfg)
|
|
log.info(f"Flash intégré ({nom}) : {'On' if actif else 'Off'}")
|
|
return
|
|
except Exception:
|
|
pass
|
|
log.debug("Aucun widget flash trouvé sur ce modèle Canon")
|
|
except Exception as e:
|
|
log.debug(f"configurer_flash : {e}")
|
|
|
|
def _desactiver_veille_init(self):
|
|
"""Desactive la mise en veille auto du boitier (Canon : autopoweroff en minutes, 0=jamais).
|
|
Sans ca, le DSLR s'eteint seul apres quelques minutes d'inactivite et apparait deconnecte."""
|
|
candidats = ["autopoweroff", "auto_power_off", "/main/settings/autopoweroff"]
|
|
try:
|
|
cfg = self.camera.get_config()
|
|
for nom in candidats:
|
|
try:
|
|
widget = cfg.get_child_by_name(nom)
|
|
widget.set_value(0)
|
|
self.camera.set_config(cfg)
|
|
log.info(f"Mise en veille DSLR désactivée ({nom})")
|
|
return
|
|
except Exception:
|
|
continue
|
|
log.debug("Widget autopoweroff non trouvé sur ce modèle")
|
|
except Exception as e:
|
|
log.debug(f"_desactiver_veille_init : {e}")
|
|
|
|
def _activer_viewfinder_init(self):
|
|
"""Active le LiveView pendant l'init (pas de lock, appelé avant que le thread démarre)."""
|
|
try:
|
|
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é (init)")
|
|
except Exception as e:
|
|
log.warning(f"Viewfinder non supporté : {e}")
|
|
|
|
def activer_viewfinder(self):
|
|
"""Active le LiveView (miroir levé) depuis le thread — thread-safe."""
|
|
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()
|