planarian tracker

This commit is contained in:
2026-04-16 19:45:47 +02:00
parent 56dbf4b4af
commit 42677121e3
11 changed files with 564 additions and 166 deletions
+43 -8
View File
@@ -22,6 +22,8 @@ from datetime import datetime, timezone
from pathlib import Path
from typing import Optional, Callable, TYPE_CHECKING
from modules.planarian_tracker import PlanarianTracker
if TYPE_CHECKING:
from .circular_crop import CircularCrop # Evite l'import circulaire au runtime
@@ -59,6 +61,20 @@ class VideoCaptureInterface(abc.ABC):
self._on_frame: Optional[Callable[[bytes, datetime], None]] = None # Callback image
self._circular_crop: Optional["CircularCrop"] = None # Recadrage circulaire optionnel
self._active_median = False
self._error_occured = False
self._tracker = PlanarianTracker(
tube_axis = "vertical", # à rendre configurable via settings
min_area_px = 20,
)
def on_well_change(self):
"""
Appelé par le CNC lors du changement de puits.
Réinitialise le fond appris et l'état inter-frame du tracker.
"""
self._tracker.reset()
# ------------------------------------------------------------------
# Méthodes abstraites — obligatoires dans les sous-classes
@@ -196,8 +212,25 @@ class VideoCaptureInterface(abc.ABC):
:return: Image traitée (JPEG ou PNG selon la stratégie)
"""
if self._circular_crop is not None:
return self._circular_crop.process(jpeg_bytes)
return jpeg_bytes
jpeg = self._circular_crop.process(jpeg_bytes)
# --- tracking ---
nparr = np.frombuffer(jpeg, np.uint8)
frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
ts = datetime.now(timezone.utc).timestamp()
#metrics = self._tracker.process(frame, ts) if frame is not None else {}
if frame is not None:
frame_annotated, metrics = self._tracker.process(frame, ts)
# Ré-encodage JPEG de la frame annotée
ok, buf = cv2.imencode(".jpg", frame_annotated, [cv2.IMWRITE_JPEG_QUALITY, 85])
if ok:
jpeg = buf.tobytes()
else:
metrics = {"detected": False}
return jpeg, metrics
return jpeg_bytes, {"detected": False}
def save_frame(self, jpeg_bytes: bytes, directory: str = ".", prefix: str = "frame") -> Path:
"""
@@ -250,9 +283,6 @@ class VideoCaptureInterface(abc.ABC):
cv2.line(frame, (0, center_y), (width, center_y), (0, 255, 0), 1)
cv2.circle(frame, (center_x, center_y), 2, (0, 0, 255), -1)
cv2.putText(frame, f"Num: {self._frame_count}", (10, 30),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
_, buffer = cv2.imencode('.jpg', frame)
jpeg_bytes = buffer.tobytes()
return jpeg_bytes
@@ -276,20 +306,25 @@ class VideoCaptureInterface(abc.ABC):
try:
jpeg = self.capture_frame()
jpeg = self.display_median(jpeg)
jpeg = self.process_frame(jpeg) # Recadrage circulaire si configuré
##
jpeg, metrics = self.process_frame(jpeg) # Recadrage circulaire si configuré
metrics.update({
"count": self._frame_count,
})
self._frame_count += 1
ts = datetime.now(timezone.utc)
if self._on_frame:
try:
self._on_frame(jpeg, ts)
self._on_frame(jpeg, ts, metrics)
except Exception as cb_err: # noqa: BLE001
logger.error("Erreur dans le callback image : %s", cb_err)
except CaptureError as err:
logger.error("Échec de capture (#%d) : %s", self._frame_count, err)
self._error_occured = True
# Compensation du temps d'exécution pour tenir la cadence
elapsed = time.monotonic() - loop_start
+10 -8
View File
@@ -234,7 +234,7 @@ class GRBLController:
class GridScanner:
def __init__(self, grbl, proc=None, **config):
def __init__(self, grbl, process=None, **config):
'''
xbase # Position X de départ (col 0) en mm
ybase # Position Y de départ (row 0) en mm
@@ -246,7 +246,7 @@ class GridScanner:
feed # Vitesse de déplacement entre éprouvettes (mm/min)
'''
self.grbl = grbl
self.proc = proc
self.process = process
self.position = config.get('position', 'HG')
self.xbase = config.get('xbase', 50)
@@ -266,7 +266,7 @@ class GridScanner:
def halt(self):
self.proc.record = False
self.process.tag.record = False
return self.stop_playing.set()
def _capture(self, uuid: str, duration: float, stop_running: Optional[threading.Event]) -> None:
@@ -274,8 +274,10 @@ class GridScanner:
Déclenche la caméra ArduCam et attend la fin de l'acquisition.
"""
print(f"# démarrer l'enregistrement {uuid}")
self.proc.uuid = uuid
self.proc.record = True
self.process.cam.on_well_change()
self.process.tag.uuid = uuid
self.process.tag.record = True
start = time.monotonic()
while not stop_running.is_set():
@@ -284,8 +286,8 @@ class GridScanner:
self.grbl.wait_for(1.0)
print("# arrêter l'enregistrement")
self.proc.record = False
self.proc.uuid = None
self.process.tag.record = False
self.process.tag.uuid = None
def start(self, xnext=None, ynext=None, position=None):
"""
@@ -347,7 +349,7 @@ class GridScanner:
self.grbl.move_to(x, y, feed=self.feed)
uuid = f'{self.proc.session}-{position}-{self.row_to_char[row]}{col+1}'
uuid = f'{self.process.tag.session}-{position}-{self.row_to_char[row]}{col+1}'
self._capture(uuid, self.duration, self.stop_playing)
# Retour à nexr après le scan
@@ -0,0 +1,242 @@
# modules/planarian_tracker.py
'''
Created on 16 avr. 2026
@author: denis
'''
import cv2
import logging
import numpy as np
logger = logging.getLogger(__name__)
class PlanarianTracker:
"""
Détection et suivi d'une planaire dans un tube.
Instancié une fois par caméra active, réutilisé frame à frame.
Utilise la soustraction de fond MOG2 — léger sur Raspberry Pi 4.
"""
def __init__(self, tube_axis: str = "vertical", min_area_px: int = 20):
# Axe du tube : "vertical" (cy) ou "horizontal" (cx)
self.tube_axis = tube_axis
self.min_area_px = min_area_px
# Etat inter-frame
self._prev_cx = None
self._prev_cy = None
self._prev_ts = None
# Soustracteur de fond adaptatif MOG2
self._bg_sub = cv2.createBackgroundSubtractorMOG2(
history = 50,
varThreshold = 25,
detectShadows= False,
)
def reset(self):
"""
Réinitialise l'état inter-frame — appeler lors du changement de puits.
"""
self._prev_cx = None
self._prev_cy = None
self._prev_ts = None
# Réinitialise le fond appris
self._bg_sub = cv2.createBackgroundSubtractorMOG2(
history = 50,
varThreshold = 25,
detectShadows= False,
)
'''
def process(self, frame: np.ndarray, ts: float) -> dict:
"""
Analyse une frame décodée numpy.
Retourne un dict de métriques attachable aux labels ReductStore.
:param frame: Frame BGR décodée (numpy array)
:param ts: Timestamp epoch secondes (float)
:return: dict métriques
"""
result = self._empty_result(ts)
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
fg_mask = self._bg_sub.apply(gray)
# Nettoyage morphologique du masque
kernel = np.ones((3, 3), np.uint8)
fg_mask = cv2.morphologyEx(fg_mask, cv2.MORPH_OPEN, kernel)
fg_mask = cv2.morphologyEx(fg_mask, cv2.MORPH_CLOSE, kernel)
contours, _ = cv2.findContours(
fg_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE
)
if not contours:
self._update_prev(None, None, ts)
return result
# Plus grand contour = planaire
largest = max(contours, key=cv2.contourArea)
area = cv2.contourArea(largest)
if area < self.min_area_px:
self._update_prev(None, None, ts)
return result
# Centre de masse
M = cv2.moments(largest)
if M["m00"] == 0:
return result
cx = int(M["m10"] / M["m00"])
cy = int(M["m01"] / M["m00"])
h, w = frame.shape[:2]
# Position normalisée sur l'axe du tube (0.0 → 1.0)
axial_pos = (cy / h) if self.tube_axis == "vertical" else (cx / w)
# Vitesse calculée entre frames
speed_px_s = None
axial_speed = None
if self._prev_cx is not None and self._prev_ts is not None:
dt = ts - self._prev_ts
if dt > 0:
dx = cx - self._prev_cx
dy = cy - self._prev_cy
speed_px_s = float(np.sqrt(dx**2 + dy**2) / dt)
# Vitesse signée sur l'axe du tube
# + = vers bas/droite, - = vers haut/gauche
axial_speed = float((dy / dt) if self.tube_axis == "vertical" else (dx / dt))
result.update({
"detected" : True,
"cx" : cx,
"cy" : cy,
"area_px" : int(area),
"speed_px_s" : round(speed_px_s, 3) if speed_px_s is not None else 0.0,
"axial_speed" : round(axial_speed, 3) if axial_speed is not None else 0.0,
"axial_pos" : round(axial_pos, 4),
})
self._update_prev(cx, cy, ts)
return result
'''
def process(self, frame: np.ndarray, ts: float) -> tuple[np.ndarray, dict]:
"""
Analyse une frame et dessine les contours détectés directement sur l'image.
Retourne (frame_annotée, métriques).
Contours fins Vert (0,255,0) Tous les contours valides détectés
Contour épais Cyan (255,255,0) Planaire principale (plus grand contour)
Croix + cercle Rouge (0,0,255) Centre de masse exact
Texte Blanc Vitesse px/s + position axiale normalisée
"""
result = self._empty_result(ts)
frame_out = frame.copy() # copie pour ne pas modifier l'original
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
fg_mask = self._bg_sub.apply(gray)
kernel = np.ones((3, 3), np.uint8)
fg_mask = cv2.morphologyEx(fg_mask, cv2.MORPH_OPEN, kernel)
fg_mask = cv2.morphologyEx(fg_mask, cv2.MORPH_CLOSE, kernel)
contours, _ = cv2.findContours(
fg_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE
)
if not contours:
self._update_prev(None, None, ts)
return frame_out, result
# Filtre les contours significatifs
valid_contours = [c for c in contours if cv2.contourArea(c) >= self.min_area_px]
if not valid_contours:
self._update_prev(None, None, ts)
return frame_out, result
# Dessine tous les contours valides en vert fin
cv2.drawContours(frame_out, valid_contours, -1, (0, 255, 0), 1)
# Plus grand contour = planaire principale
largest = max(valid_contours, key=cv2.contourArea)
area = cv2.contourArea(largest)
# Contour principal en cyan plus épais
cv2.drawContours(frame_out, [largest], -1, (255, 255, 0), 2)
M = cv2.moments(largest)
if M["m00"] == 0:
return frame_out, result
cx = int(M["m10"] / M["m00"])
cy = int(M["m01"] / M["m00"])
h, w = frame.shape[:2]
axial_pos = (cy / h) if self.tube_axis == "vertical" else (cx / w)
speed_px_s = None
axial_speed = None
if self._prev_cx is not None and self._prev_ts is not None:
dt = ts - self._prev_ts
if dt > 0:
dx = cx - self._prev_cx
dy = cy - self._prev_cy
speed_px_s = float(np.sqrt(dx**2 + dy**2) / dt)
axial_speed = float((dy / dt) if self.tube_axis == "vertical" else (dx / dt))
# Croix sur le centre de masse
cross_size = 8
cv2.line(frame_out, (cx - cross_size, cy), (cx + cross_size, cy), (0, 0, 255), 1)
cv2.line(frame_out, (cx, cy - cross_size), (cx, cy + cross_size), (0, 0, 255), 1)
# Cercle centré sur la planaire
cv2.circle(frame_out, (cx, cy), 12, (0, 0, 255), 1)
# Texte vitesse + position axiale
label = f"v={speed_px_s:.1f}px/s ax={axial_pos:.2f}" if speed_px_s is not None else f"ax={axial_pos:.2f}"
cv2.putText(
frame_out, label,
(max(cx - 60, 0), max(cy - 18, 12)),
cv2.FONT_HERSHEY_SIMPLEX, 0.4, (255, 255, 255), 1, cv2.LINE_AA,
)
result.update({
"detected" : True,
"cx" : cx,
"cy" : cy,
"area_px" : int(area),
"speed_px_s" : round(speed_px_s, 3) if speed_px_s is not None else 0.0,
"axial_speed" : round(axial_speed, 3) if axial_speed is not None else 0.0,
"axial_pos" : round(axial_pos, 4),
})
self._update_prev(cx, cy, ts)
return frame_out, result
# ------------------------------------------------------------------ #
def _empty_result(self, ts: float) -> dict:
return {
"timestamp" : ts,
"detected" : False,
"cx" : 0,
"cy" : 0,
"area_px" : 0,
"speed_px_s" : 0.0,
"axial_speed": 0.0,
"axial_pos" : 0.0,
}
def _update_prev(self, cx, cy, ts):
self._prev_cx = cx
self._prev_cy = cy
self._prev_ts = ts
@@ -0,0 +1,171 @@
"""
Implémentation de la capture vidéo à partir d'un fichier video via OpenCV (cv2).
Dépendance : opencv-python (pip install opencv-python)
OpenCV (cv2) avec import local pour éviter une dépendance globale
Résolution configurable, qualité JPEG réglable à chaud, accès V4L2 par index
get_resolution() pour lire la résolution effective appliquée par le pilote
"""
import os
os.environ['OPENCV_LOG_LEVEL']="0"
os.environ['OPENCV_FFMPEG_LOGLEVEL']="0"
import cv2
import logging
from typing import Optional
from .capture_interface import CaptureError, VideoCaptureInterface
logger = logging.getLogger(__name__)
class VideoFileCapture(VideoCaptureInterface):
"""
Capture JPEG depuis une webcam USB/intégrée via OpenCV.
Exemple d'utilisation ::
cam = VideoFileCapture(video_file=0, fps=5)
cam.set_frame_callback(lambda data, ts: print(f"{ts}: {len(data)} octets"))
cam.start()
time.sleep(10)
cam.stop()
"""
def __init__(
self,
video_file: str = None,
fps: float = VideoCaptureInterface.DEFAULT_FPS,
jpeg_quality: int = 85,
width: Optional[int] = None,
height: Optional[int] = None,
video_lists = []
):
"""
:param video_file: fichier video
:param fps: Cadence cible en images par seconde
:param jpeg_quality: Qualité de compression JPEG [0-100]
:param width: Largeur souhaitée (None = valeur par défaut du pilote)
:param height: Hauteur souhaitée (None = valeur par défaut du pilote)
"""
super().__init__(fps=fps)
self._video_file: str = video_file
self._jpeg_quality: int = jpeg_quality
self._width: Optional[int] = width
self._height: Optional[int] = height
self._video_lists = video_lists
self.ptf = 0
self._cap = None # Instance cv2.VideoCapture
def get_file(self):
if self._video_lists:
self._video_file = self._video_lists[self.ptf]
self.ptf += 1
if self.ptf >= len(self._video_lists):
self.ptf = 0
return self._video_file
# ------------------------------------------------------------------
# Implémentation des méthodes abstraites
# ------------------------------------------------------------------
def open(self) -> None:
"""Ouvre le flux V4L2 via OpenCV et configure la résolution."""
self.get_file()
self._cap = cv2.VideoCapture(self._video_file)
if not self._cap.isOpened():
raise CaptureError(
f"Impossible d'ouvrir le fichier (index={self._video_file})"
)
# Application de la résolution demandée
if self._width:
self._cap.set(cv2.CAP_PROP_FRAME_WIDTH, self._width)
if self._height:
self._cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self._height)
# Lecture de la résolution effectivement appliquée par le pilote
actual_w = int(self._cap.get(cv2.CAP_PROP_FRAME_WIDTH))
actual_h = int(self._cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
logger.info(
f"Fichier ouvert: index=%s résolution=%dx%d",
self._video_file, actual_w, actual_h,
)
def close(self) -> None:
"""Libère le flux OpenCV."""
if self._cap and self._cap.isOpened():
self._cap.release()
logger.info("Fichier fermé (index=%s)", self._video_file)
self._cap = None
def capture_frame(self) -> bytes:
"""
Lit une trame brute depuis OpenCV et l'encode en JPEG.
:return: Données JPEG brutes
:raises CaptureError: Si la lecture ou l'encodage échoue
"""
#import cv2
#import numpy as np # noqa: F401 — utilisé implicitement par cv2
if self._error_occured:
self.close()
self.open()
self._error_occured = False
if not self._cap or not self._cap.isOpened():
raise CaptureError("Le fichier n'est pas ouvert")
ret, frame = self._cap.read()
if not ret or frame is None:
raise CaptureError("Échec de lecture de la trame ou fin de fichier")
# Encodage BGR → JPEG avec la qualité configurée
encode_params = [cv2.IMWRITE_JPEG_QUALITY, self._jpeg_quality]
success, buffer = cv2.imencode(".jpg", frame, encode_params)
if not success:
raise CaptureError("Échec d'encodage JPEG")
return buffer.tobytes()
def is_available(self) -> bool:
"""Retourne True si le flux OpenCV est ouvert et prêt."""
return self._cap is not None and self._cap.isOpened()
# ------------------------------------------------------------------
# Accesseurs spécifiques à la webcam
# ------------------------------------------------------------------
@property
def video_file(self) -> int:
"""Index du périphérique V4L2."""
return self._video_file
@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
def get_resolution(self) -> Optional[tuple[int, int]]:
"""
Retourne la résolution effective du flux.
:return: Tuple (largeur, hauteur) ou None si la webcam est fermée
"""
if not self.is_available():
return None
w = int(self._cap.get(cv2.CAP_PROP_FRAME_WIDTH))
h = int(self._cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
return (w, h)