From 1575445df4f78f466fe2447990dd246bac5ebaa3 Mon Sep 17 00:00:00 2001 From: denis defolie Date: Thu, 7 May 2026 11:13:51 +0200 Subject: [PATCH] tracking --- test_tube_scanner/.env.example | 1 + test_tube_scanner/home/settings.py | 2 + .../modules/planarian_metrics.py | 155 ++++++++++++++---- test_tube_scanner/planarian/views.py | 24 +-- test_tube_scanner/scanner/process.py | 6 +- .../scanner/templates/scanner/scanning.html | 4 +- test_tube_scanner/scanner/urls.py | 1 - test_tube_scanner/scanner/views.py | 12 -- 8 files changed, 141 insertions(+), 64 deletions(-) diff --git a/test_tube_scanner/.env.example b/test_tube_scanner/.env.example index c1be9c1..a9e77e1 100644 --- a/test_tube_scanner/.env.example +++ b/test_tube_scanner/.env.example @@ -19,6 +19,7 @@ APP_DATAS=test_tube_scanner EXPORTS_LOCAL_PATH=/home/rpi4/exports EXPORT_REMOTE_PATH=/mnt/exports REDUCTSTORE_PATH=/home/rpi4/medias +CSV_EXPORT_DIR=/home/rpi4/exports/csv #### # django email diff --git a/test_tube_scanner/home/settings.py b/test_tube_scanner/home/settings.py index 0bb2a92..86143ef 100644 --- a/test_tube_scanner/home/settings.py +++ b/test_tube_scanner/home/settings.py @@ -393,6 +393,8 @@ EXPORT_REMOTE_PATH = config("EXPORT_REMOTE_PATH") EXPORT_DESTINATIONS = ["local", "remote"] #EXPORT_DESTINATIONS = ["remote"] # only remote +CSV_EXPORT_DIR = config("CSV_EXPORT_DIR") # ou None pour mémoire uniquement + ## tracker default parameters # TRACKER_TUBE_AXIS = "vertical" diff --git a/test_tube_scanner/modules/planarian_metrics.py b/test_tube_scanner/modules/planarian_metrics.py index acee2ce..eaf9812 100644 --- a/test_tube_scanner/modules/planarian_metrics.py +++ b/test_tube_scanner/modules/planarian_metrics.py @@ -34,7 +34,7 @@ Métriques résumé (summary) : Created on 25 avr. 2026 @author: denis """ -import asyncio +#import asyncio import csv import io import json @@ -42,8 +42,8 @@ import logging import math import os import time +from datetime import datetime, timezone -from datetime import datetime from typing import Optional logger = logging.getLogger(__name__) @@ -667,14 +667,14 @@ class ReductStoreClient: self.bucket_name = bucket self.quota_type = quota_type self.quota_size = quota_size - self._client = None - self._bucket = asyncio.run(self.create_bucket()) + self.entry_name = "metrics" + self._client = None + self._bucket = None - async def create_bucket(self): + async def _create_bucket(self): from reduct import Client, BucketSettings self._client = Client(self.url, api_token=self.token) - settings = BucketSettings( quota_type=self.quota_type, quota_size=self.quota_size, @@ -682,43 +682,45 @@ class ReductStoreClient: ) return await self._client.create_bucket(self.bucket_name, settings, exist_ok=True) - ''' + async def connect(self): """Initialise la connexion et crée le bucket si nécessaire.""" - 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, - ) + self._bucket = await self._create_bucket() logger.info(f"ReductStore connecté : {self.url} / {self.bucket_name}") - ''' - async def store_metric( self, record: dict, experiment: str, well: str, - entry_name: str = "metrics", planarian: int = 0, - record_type: str = "metrics", + record_type: str = "frame", + uuid: str = "", ts_us: Optional[int] = None, ): - """Stocke un enregistrement dans ReductStore.""" + """ + Stocke un enregistrement dans ReductStore. + + Le timestamp est rendu unique par planaire en ajoutant l'index + du planaire comme offset sub-microseconde — évite le 409 Conflict + quand plusieurs planaires du même puits écrivent dans la même frame. + """ if self._bucket is None: await self.connect() - ts_us = ts_us or int(time.time() * 1_000_000) + # ts_us de base + offset planaire (0, 1, 2…) pour unicité garantie + base_ts = ts_us or int(time.time() * 1_000_000) + unique_ts = base_ts + planarian + await self._bucket.write( - entry_name = entry_name, + entry_name = "metrics", data = json.dumps(record).encode("utf-8"), - timestamp = ts_us, + timestamp = unique_ts, labels = { "experiment": experiment, "well": well, "planarian": str(planarian), "record_type": record_type, + "uuid": uuid, }, content_type = "application/json", ) @@ -733,7 +735,7 @@ class ReductStoreClient: experiment: str, well: str, planarian: int = 0, - record_type: str = "frame", + record_type: str = "metrics", start: Optional[datetime] = None, stop: Optional[datetime] = None, ) -> list: @@ -756,45 +758,121 @@ class ReductStoreClient: logger.warning(f"Entrée illisible ignorée : {e}") return records + @staticmethod + def _convert_timestamps(records: list) -> list: + """ + Convertit le champ 'timestamp' (epoch float secondes) en ISO 8601 UTC + dans chaque enregistrement. + + Args: + records : liste de dicts issus de ReductStore + + Returns: + nouvelle liste avec timestamp converti (originaux non modifiés) + """ + converted = [] + for r in records: + row = dict(r) + ts = row.get("timestamp") + if ts is not None: + try: + row["timestamp"] = ( + datetime.fromtimestamp(float(ts), tz=timezone.utc) + .strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z" + ) + except (ValueError, TypeError, OSError): + pass + converted.append(row) + return converted + + @staticmethod + def _build_filepath(output_dir: str, experiment: str, + well: str, planarian: int, record_type: str) -> str: + """ + Construit le chemin du fichier CSV de sortie. + Nom : __planaire_.csv + + Args: + output_dir : répertoire de sortie (créé si absent) + experiment : identifiant de l'expérience + well : identifiant du puits + planarian : index du planaire + record_type : "frame" ou "summary" + + Returns: + chemin absolu du fichier CSV + """ + dirpath = os.path.abspath(output_dir) + os.makedirs(dirpath, exist_ok=True) + filename = f"{experiment}_{well}_planaire{planarian:02d}_{record_type}.csv" + return os.path.join(dirpath, filename) + async def export_csv( self, - filepath: str, experiment: str, well: str, planarian: int = 0, - record_type: str = "frame", + record_type: str = "metrics", + output_dir: str = ".", start: Optional[datetime] = None, stop: Optional[datetime] = None, - ) -> int: - """Exporte les données vers un fichier CSV.""" + ) -> tuple: + """ + Exporte les données depuis ReductStore vers un fichier CSV. + Le répertoire de sortie est choisi via output_dir. + Le champ timestamp est converti en ISO 8601 UTC. + + Args: + experiment : identifiant de l'expérience + well : identifiant du puits + planarian : index du planaire + record_type : "frame" | "summary" + output_dir : répertoire de sortie (défaut : répertoire courant) + start, stop : plage temporelle (datetime UTC, optionnel) + + Returns: + tuple (filepath, nb_lignes) + """ records = await self.get_tracking_data( experiment, well, planarian, record_type, start, 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) + return "", 0 + + records = self._convert_timestamps(records) + filepath = self._build_filepath(output_dir, experiment, well, + planarian, record_type) 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() writer.writerows(records) + logger.info(f"Export CSV : {len(records)} lignes → {filepath}") - return len(records) + return filepath, len(records) async def export_csv_response( self, experiment: str, well: str, planarian: int = 0, - record_type: str = "frame", + record_type: str = "metrics", start: Optional[datetime] = None, stop: Optional[datetime] = None, ) -> tuple: - """Génère le contenu CSV en mémoire (pour réponse HTTP Django).""" + """ + Génère le contenu CSV en mémoire (pour réponse HTTP Django). + Le champ timestamp est converti en ISO 8601 UTC. + + Returns: + tuple (contenu_csv_str, nb_lignes) + """ records = await self.get_tracking_data( experiment, well, planarian, record_type, start, stop) if not records: return "", 0 + records = self._convert_timestamps(records) fieldnames = list(dict.fromkeys(k for r in records for k in r.keys())) out = io.StringIO() writer = csv.DictWriter(out, fieldnames=fieldnames, extrasaction="ignore") @@ -803,7 +881,12 @@ class ReductStoreClient: return out.getvalue(), len(records) async def close(self): - """Ferme la connexion ReductStore.""" - if self._client: - await self._client.close() - logger.info("ReductStore déconnecté") + """ + Ferme la connexion ReductStore. + Note : reduct-py >= 1.x ne nécessite pas de fermeture explicite — + la méthode est conservée pour compatibilité d'interface. + """ + self._client = None + self._bucket = None + logger.info("ReductStore déconnecté") + diff --git a/test_tube_scanner/planarian/views.py b/test_tube_scanner/planarian/views.py index 177f453..22160c8 100644 --- a/test_tube_scanner/planarian/views.py +++ b/test_tube_scanner/planarian/views.py @@ -19,6 +19,7 @@ from modules.planarian_metrics import ExperimentParams, ReductStoreClient from modules.system_stats import get_cached_stats, start_background_updater from scanner.constants import ScannerConstants + logger = logging.getLogger(__name__) start_background_updater() @@ -49,11 +50,7 @@ def global_context(request, **ctx): def _get_reduct_client() -> ReductStoreClient: """Instancie le client ReductStore depuis les settings Django.""" - return ReductStoreClient( - url = getattr(settings, "REDUCTSTORE_URL", "http://localhost:8383"), - token = getattr(settings, "REDUCTSTORE_TOKEN", ""), - bucket = getattr(settings, "REDUCTSTORE_BUCKET", "planarian_metrics"), - ) + return ReductStoreClient(url=settings.REDUCTSTORE_URL, token=settings.REDUCTSTORE_TOKEN) # --------------------------------------------------------------------------- @@ -198,7 +195,7 @@ class ExportCsvView(FormView): def form_valid(self, form): d = form.cleaned_data - + @async_to_sync async def _do_export(): client = _get_reduct_client() @@ -211,13 +208,18 @@ class ExportCsvView(FormView): record_type = d["record_type"], start = d.get("start_dt"), stop = d.get("stop_dt"), - ) - finally: - await client.close() + ) + print(f"Export CSV: export_csv_response done, {n} lignes, content size={len(csv_content)}, {csv_content}") + except Exception as e: + logger.error(f"Erreur export CSV: {e}") + messages.error(self.request, _("Erreur lors de l'export CSV: %(error)s") % {"error": str(e)}) + return None, 0 + return csv_content, n csv_content, n = _do_export() + print(f"Export CSV: {n} lignes, content size={len(csv_content)}") if not csv_content: messages.warning(self.request, _("Aucune donnée trouvée.")) return self.form_invalid(form) @@ -268,8 +270,8 @@ class TrackingDataView(View): planarian = planarian, record_type = record_type, ) - finally: - await client.close() + except Exception as e: + logger.error(f"Erreur fetching tracking data: {e}") records = _fetch() return JsonResponse({"count": len(records), "records": records}) diff --git a/test_tube_scanner/scanner/process.py b/test_tube_scanner/scanner/process.py index 0d92d02..a378b0b 100644 --- a/test_tube_scanner/scanner/process.py +++ b/test_tube_scanner/scanner/process.py @@ -43,6 +43,7 @@ logger = get_task_logger(__name__) redisDB = Redis(host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=0, decode_responses=True) cameraDB = reductstore.ReductStore(name='camera') planarianDB = planarian_metrics.ReductStoreClient(url=settings.REDUCTSTORE_URL, token=settings.REDUCTSTORE_TOKEN) +async_to_sync(planarianDB.connect)() class CameraRecordManager(): @@ -281,9 +282,10 @@ class ScannerProcess(Task): record, self.cam._params.experiment, self.cam._params.well, - entry_name=uuid, + uuid=uuid, planarian=pid, record_type='metrics', + ts_us=ts, ) def _store_frame(self, uuid, frame, ts, frame_count): @@ -311,7 +313,7 @@ class ScannerProcess(Task): try: (uuid, ts, frame, metrics, frame_count) = self.record_queue.get() if self.cam.use_tracking: - self._store_metric(uuid, metrics, ts) + self._store_metric(uuid, metrics, ts.timestamp()) self._store_frame(uuid, frame, ts, frame_count) diff --git a/test_tube_scanner/scanner/templates/scanner/scanning.html b/test_tube_scanner/scanner/templates/scanner/scanning.html index 5df6cb4..7d256ee 100644 --- a/test_tube_scanner/scanner/templates/scanner/scanning.html +++ b/test_tube_scanner/scanner/templates/scanner/scanning.html @@ -58,8 +58,8 @@
-
- +
+
diff --git a/test_tube_scanner/scanner/urls.py b/test_tube_scanner/scanner/urls.py index cd65b78..1bbfc6a 100644 --- a/test_tube_scanner/scanner/urls.py +++ b/test_tube_scanner/scanner/urls.py @@ -19,6 +19,5 @@ urlpatterns = [ path('replay/', views.replay_view, name='replay'), path('api/stats/', views.stats_view, name='api_stats'), path('api/video/', views.download_api, name='download_api'), - path('api/restart/', views.restart_all, name='restart_all'), path('api/export/', views.export_api, name='export_api'), ] diff --git a/test_tube_scanner/scanner/views.py b/test_tube_scanner/scanner/views.py index d18b55b..e173c15 100644 --- a/test_tube_scanner/scanner/views.py +++ b/test_tube_scanner/scanner/views.py @@ -156,18 +156,6 @@ def images_view(request): return render(request, "scanner/images.html", context=global_context(request, **ctx)) -## replay -@require_GET -@csrf_exempt -def restart_all(request): - #http://scanner.local:9001/index.html?processname=test_tube%3Aservices&action=restart - #http://scanner.local:9001/index.html?processname=test_tube%3AwebUI&action=start - #http://scanner.local:9001/index.html?action=restartall - #supervisor_restart_service('index.html?action=restartall') - supervisor_restart_service('index.html?processname=test_tube%3Aservices&action=restart') - return JsonResponse({"state": True}) - - @require_POST @csrf_exempt def download_api(request):