Jules
e9143f7a72
Sans ça le Canon reste en LiveView et utilise le contrast-detect AF (lent et imprécis). Le sleep(0.2) reste supprimé. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
441 lines
17 KiB
Python
441 lines
17 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:
|
|
cam = None
|
|
try:
|
|
cam = gp.Camera()
|
|
try:
|
|
cam.exit()
|
|
except Exception:
|
|
pass
|
|
cam.init()
|
|
self.camera = cam
|
|
self._desactiver_veille_init()
|
|
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 Exception as e:
|
|
log.warning(f"Pas de DSLR : {e}")
|
|
if cam is not None:
|
|
def _force_release(c):
|
|
try:
|
|
c.exit()
|
|
except Exception:
|
|
pass
|
|
t = threading.Thread(target=_force_release, args=(cam,), daemon=True)
|
|
t.start()
|
|
t.join(timeout=2.0)
|
|
del cam
|
|
self.camera = None
|
|
|
|
# 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, timeout: float = 3.0):
|
|
"""Deconnecte l'appareil photo avec timeout pour eviter le blocage PTP."""
|
|
cam_ref = self.camera
|
|
ctx_ref = self.contexte
|
|
if cam_ref and GPHOTO2_DISPONIBLE:
|
|
def _exit():
|
|
try:
|
|
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 — 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:
|
|
"""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:
|
|
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)
|
|
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)
|
|
threading.Thread(target=self._post_capture_warmup, daemon=True).start()
|
|
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}")
|
|
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 _post_capture_warmup(self):
|
|
self.activer_viewfinder()
|
|
self._chauffer_liveview()
|
|
|
|
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 keepalive(self) -> bool:
|
|
"""Envoie une commande legere au Canon pour empecher l'auto power off.
|
|
Utilise get_config qui declenche internement ptp_canon_eos_keepdeviceon."""
|
|
if self.mode != "gphoto2" or not self.connectee or not self.camera:
|
|
return False
|
|
try:
|
|
with self._gp_lock:
|
|
self.camera.get_config()
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
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)
|
|
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 Exception 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()
|