344 lines
13 KiB
Python
344 lines
13 KiB
Python
"""
|
||
Implémentation de la capture vidéo pour Raspberry Pi via PiCamera2.
|
||
Dépendance : picamera2 (sudo apt install python3-picamera2)
|
||
Compatible Pi 4 / Pi 5 avec le module caméra officiel (v2, v3, HQ).
|
||
"""
|
||
|
||
import io
|
||
import logging
|
||
from typing import Optional
|
||
|
||
from .capture_interface import CaptureError, VideoCaptureInterface
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class PiCamera2Capture(VideoCaptureInterface):
|
||
"""
|
||
Capture JPEG depuis la caméra officielle Raspberry Pi via PiCamera2.
|
||
|
||
Utilise le pipeline libcamera pour la capture basse latence.
|
||
Supporte les modules Camera Module v1, v2, v3 et HQ Camera.
|
||
|
||
La résolution demandée (width, height) est approchée au mieux :
|
||
libcamera sélectionne automatiquement le mode sensor dont la résolution
|
||
native est la plus proche, puis redimensionne en ISP.
|
||
Utiliser list_sensor_modes() pour connaître les modes disponibles.
|
||
|
||
Exemple d'utilisation ::
|
||
|
||
# Afficher les modes disponibles avant d'instancier
|
||
PiCamera2Capture.list_sensor_modes()
|
||
|
||
cam = PiCamera2Capture(fps=5, width=1280, height=720)
|
||
cam.set_frame_callback(lambda data, ts: print(f"{ts}: {len(data)} octets"))
|
||
cam.start()
|
||
time.sleep(10)
|
||
cam.stop()
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
fps: float = VideoCaptureInterface.DEFAULT_FPS,
|
||
width: int = 1280,
|
||
height: int = 720,
|
||
jpeg_quality: int = 85,
|
||
camera_index: int = 0,
|
||
use_video_config: bool = True,
|
||
use_tracking: bool = False,
|
||
display = None,
|
||
parent = None
|
||
|
||
):
|
||
"""
|
||
:param fps: Cadence cible en images par seconde
|
||
:param width: Largeur souhaitée en pixels (approchée au mode sensor le plus proche)
|
||
:param height: Hauteur souhaitée en pixels (approchée au mode sensor le plus proche)
|
||
:param jpeg_quality: Qualité de compression JPEG [0-100]
|
||
:param camera_index: Index de la caméra (0 par défaut, utile sur Pi 5 dual-cam)
|
||
:param use_video_config: True = VideoConfiguration (flux continu, basse latence)
|
||
False = StillConfiguration (haute résolution, plus lent)
|
||
"""
|
||
super().__init__(fps=fps, use_tracking=use_tracking, display=display, parent=parent, jpeg_quality=jpeg_quality)
|
||
self._width: int = width
|
||
self._height: int = height
|
||
self._jpeg_quality: int = jpeg_quality
|
||
self._camera_index: int = camera_index
|
||
self._use_video_config: bool = use_video_config
|
||
self._picam2 = None # Instance Picamera2
|
||
self._effective_size: Optional[tuple[int, int]] = None # Résolution réellement appliquée
|
||
|
||
# ------------------------------------------------------------------
|
||
# Méthode statique utilitaire — à appeler avant d'instancier
|
||
# ------------------------------------------------------------------
|
||
|
||
@staticmethod
|
||
def list_sensor_modes(camera_index: int = 0) -> list[dict]:
|
||
"""
|
||
Affiche et retourne tous les modes sensor disponibles pour la caméra.
|
||
|
||
À appeler avant d'instancier PiCamera2Capture pour choisir
|
||
une résolution compatible avec un mode sensor natif.
|
||
|
||
:param camera_index: Index de la caméra à interroger
|
||
:return: Liste de dicts décrivant chaque mode sensor
|
||
:raises CaptureError: Si picamera2 n'est pas disponible
|
||
"""
|
||
try:
|
||
from picamera2 import Picamera2
|
||
except ImportError as exc:
|
||
raise CaptureError(
|
||
"picamera2 introuvable — installez-le avec : "
|
||
"sudo apt install python3-picamera2"
|
||
) from exc
|
||
|
||
picam2 = Picamera2(camera_index)
|
||
try:
|
||
# sensor_modes doit être interrogé avant configure()
|
||
modes = picam2.sensor_modes
|
||
print(f"\n=== Modes sensor disponibles (caméra index={camera_index}) ===")
|
||
for i, mode in enumerate(modes):
|
||
size = mode.get("size", "?")
|
||
fps = mode.get("fps", "?")
|
||
crop = mode.get("crop_limits", "?")
|
||
fmt = mode.get("format", "?")
|
||
print(
|
||
f" [{i}] {size[0]}×{size[1]}px "
|
||
f"fps_max={fps:.1f} format={fmt} crop={crop}"
|
||
)
|
||
print()
|
||
return modes
|
||
finally:
|
||
picam2.close()
|
||
|
||
# ------------------------------------------------------------------
|
||
# Implémentation des méthodes abstraites
|
||
# ------------------------------------------------------------------
|
||
|
||
def open(self) -> None:
|
||
"""
|
||
Initialise PiCamera2, sélectionne le mode sensor le plus adapté
|
||
à la résolution demandée, configure le flux et démarre libcamera.
|
||
|
||
Le mode sensor est choisi en minimisant la distance euclidienne
|
||
entre la résolution native du mode et (width, height) demandés.
|
||
"""
|
||
try:
|
||
from picamera2 import Picamera2
|
||
except ImportError as exc:
|
||
raise CaptureError(
|
||
"picamera2 introuvable — installez-le avec : "
|
||
"sudo apt install python3-picamera2"
|
||
) from exc
|
||
|
||
try:
|
||
self._picam2 = Picamera2(self._camera_index)
|
||
|
||
# Sélection du mode sensor le plus proche de la résolution demandée
|
||
best_mode = self._select_best_sensor_mode(self._picam2)
|
||
|
||
# Construction de la configuration avec le mode sensor forcé
|
||
if self._use_video_config:
|
||
config = self._picam2.create_video_configuration(
|
||
main={"size": (self._width, self._height), "format": "RGB888"},
|
||
raw=best_mode, # Force le mode sensor natif
|
||
)
|
||
logger.debug("Configuration VideoConfiguration sélectionnée")
|
||
else:
|
||
config = self._picam2.create_still_configuration(
|
||
main={"size": (self._width, self._height), "format": "RGB888"},
|
||
raw=best_mode,
|
||
)
|
||
logger.debug("Configuration StillConfiguration sélectionnée")
|
||
|
||
self._picam2.configure(config)
|
||
self._picam2.start()
|
||
|
||
# Lecture de la résolution effectivement appliquée par l'ISP
|
||
actual = config["main"]["size"]
|
||
self._effective_size = actual
|
||
|
||
logger.info(
|
||
"PiCamera2 ouverte : index=%d demandé=%dx%d effectif=%dx%d "
|
||
"mode_sensor=%dx%d mode=%s",
|
||
self._camera_index,
|
||
self._width, self._height,
|
||
actual[0], actual[1],
|
||
best_mode["size"][0], best_mode["size"][1],
|
||
"video" if self._use_video_config else "still",
|
||
)
|
||
|
||
# Avertissement si la résolution effective diffère de la demande
|
||
if actual != (self._width, self._height):
|
||
logger.warning(
|
||
"Résolution ajustée par libcamera : %dx%d → %dx%d. "
|
||
"Utilisez list_sensor_modes() pour connaître les tailles compatibles.",
|
||
self._width, self._height, actual[0], actual[1],
|
||
)
|
||
|
||
except Exception as exc:
|
||
if self._picam2 is not None:
|
||
try:
|
||
self._picam2.close()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
self._picam2 = None
|
||
raise CaptureError(f"Impossible d'ouvrir PiCamera2 : {exc}") from exc
|
||
|
||
def close(self) -> None:
|
||
"""Arrête le pipeline libcamera et libère les ressources."""
|
||
if self._picam2 is not None:
|
||
try:
|
||
self._picam2.stop()
|
||
self._picam2.close()
|
||
logger.info("PiCamera2 fermée (index=%d)", self._camera_index)
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.warning("Erreur lors de la fermeture de PiCamera2 : %s", exc)
|
||
finally:
|
||
self._picam2 = None
|
||
self._effective_size = None
|
||
|
||
def capture_frame(self) -> bytes:
|
||
"""
|
||
Capture une image depuis le flux libcamera et l'encode en JPEG.
|
||
|
||
Stratégie : capture_array() → tableau NumPy RGB → encodage Pillow.
|
||
capture_file() ne supporte pas le paramètre quality ; on encode
|
||
manuellement pour contrôler le taux de compression.
|
||
|
||
:return: Données JPEG brutes
|
||
:raises CaptureError: Si la capture ou l'encodage échoue
|
||
"""
|
||
if self._picam2 is None:
|
||
raise CaptureError("PiCamera2 n'est pas initialisée")
|
||
|
||
try:
|
||
from PIL import Image
|
||
|
||
# Récupération du tableau RGB depuis le flux libcamera
|
||
arr = self._picam2.capture_array("main") # shape (H, W, 3) uint8
|
||
|
||
# Encodage manuel en JPEG avec la qualité configurée
|
||
buf = io.BytesIO()
|
||
Image.fromarray(arr).save(buf, format="JPEG", quality=self._jpeg_quality)
|
||
return buf.getvalue()
|
||
|
||
except Exception as exc:
|
||
raise CaptureError(f"Échec de capture PiCamera2 : {exc}") from exc
|
||
|
||
def is_available(self) -> bool:
|
||
"""Retourne True si le pipeline libcamera est démarré."""
|
||
return self._picam2 is not None
|
||
|
||
# ------------------------------------------------------------------
|
||
# Méthodes spécifiques à PiCamera2
|
||
# ------------------------------------------------------------------
|
||
|
||
def capture_high_res(self, width: int, height: int) -> bytes:
|
||
"""
|
||
Capture une image haute résolution hors flux principal (photo ponctuelle).
|
||
|
||
Utile pour déclencher une capture pleine résolution pendant un flux 5 fps.
|
||
Utilise capture_request() + make_array() + encodage Pillow.
|
||
|
||
:param width: Largeur souhaitée en pixels (indicatif, dépend de la config active)
|
||
:param height: Hauteur souhaitée en pixels
|
||
:return: Données JPEG brutes
|
||
:raises CaptureError: Si PiCamera2 n'est pas initialisée
|
||
"""
|
||
if self._picam2 is None:
|
||
raise CaptureError("PiCamera2 n'est pas initialisée")
|
||
|
||
try:
|
||
from PIL import Image
|
||
|
||
# Capture d'une requête unique depuis le flux actif
|
||
request = self._picam2.capture_request()
|
||
arr = request.make_array("main") # shape (H, W, 3) uint8 RGB
|
||
request.release()
|
||
|
||
buf = io.BytesIO()
|
||
Image.fromarray(arr).save(buf, format="JPEG", quality=self._jpeg_quality)
|
||
return buf.getvalue()
|
||
|
||
except Exception as exc:
|
||
raise CaptureError(f"Échec de capture haute résolution : {exc}") from exc
|
||
|
||
def set_controls(self, **kwargs) -> None:
|
||
"""
|
||
Applique des contrôles libcamera directement (exposition, gain, balance des blancs…).
|
||
|
||
Exemple ::
|
||
|
||
cam.set_controls(ExposureTime=10000, AnalogueGain=2.0)
|
||
|
||
:param kwargs: Contrôles libcamera valides pour le module connecté
|
||
"""
|
||
if self._picam2 is None:
|
||
raise CaptureError("PiCamera2 n'est pas initialisée")
|
||
self._picam2.set_controls(kwargs)
|
||
logger.debug("Contrôles appliqués : %s", kwargs)
|
||
|
||
def get_camera_properties(self) -> dict:
|
||
"""
|
||
Retourne les métadonnées du module caméra détecté.
|
||
|
||
:return: Dictionnaire des propriétés (modèle, résolution max, etc.)
|
||
"""
|
||
if self._picam2 is None:
|
||
raise CaptureError("PiCamera2 n'est pas initialisée")
|
||
return self._picam2.camera_properties
|
||
|
||
@property
|
||
def jpeg_quality(self) -> int:
|
||
"""Qualité JPEG [0-100]."""
|
||
return self._jpeg_quality
|
||
|
||
@jpeg_quality.setter
|
||
def jpeg_quality(self, value: int) -> None:
|
||
if not 0 <= value <= 100:
|
||
raise ValueError("La qualité JPEG doit être comprise entre 0 et 100")
|
||
self._jpeg_quality = value
|
||
|
||
@property
|
||
def resolution(self) -> tuple[int, int]:
|
||
"""Résolution de capture demandée (largeur, hauteur)."""
|
||
return (self._width, self._height)
|
||
|
||
@property
|
||
def effective_resolution(self) -> Optional[tuple[int, int]]:
|
||
"""
|
||
Résolution effectivement appliquée par l'ISP après ouverture.
|
||
None si la caméra n'est pas encore ouverte.
|
||
"""
|
||
return self._effective_size
|
||
|
||
# ------------------------------------------------------------------
|
||
# Helpers privés
|
||
# ------------------------------------------------------------------
|
||
|
||
def _select_best_sensor_mode(self, picam2) -> dict:
|
||
"""
|
||
Choisit le mode sensor dont la résolution native est la plus proche
|
||
de (width, height) en minimisant la distance euclidienne.
|
||
|
||
:param picam2: Instance Picamera2 déjà créée mais pas encore configurée
|
||
:return: Dict du mode sensor sélectionné
|
||
"""
|
||
modes = picam2.sensor_modes
|
||
if not modes:
|
||
raise CaptureError("Aucun mode sensor disponible")
|
||
|
||
def distance(mode: dict) -> float:
|
||
mw, mh = mode["size"]
|
||
# Distance euclidienne normalisée entre la résolution du mode et la cible
|
||
return ((mw - self._width) ** 2 + (mh - self._height) ** 2) ** 0.5
|
||
|
||
best = min(modes, key=distance)
|
||
logger.debug(
|
||
"Mode sensor sélectionné : %dx%d (demandé : %dx%d)",
|
||
best["size"][0], best["size"][1],
|
||
self._width, self._height,
|
||
)
|
||
return best
|