""" modules/planarian_metrics.py Intégration des métriques EthoVision XT dans PlanarianScanner. Architecture : PlanarianTracker.process() → dict brut (cx, cy, speed_px_s, ...) EthoVisionMetrics.update() → enrichit avec métriques EthoVision ReductStoreClient.store() → stocke dans ReductStore avec labels ReductStoreClient.export_csv() → exporte vers CSV Schéma des labels ReductStore : experiment : identifiant de l'expérience (ex: "exp_2026_04_25") well : identifiant du puits (ex: "A1", "B3") planarian : index du planaire dans le puits (ex: "0", "1") bucket : nom du bucket (ex: "planarian_metrics") Created on 25 avr. 2026 @author: denis """ import asyncio import csv import io import json import logging import math import os import time from datetime import datetime, timezone from typing import Optional logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Constantes EthoVision (seuils de mobilité par défaut) # --------------------------------------------------------------------------- # Seuils en mm/s — identiques à ceux de la simulation THRESH_IMMOBILE_DEFAULT = 0.2 # en-dessous : Immobile THRESH_MOBILE_DEFAULT = 1.5 # entre les deux : Mobile, au-delà : Highly mobile # États de mobilité (nomenclature EthoVision XT) STATE_IMMOBILE = "Immobile" STATE_MOBILE = "Mobile" STATE_HIGH_MOBILE = "Highly mobile" # Paramètres comportementaux (défauts — peuvent être importés depuis CSV/Django) BEHAVIOUR_DEFAULTS = { # Thigmotactisme "thigmotaxis_wall_dist_mm": 1.0, # distance à la paroi considérée "near wall" # Phototactisme "photo_mode": "none", # none | fixed | sine | radial "photo_strength": 0.0, # Chimiotactisme "chemo_strength": 0.0, "chemo_x": 0.5, # fraction 0-1 "chemo_y": 0.5, "chemo_radius_mm": 2.0, # Interactions inter-individus "avoid_radius_mm": 3.0, "aggreg_radius_mm": 6.0, } # --------------------------------------------------------------------------- # Classe EthoVisionMetrics # --------------------------------------------------------------------------- class EthoVisionMetrics: """ Calcule et accumule les métriques compatibles EthoVision XT à partir des données brutes de PlanarianTracker. Gère la conversion pixels → mm via le facteur px_per_mm. Une instance par planaire suivi (un puits = une instance). Usage : metrics = EthoVisionMetrics(px_per_mm=26.25, fps=10) for frame, ts in capture: annotated, raw = tracker.process(frame, ts) record = metrics.update(raw, well_radius_mm=8.0) await reduct_client.store(record, labels=...) summary = metrics.summary() """ def __init__( self, px_per_mm: float, fps: float, thresh_immobile: float = THRESH_IMMOBILE_DEFAULT, thresh_mobile: float = THRESH_MOBILE_DEFAULT, behaviour: Optional[dict] = None, ): """ Args: px_per_mm : facteur de conversion pixels → mm (calibration optique) fps : fréquence de capture en images/seconde thresh_immobile : seuil vitesse Immobile/Mobile en mm/s thresh_mobile : seuil vitesse Mobile/Très mobile en mm/s behaviour : dict de paramètres comportementaux (cf. BEHAVIOUR_DEFAULTS) """ self.px_per_mm = px_per_mm self.fps = fps self.dt = 1.0 / fps self.thresh_immobile = thresh_immobile self.thresh_mobile = thresh_mobile self.behaviour = {**BEHAVIOUR_DEFAULTS, **(behaviour or {})} # --- Accumulateurs globaux --- self.total_distance_mm = 0.0 self.duration_moving_s = 0.0 self.duration_stopped_s = 0.0 self.frame_count = 0 # --- Accumulateurs par état de mobilité --- self._mob_counts = { STATE_IMMOBILE: 0, STATE_MOBILE: 0, STATE_HIGH_MOBILE: 0, } self._mob_durations = { STATE_IMMOBILE: 0.0, STATE_MOBILE: 0.0, STATE_HIGH_MOBILE: 0.0, } self._current_state = None # --- Thigmotactisme --- self._near_wall_frames = 0 # --- Historique positions (pour calcul vitesse inter-frame) --- self._prev_cx_px = None self._prev_cy_px = None self._prev_ts = None def _px_to_mm(self, px: float) -> float: """Convertit des pixels en millimètres.""" return px / self.px_per_mm def _classify(self, velocity_mm_s: float) -> str: """ Classifie la vitesse selon les seuils EthoVision. Args: velocity_mm_s : vitesse instantanée en mm/s Returns: str : STATE_IMMOBILE | STATE_MOBILE | STATE_HIGH_MOBILE """ if velocity_mm_s <= self.thresh_immobile: return STATE_IMMOBILE elif velocity_mm_s <= self.thresh_mobile: return STATE_MOBILE return STATE_HIGH_MOBILE def update(self, raw: dict, well_radius_mm: float = 8.0) -> dict: """ Calcule les métriques EthoVision pour une frame à partir du résultat brut de PlanarianTracker.process(). Args: raw : dict retourné par PlanarianTracker.process() clés attendues : detected, cx, cy, speed_px_s, ts well_radius_mm : rayon du puits en mm (pour le thigmotactisme) Returns: dict complet avec métriques EthoVision prêtes pour ReductStore """ self.frame_count += 1 ts = raw.get("timestamp", time.time()) if not raw.get("detected", False): # Planaire non détecté : on accumule l'arrêt et on retourne vide self.duration_stopped_s += self.dt state = self._current_state or STATE_IMMOBILE self._mob_durations[state] += self.dt return self._empty_record(ts) cx_px = raw["cx"] cy_px = raw["cy"] # --- Conversion en mm --- cx_mm = self._px_to_mm(cx_px) cy_mm = self._px_to_mm(cy_px) # --- Vitesse en mm/s depuis la vitesse brute pixels/s --- speed_px_s = raw.get("speed_px_s", 0.0) velocity_mm_s = self._px_to_mm(speed_px_s) # --- Distance parcourue cette frame --- dist_mm = velocity_mm_s * self.dt self.total_distance_mm += dist_mm # --- Mouvement / arrêt --- is_moving = velocity_mm_s > self.thresh_immobile if is_moving: self.duration_moving_s += self.dt else: self.duration_stopped_s += self.dt # --- État de mobilité --- new_state = self._classify(velocity_mm_s) if new_state != self._current_state: self._mob_counts[new_state] += 1 self._current_state = new_state self._mob_durations[new_state] += self.dt # --- Thigmotactisme --- # Distance à la paroi du puits (centre = 0, paroi = well_radius_mm) well_radius_px = well_radius_mm * self.px_per_mm dist_center_px = math.sqrt(cx_px**2 + cy_px**2) dist_wall_mm = self._px_to_mm(well_radius_px - dist_center_px) near_wall_dist = self.behaviour.get("thigmotaxis_wall_dist_mm", 1.0) is_near_wall = dist_wall_mm < near_wall_dist if is_near_wall: self._near_wall_frames += 1 self._prev_cx_px = cx_px self._prev_cy_px = cy_px self._prev_ts = ts # --- Record complet --- return { # Identification temporelle "timestamp": ts, "detected": True, # Position brute (pixels) "cx_px": cx_px, "cy_px": cy_px, # Position en mm "x_mm": round(cx_mm, 4), "y_mm": round(cy_mm, 4), # Vitesse "velocity_mm_s": round(velocity_mm_s, 4), "distance_mm": round(dist_mm, 4), # Distance totale cumulée (EthoVision : movedCenter-pointTotalmm) "total_distance_mm": round(self.total_distance_mm, 4), # Mouvement / arrêt (EthoVision : MovementMoving / Not Moving) "moving": int(is_moving), "duration_moving_s": round(self.duration_moving_s, 3), "duration_stopped_s": round(self.duration_stopped_s, 3), # État de mobilité (EthoVision : Mobility state) "mobility_state": new_state, "mobility_immobile_freq": self._mob_counts[STATE_IMMOBILE], "mobility_immobile_duration_s": round(self._mob_durations[STATE_IMMOBILE], 3), "mobility_mobile_freq": self._mob_counts[STATE_MOBILE], "mobility_mobile_duration_s": round(self._mob_durations[STATE_MOBILE], 3), "mobility_high_mobile_freq": self._mob_counts[STATE_HIGH_MOBILE], "mobility_high_mobile_duration_s": round(self._mob_durations[STATE_HIGH_MOBILE], 3), # Thigmotactisme "dist_to_wall_mm": round(dist_wall_mm, 4), "near_wall": int(is_near_wall), # Données brutes tracker (passthrough) "area_px": raw.get("area_px", 0), "axial_pos": raw.get("axial_pos", 0.0), "axial_speed": raw.get("axial_speed", 0.0), } def summary(self) -> dict: """ Retourne le résumé global de la session (nomenclature EthoVision XT). À appeler en fin d'expérience pour stocker le résumé dans ReductStore. Returns: dict avec toutes les métriques agrégées """ total_s = self.frame_count * self.dt return { "total_frames": self.frame_count, "total_duration_s": round(total_s, 3), # Distance / vitesse (EthoVision : movedCenter-pointTotalmm / VelocityCenter-pointMeanmm/s) "movedCenter_pointTotal_mm": round(self.total_distance_mm, 4), "velocity_mean_mm_s": round( self.total_distance_mm / total_s if total_s > 0 else 0.0, 4 ), # Mouvement / arrêt "movement_moving_duration_s": round(self.duration_moving_s, 3), "movement_not_moving_duration_s": round(self.duration_stopped_s, 3), # Immobile "mobility_immobile_frequency": self._mob_counts[STATE_IMMOBILE], "mobility_immobile_duration_s": round(self._mob_durations[STATE_IMMOBILE], 3), # Mobile "mobility_mobile_frequency": self._mob_counts[STATE_MOBILE], "mobility_mobile_duration_s": round(self._mob_durations[STATE_MOBILE], 3), # Très mobile "mobility_highly_mobile_frequency": self._mob_counts[STATE_HIGH_MOBILE], "mobility_highly_mobile_duration_s": round(self._mob_durations[STATE_HIGH_MOBILE], 3), # Thigmotactisme "thigmotaxis_pct_time_near_wall": round( 100.0 * self._near_wall_frames / max(self.frame_count, 1), 2 ), } def reset(self): """ Réinitialise tous les accumulateurs. À appeler lors d'un changement de puits ou de planaire. """ self.__init__( self.px_per_mm, self.fps, self.thresh_immobile, self.thresh_mobile, self.behaviour, ) @staticmethod def _empty_record(ts: float) -> dict: """Retourne un enregistrement vide (planaire non détecté).""" return { "timestamp": ts, "detected": False, } # --------------------------------------------------------------------------- # Paramètres expérimentaux (importables depuis CSV ou Django) # --------------------------------------------------------------------------- class ExperimentParams: """ Conteneur des paramètres d'une expérience. Peut être instancié depuis un dict, un fichier CSV ou un modèle Django. Champs obligatoires : experiment, well, px_per_mm, fps Tous les autres ont des valeurs par défaut. """ REQUIRED = {"experiment", "well", "px_per_mm", "fps"} DEFAULTS = { "well_radius_mm": 8.0, "thresh_immobile": THRESH_IMMOBILE_DEFAULT, "thresh_mobile": THRESH_MOBILE_DEFAULT, "planarian_count": 1, "tube_axis": "vertical", "min_area_px": 20, **BEHAVIOUR_DEFAULTS, } def __init__(self, data: dict): """ Args: data : dict contenant au moins les champs REQUIRED """ missing = self.REQUIRED - set(data.keys()) if missing: raise ValueError(f"Paramètres manquants : {missing}") merged = {**self.DEFAULTS, **data} for k, v in merged.items(): # Conversion de type automatique si valeur string (vient du CSV) setattr(self, k, self._cast(k, v)) @staticmethod def _cast(key: str, value): """ Convertit la valeur en type approprié. Les valeurs CSV sont toutes des strings — on les cast automatiquement. Args: key : nom du paramètre value : valeur brute (str ou type natif) Returns: valeur convertie """ float_keys = { "px_per_mm", "fps", "well_radius_mm", "thresh_immobile", "thresh_mobile", "photo_strength", "chemo_strength", "chemo_x", "chemo_y", "chemo_radius_mm", "thigmotaxis_wall_dist_mm", "avoid_radius_mm", "aggreg_radius_mm", } int_keys = {"planarian_count", "min_area_px"} if key in float_keys: return float(value) if key in int_keys: return int(value) # Booléens CSV ("true"/"false") if isinstance(value, str) and value.lower() in ("true", "false"): return value.lower() == "true" return value @classmethod def from_csv_row(cls, row: dict) -> "ExperimentParams": """ Instancie depuis une ligne de DictReader CSV. Args: row : dict issu de csv.DictReader Returns: ExperimentParams """ return cls(row) @classmethod def from_csv_file(cls, filepath: str) -> list: """ Charge tous les paramètres d'un fichier CSV (une expérience par ligne). Args: filepath : chemin vers le fichier CSV Returns: liste d'ExperimentParams """ results = [] with open(filepath, newline="", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: try: results.append(cls.from_csv_row(row)) except ValueError as e: logger.warning(f"Ligne ignorée : {e} — {row}") return results def to_dict(self) -> dict: """Sérialise les paramètres en dict (pour stockage ou affichage Django).""" return {k: getattr(self, k) for k in {**self.DEFAULTS, **{r: None for r in self.REQUIRED}}} def build_metrics(self) -> "EthoVisionMetrics": """ Construit l'instance EthoVisionMetrics correspondant à ces paramètres. Returns: EthoVisionMetrics configurée """ behaviour = {k: getattr(self, k) for k in BEHAVIOUR_DEFAULTS if hasattr(self, k)} return EthoVisionMetrics( px_per_mm = self.px_per_mm, fps = self.fps, thresh_immobile = self.thresh_immobile, thresh_mobile = self.thresh_mobile, behaviour = behaviour, ) # --------------------------------------------------------------------------- # Client ReductStore # --------------------------------------------------------------------------- class ReductStoreClient: """ Interface asynchrone avec ReductStore pour PlanarianScanner. Schéma des labels : experiment → identifiant de l'expérience well → identifiant du puits (A1, B3, ...) planarian → index du planaire dans le puits record_type → "frame" | "summary" Chaque entrée stockée contient un payload JSON avec toutes les métriques. Le timestamp ReductStore est l'epoch µs de la frame. """ def __init__( self, url: str = "http://localhost:8383", token: str = "", bucket: str = "planarian_metrics", ): """ Args: url : URL du serveur ReductStore token : token d'authentification (vide si pas d'auth) bucket : nom du bucket cible """ self.url = url self.token = token self.bucket_name = bucket self._client = None self._bucket = None async def connect(self): """ Initialise la connexion et crée le bucket s'il n'existe pas. À appeler une fois au démarrage. """ from reduct import Client, BucketSettings, QuotaType self._client = Client(self.url, api_token=self.token) self._bucket = await self._client.create_bucket( self.bucket_name, BucketSettings(quota_type=QuotaType.NONE), exist_ok=True, ) logger.info(f"ReductStore connecté : {self.url} / bucket={self.bucket_name}") async def store_metric( self, record: dict, experiment: str, well: str, planarian: int = 0, record_type: str = "frame", ts_us: Optional[int] = None, ): """ Stocke un enregistrement de métriques dans ReductStore. Args: record : dict de métriques (issu de EthoVisionMetrics.update()) experiment : identifiant de l'expérience well : identifiant du puits planarian : index du planaire (défaut 0) record_type : "frame" ou "summary" ts_us : timestamp en microsecondes (défaut : maintenant) """ if self._bucket is None: await self.connect() ts_us = ts_us or int(time.time() * 1_000_000) labels = { "experiment": experiment, "well": well, "planarian": str(planarian), "record_type": record_type, } payload = json.dumps(record).encode("utf-8") await self._bucket.write( entry_name = "metrics", data = payload, timestamp = ts_us, labels = labels, content_type= "application/json", ) async def store_summary( self, summary: dict, experiment: str, well: str, planarian: int = 0, ): """ Stocke le résumé de fin de session dans ReductStore. Args: summary : dict issu de EthoVisionMetrics.summary() experiment : identifiant de l'expérience well : identifiant du puits planarian : index du planaire """ await self.store_metric( record = summary, experiment = experiment, well = well, planarian = planarian, record_type = "summary", ) async def get_tracking_data( self, experiment: str, well: str, planarian: int = 0, record_type: str = "frame", start: Optional[datetime] = None, stop: Optional[datetime] = None, ) -> list: """ Récupère les enregistrements depuis ReductStore avec filtrage par labels. Args: experiment : identifiant de l'expérience well : identifiant du puits planarian : index du planaire record_type : "frame" | "summary" start, stop : plage temporelle (datetime UTC, optionnel) Returns: liste de dicts métriques """ if self._bucket is None: await self.connect() labels = { "experiment": experiment, "well": well, "planarian": str(planarian), "record_type": record_type, } kwargs = {"include": labels} if start: kwargs["start"] = int(start.timestamp() * 1_000_000) if stop: kwargs["stop"] = int(stop.timestamp() * 1_000_000) records = [] async for record in self._bucket.query("metrics", **kwargs): try: data = json.loads(await record.read_all()) records.append(data) except Exception as e: logger.warning(f"Entrée illisible ignorée : {e}") return records async def export_csv( self, filepath: str, experiment: str, well: str, planarian: int = 0, record_type: str = "frame", start: Optional[datetime] = None, stop: Optional[datetime] = None, ) -> int: """ Exporte les données depuis ReductStore vers un fichier CSV. Args: filepath : chemin du fichier CSV de sortie experiment : identifiant de l'expérience well : identifiant du puits planarian : index du planaire record_type : "frame" | "summary" start, stop : plage temporelle (datetime UTC, optionnel) Returns: nombre de lignes exportées """ records = await self.get_tracking_data( experiment = experiment, well = well, planarian = planarian, record_type = record_type, start = start, stop = stop, ) if not records: logger.warning(f"Aucune donnée pour {experiment}/{well}/{planarian}") return 0 os.makedirs(os.path.dirname(os.path.abspath(filepath)), exist_ok=True) # Collecte de toutes les clés présentes (union de tous les records) fieldnames = list(dict.fromkeys(k for r in records for k in r.keys())) with open(filepath, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() for r in records: writer.writerow(r) logger.info(f"Export CSV : {len(records)} lignes → {filepath}") return len(records) async def export_csv_response( self, experiment: str, well: str, planarian: int = 0, record_type: str = "frame", start: Optional[datetime] = None, stop: Optional[datetime] = None, ) -> tuple[str, int]: """ Génère le contenu CSV en mémoire (pour une réponse HTTP Django). Args: experiment, well, planarian, record_type, start, stop : cf. export_csv Returns: tuple (contenu_csv_str, nb_lignes) """ records = await self.get_tracking_data( experiment = experiment, well = well, planarian = planarian, record_type = record_type, start = start, stop = stop, ) if not records: return "", 0 fieldnames = list(dict.fromkeys(k for r in records for k in r.keys())) output = io.StringIO() writer = csv.DictWriter(output, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() for r in records: writer.writerow(r) return output.getvalue(), len(records) async def close(self): """Ferme la connexion ReductStore.""" if self._client: await self._client.close() logger.info("ReductStore déconnecté")