227 lines
8.5 KiB
Python
227 lines
8.5 KiB
Python
"""
|
||
VideoPlateCapture — capture par extraction de région dans une vidéo plaque entière.
|
||
|
||
La vidéo montre l'ensemble de la plaque multi-puits (vue de dessus).
|
||
La position GRBL (x, y en mm) détermine la région extraite via px_per_mm.
|
||
Le résultat est un carré centré sur le puits courant, compatible avec
|
||
le recadrage circulaire de process_frame() comme pour toute autre capture.
|
||
|
||
Flux : video frame → crop carré (GRBL pos) → CircularCrop → tracking/display
|
||
|
||
Hot swap : set_video_file(path) remplace la vidéo sans arrêter la capture.
|
||
Thread-safe via _cap_lock.
|
||
"""
|
||
import os
|
||
os.environ['OPENCV_LOG_LEVEL'] = "0"
|
||
os.environ['OPENCV_FFMPEG_LOGLEVEL'] = "0"
|
||
import cv2
|
||
import numpy as np
|
||
import logging
|
||
import threading
|
||
from pathlib import Path
|
||
|
||
from modules.capture_interface import VideoCaptureInterface, CaptureError
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class VideoPlateCapture(VideoCaptureInterface):
|
||
"""
|
||
Lecture d'une vidéo de plaque complète avec crop dynamique à la position GRBL.
|
||
|
||
La position GRBL (x_mm, y_mm) est convertie en coordonnées pixel via px_per_mm.
|
||
Un carré de côté 2*crop_radius_px est extrait à cette position, puis
|
||
process_frame() applique le masque circulaire habituel.
|
||
|
||
La vidéo boucle automatiquement. La cadence est adaptée via frame_step
|
||
pour correspondre au fps cible.
|
||
|
||
Calibration : set_px_per_mm() met à jour le facteur de conversion à chaud.
|
||
Hot swap : set_video_file(path) change la vidéo sans interruption.
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
video_dir,
|
||
fps: float = 5.0,
|
||
jpeg_quality: int = 90,
|
||
use_tracking: bool = False,
|
||
display=None,
|
||
parent=None,
|
||
crop_radius_px: int = 150,
|
||
px_per_mm: float = 15.0,
|
||
x_offset_mm: float = 0.0,
|
||
y_offset_mm: float = 0.0,
|
||
initial_video_path: str | None = None,
|
||
):
|
||
super().__init__(fps, use_tracking, display, parent, jpeg_quality)
|
||
self._video_dir = Path(video_dir)
|
||
self._cap: cv2.VideoCapture | None = None
|
||
self._cap_lock = threading.Lock()
|
||
self._video_path: Path | None = (
|
||
Path(initial_video_path) if initial_video_path else None
|
||
)
|
||
self._frame_w: int = 0
|
||
self._frame_h: int = 0
|
||
self._crop_radius_px: int = crop_radius_px
|
||
self._px_per_mm: float = px_per_mm
|
||
self._x_offset_mm: float = x_offset_mm
|
||
self._y_offset_mm: float = y_offset_mm
|
||
self._frame_step: int = 1
|
||
|
||
# ------------------------------------------------------------------
|
||
# API publique
|
||
# ------------------------------------------------------------------
|
||
|
||
def set_px_per_mm(self, px_per_mm: float) -> None:
|
||
"""Met à jour le facteur de conversion mm → pixel à chaud (depuis WellPosition)."""
|
||
self._px_per_mm = px_per_mm
|
||
logger.info(f"VideoPlateCapture: px_per_mm={px_per_mm:.3f}")
|
||
|
||
def set_crop_radius_px(self, r: int) -> None:
|
||
"""Met à jour le rayon de découpe en pixels à chaud (depuis MultiWell.crop_radius)."""
|
||
self._crop_radius_px = r
|
||
logger.info(f"VideoPlateCapture: crop_radius_px={r}")
|
||
|
||
def set_offset_mm(self, x_offset_mm: float, y_offset_mm: float) -> None:
|
||
"""Met à jour l'offset d'origine (xbase, ybase du MultiWell) à chaud."""
|
||
self._x_offset_mm = x_offset_mm
|
||
self._y_offset_mm = y_offset_mm
|
||
|
||
def set_video_file(self, path: str) -> None:
|
||
"""
|
||
Hot swap : remplace la vidéo courante sans arrêter la capture.
|
||
Thread-safe — peut être appelé depuis n'importe quel thread.
|
||
"""
|
||
new_cap = cv2.VideoCapture(path)
|
||
if not new_cap.isOpened():
|
||
logger.error(f"VideoPlateCapture hot swap: impossible d'ouvrir {path}")
|
||
return
|
||
|
||
new_w = int(new_cap.get(cv2.CAP_PROP_FRAME_WIDTH))
|
||
new_h = int(new_cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
|
||
nat_fps = new_cap.get(cv2.CAP_PROP_FPS) or self._fps
|
||
new_step = max(1, round(nat_fps / self._fps))
|
||
|
||
with self._cap_lock:
|
||
old_cap = self._cap
|
||
self._cap = new_cap
|
||
self._video_path = Path(path)
|
||
self._frame_w = new_w
|
||
self._frame_h = new_h
|
||
self._frame_step = new_step
|
||
|
||
if old_cap:
|
||
old_cap.release()
|
||
|
||
logger.info(f"VideoPlateCapture: hot swap → {Path(path).name} {new_w}×{new_h}")
|
||
|
||
@property
|
||
def video_path(self) -> Path | None:
|
||
return self._video_path
|
||
|
||
# ------------------------------------------------------------------
|
||
# Implémentation VideoCaptureInterface
|
||
# ------------------------------------------------------------------
|
||
|
||
def open(self) -> None:
|
||
path = self._video_path if (self._video_path and self._video_path.exists()) \
|
||
else self._find_video()
|
||
if path is None:
|
||
raise CaptureError(f"Aucune vidéo trouvée dans {self._video_dir}")
|
||
|
||
cap = cv2.VideoCapture(str(path))
|
||
if not cap.isOpened():
|
||
raise CaptureError(f"Impossible d'ouvrir {path.name}")
|
||
|
||
nat_fps = cap.get(cv2.CAP_PROP_FPS) or self._fps
|
||
|
||
with self._cap_lock:
|
||
self._cap = cap
|
||
self._video_path = path
|
||
self._frame_w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
|
||
self._frame_h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
|
||
self._frame_step = max(1, round(nat_fps / self._fps))
|
||
|
||
logger.info(
|
||
f"VideoPlateCapture: {path.name} {self._frame_w}×{self._frame_h} "
|
||
f"@ {nat_fps:.1f} fps → step={self._frame_step}, "
|
||
f"px_per_mm={self._px_per_mm:.2f}"
|
||
)
|
||
|
||
def close(self) -> None:
|
||
with self._cap_lock:
|
||
cap, self._cap = self._cap, None
|
||
if cap:
|
||
cap.release()
|
||
|
||
def is_available(self) -> bool:
|
||
with self._cap_lock:
|
||
return self._cap is not None and self._cap.isOpened()
|
||
|
||
def capture_frame(self) -> bytes:
|
||
with self._cap_lock:
|
||
if self._cap is None or not self._cap.isOpened():
|
||
raise CaptureError("Vidéo non disponible")
|
||
|
||
# Avancer dans la vidéo pour correspondre au fps cible
|
||
for _ in range(self._frame_step - 1):
|
||
self._cap.grab()
|
||
|
||
ret, frame = self._cap.read()
|
||
if not ret:
|
||
self._cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
|
||
ret, frame = self._cap.read()
|
||
if not ret:
|
||
raise CaptureError("Impossible de relire la vidéo")
|
||
|
||
frame_w = self._frame_w
|
||
frame_h = self._frame_h
|
||
|
||
# Position GRBL (mm) → coordonnées pixel dans la vidéo plaque
|
||
grbl = getattr(self.parent, 'grbl', None) if self.parent else None
|
||
x_mm = float(getattr(grbl, 'x', None) or 0.0)
|
||
y_mm = float(getattr(grbl, 'y', None) or 0.0)
|
||
|
||
# À l'origine CNC (0, 0) : retourner la plaque entière à sa résolution native
|
||
if x_mm == 0.0 and y_mm == 0.0:
|
||
ok, buf = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, self.jpeg_quality])
|
||
if not ok:
|
||
raise CaptureError("Encodage JPEG échoué")
|
||
return buf.tobytes()
|
||
|
||
cx = int((x_mm - self._x_offset_mm) * self._px_per_mm)
|
||
cy = int((y_mm - self._y_offset_mm) * self._px_per_mm)
|
||
r = self._crop_radius_px
|
||
|
||
# Extraction du carré centré sur le puits courant
|
||
x1 = max(0, cx - r)
|
||
y1 = max(0, cy - r)
|
||
x2 = min(frame_w, cx + r)
|
||
y2 = min(frame_h, cy + r)
|
||
crop = frame[y1:y2, x1:x2]
|
||
|
||
# Padding noir si le crop déborde du bord de la vidéo
|
||
if crop.shape[0] != 2 * r or crop.shape[1] != 2 * r:
|
||
padded = np.zeros((2 * r, 2 * r, 3), dtype=np.uint8)
|
||
padded[:crop.shape[0], :crop.shape[1]] = crop
|
||
crop = padded
|
||
|
||
ok, buf = cv2.imencode('.jpg', crop, [cv2.IMWRITE_JPEG_QUALITY, self.jpeg_quality])
|
||
if not ok:
|
||
raise CaptureError("Encodage JPEG échoué")
|
||
return buf.tobytes()
|
||
|
||
# ------------------------------------------------------------------
|
||
# Privé
|
||
# ------------------------------------------------------------------
|
||
|
||
def _find_video(self) -> Path | None:
|
||
"""Retourne le premier fichier vidéo trouvé dans video_dir."""
|
||
self._video_dir.mkdir(parents=True, exist_ok=True)
|
||
for ext in ('*.mp4', '*.avi', '*.MP4', '*.AVI'):
|
||
files = sorted(self._video_dir.glob(ext))
|
||
if files:
|
||
return files[0]
|
||
return None
|