From 308ddaa048d277acb5662295fdba03230541ead3 Mon Sep 17 00:00:00 2001 From: denis defolie Date: Tue, 19 May 2026 10:53:51 +0200 Subject: [PATCH] export-all-verif --- test_tube_scanner/.env.example | 1 + test_tube_scanner/backup/services.py | 61 +++++++++++ test_tube_scanner/backup/tasks.py | 18 ++++ test_tube_scanner/home/celerymodule.py | 3 + test_tube_scanner/home/settings.py | 3 + .../modules/planarian_metrics.py | 102 ++++++++++-------- test_tube_scanner/planarian/export_service.py | 68 ++++++++++++ test_tube_scanner/planarian/models.py | 3 +- test_tube_scanner/planarian/tasks.py | 96 +++++++++++------ .../planarian/templates/planarian/base.html | 7 ++ .../templates/planarian/export_csv.html | 1 + test_tube_scanner/planarian/views.py | 81 ++++++-------- test_tube_scanner/scanner/models.py | 16 ++- test_tube_scanner/scanner/views.py | 17 ++- 14 files changed, 345 insertions(+), 132 deletions(-) create mode 100644 test_tube_scanner/backup/services.py create mode 100644 test_tube_scanner/backup/tasks.py create mode 100644 test_tube_scanner/planarian/export_service.py diff --git a/test_tube_scanner/.env.example b/test_tube_scanner/.env.example index 5a9edb8..a212b9b 100644 --- a/test_tube_scanner/.env.example +++ b/test_tube_scanner/.env.example @@ -20,6 +20,7 @@ EXPORTS_LOCAL_PATH=/home/rpi4/exports EXPORT_REMOTE_PATH=/mnt/exports REDUCTSTORE_PATH=/home/rpi4/medias CSV_EXPORT_DIR=/home/rpi4/exports/csv +BACKUP_DIR=/home/rpi4/backup/mariadb #### # django email diff --git a/test_tube_scanner/backup/services.py b/test_tube_scanner/backup/services.py new file mode 100644 index 0000000..e219eae --- /dev/null +++ b/test_tube_scanner/backup/services.py @@ -0,0 +1,61 @@ +''' +Created on 18 mai 2026 + +@author: denis +''' +from pathlib import Path +from datetime import datetime +import subprocess +import gzip +import shutil + +from django.conf import settings + +BACKUP_DIR = Path(settings.BACKUP_DIR) + + +def mariadb_backup(): + BACKUP_DIR.mkdir(parents=True, exist_ok=True) + + db = settings.DATABASES["default"] + + db_name = db["NAME"] + db_user = db["USER"] + db_password = db["PASSWORD"] + db_host = db.get("HOST", "localhost") + db_port = str(db.get("PORT", 3306)) + + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + + sql_file = BACKUP_DIR / f"{db_name}_{timestamp}.sql" + gz_file = BACKUP_DIR / f"{db_name}_{timestamp}.sql.gz" + + cmd = [ + "mariadb-dump", + "--single-transaction", + "--quick", + "--routines", + "--triggers", + "-h", db_host, + "-P", db_port, + "-u", db_user, + f"-p{db_password}", + db_name, + ] + + with open(sql_file, "wb") as f: + result = subprocess.run( + cmd, + stdout=f, + stderr=subprocess.PIPE, + check=True, + ) + + with open(sql_file, "rb") as f_in: + with gzip.open(gz_file, "wb") as f_out: + shutil.copyfileobj(f_in, f_out) + + sql_file.unlink() + + return str(gz_file) + diff --git a/test_tube_scanner/backup/tasks.py b/test_tube_scanner/backup/tasks.py new file mode 100644 index 0000000..5d96064 --- /dev/null +++ b/test_tube_scanner/backup/tasks.py @@ -0,0 +1,18 @@ +''' +Created on 18 mai 2026 + +@author: denis +''' +from celery import shared_task + +from .services import mariadb_backup + + +@shared_task +def backup_mariadb_task(): + path = mariadb_backup() + + return { + "status": "ok", + "backup": path, + } \ No newline at end of file diff --git a/test_tube_scanner/home/celerymodule.py b/test_tube_scanner/home/celerymodule.py index 05eaf15..3482f1a 100644 --- a/test_tube_scanner/home/celerymodule.py +++ b/test_tube_scanner/home/celerymodule.py @@ -8,3 +8,6 @@ os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'home.settings') appc = Celery('home') appc.config_from_object('django.conf:settings', namespace='CELERY') appc.autodiscover_tasks() + +worker_prefetch_multiplier = 1 +task_acks_late = True diff --git a/test_tube_scanner/home/settings.py b/test_tube_scanner/home/settings.py index 7a39ad7..30bbfc5 100644 --- a/test_tube_scanner/home/settings.py +++ b/test_tube_scanner/home/settings.py @@ -345,6 +345,7 @@ DJANGO_CELERY_BEAT_TZ_AWARE = False CELERY_ENABLE_UTC = True CELERY_TASK_TRACK_STARTED = True CELERY_TASK_TIME_LIMIT = 30 * 60 +CELERY_TASK_SOFT_TIME_LIMIT = 20 * 60 CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler' CELERY_ACCEPT_CONTENT = ['application/json'] CELERY_RESULT_SERIALIZER = 'json' @@ -362,6 +363,8 @@ REDUCTSTORE_PORT = config('REDUCTSTORE_PORT', cast=int) REDUCTSTORE_URL = f'http://{REDUCTSTORE_HOST}:{REDUCTSTORE_PORT}' REDUCTSTORE_PATH = config("REDUCTSTORE_PATH") +BACKUP_DIR = config('BACKUP_DIR') + ## servers app # USER = config('USER') diff --git a/test_tube_scanner/modules/planarian_metrics.py b/test_tube_scanner/modules/planarian_metrics.py index 21e9fca..aadd745 100644 --- a/test_tube_scanner/modules/planarian_metrics.py +++ b/test_tube_scanner/modules/planarian_metrics.py @@ -687,16 +687,8 @@ class ReductStoreClient: logger.info(f"ReductStore connecté : {self.url} / {self.bucket_name}") ''' - async def store_metric( - self, - record: dict, - experiment: str, - well: str, - planarian: int = 0, - record_type: str = "frame", - uuid: str = "", - ts_us: Optional[int] = None, - ): + async def store_metric(self, record: dict, experiment: str, well: str, uuid: str, + planarian: int = 0, record_type: str = "frame", ts_us: Optional[int] = None, ): """ Stocke un enregistrement dans ReductStore. @@ -705,17 +697,18 @@ class ReductStoreClient: quand plusieurs planaires du même puits écrivent dans la même frame. Args: + uuid : identifiant unique de session (permet de filtrer + plusieurs sessions d'un même puits/expérience) 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" - uuid : identifiant unique de session (permet de filtrer - plusieurs sessions d'un même puits/expérience) ts_us : timestamp en microsecondes (défaut : maintenant) """ if self._bucket is None: await self.connect() + # 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 @@ -725,10 +718,13 @@ class ReductStoreClient: "planarian": str(planarian), "record_type": record_type, } + """ if uuid: - labels["uuid"] = uuid + labels["uuid"] = uuid""" + await self._bucket.write( - entry_name = "metrics", + #entry_name = "metrics", + entry_name = uuid, data = json.dumps(record).encode("utf-8"), timestamp = unique_ts, labels = labels, @@ -740,8 +736,9 @@ class ReductStoreClient: summary: dict, experiment: str, well: str, + uuid: str, planarian: int = 0, - uuid: str = "", + record_type: str = "summary", ): """ Stocke le résumé de fin de session dans ReductStore. @@ -750,26 +747,20 @@ class ReductStoreClient: summary : dict issu de EthoVisionMetrics.summary() experiment : identifiant de l'expérience well : identifiant du puits - planarian : index du planaire uuid : identifiant unique de session (même valeur que - celle utilisée dans store_metric pour cette session) + celle utilisée dans store_metric pour cette session) + planarian : index du planaire """ - await self.store_metric( - record = summary, - experiment = experiment, - well = well, - planarian = planarian, - record_type = "summary", - uuid = uuid, - ) + await self.store_metric(record=summary,experiment=experiment, + well=well, uuid=uuid, planarian=planarian, record_type=record_type) async def get_tracking_data( self, experiment: str, well: str, + uuid: str, planarian: int = 0, record_type: str = "frame", - uuid: str = "", start: Optional[datetime] = None, stop: Optional[datetime] = None, ) -> list: @@ -779,29 +770,31 @@ class ReductStoreClient: Args: experiment : identifiant de l'expérience well : identifiant du puits + uuid : filtre sur une session spécifique (optionnel — + si vide, retourne toutes les sessions) planarian : index du planaire record_type : "frame" | "summary" - uuid : filtre sur une session spécifique (optionnel — - si vide, retourne toutes les sessions) start, stop : plage temporelle (datetime UTC, optionnel) """ if self._bucket is None: await self.connect() - labels = { - "experiment": experiment, - "well": well, - "planarian": str(planarian), - "record_type": record_type, + + when = { + "$and": [ + #{"&experiment": {"$contains": experiment}}, + #{"&well": {"$contains": well}}, + {"&planarian": {"$contains": str(planarian)}}, + {"&record_type": {"$contains": record_type}}, + ] } - if uuid: - labels["uuid"] = uuid - kwargs = {"include": labels} + kwargs = {'when': when} if start: kwargs["start"] = int(start.timestamp() * 1_000_000) if stop: kwargs["stop"] = int(stop.timestamp() * 1_000_000) + records = [] - async for rec in self._bucket.query("metrics", **kwargs): + async for rec in self._bucket.query(uuid, **kwargs): try: records.append(json.loads(await rec.read_all())) except Exception as e: @@ -854,16 +847,16 @@ class ReductStoreClient: """ dirpath = os.path.abspath(output_dir) os.makedirs(dirpath, exist_ok=True) - filename = f"{experiment}_{well}_planaire{planarian:02d}_{record_type}.csv" + filename = f"{experiment}_{well}_planaire-{planarian}-{record_type}.csv" return os.path.join(dirpath, filename) async def export_csv( self, experiment: str, well: str, + uuid: str, planarian: int = 0, record_type: str = "frame", - uuid: str = "", output_dir: str = ".", start: Optional[datetime] = None, stop: Optional[datetime] = None, @@ -876,25 +869,34 @@ class ReductStoreClient: Args: experiment : identifiant de l'expérience well : identifiant du puits + uuid : filtre sur une session spécifique (optionnel — + si vide, retourne toutes les sessions) planarian : index du planaire record_type : "frame" | "summary" output_dir : répertoire de sortie (défaut : répertoire courant) - uuid : filtre sur une session spécifique (optionnel — - si vide, retourne toutes les sessions) start, stop : plage temporelle (datetime UTC, optionnel) Returns: tuple (filepath, nb_lignes) """ + + # record, experiment, well, uuid, records = await self.get_tracking_data( - experiment, well, planarian, record_type, uuid, start, stop) + experiment, + well, + uuid, + 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 records = self._convert_timestamps(records) - filepath = self._build_filepath(output_dir, experiment, well, - planarian, record_type) + 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: @@ -909,9 +911,9 @@ class ReductStoreClient: self, experiment: str, well: str, + uuid: str, planarian: int = 0, record_type: str = "frame", - uuid: str = "", start: Optional[datetime] = None, stop: Optional[datetime] = None, ) -> tuple: @@ -923,7 +925,15 @@ class ReductStoreClient: tuple (contenu_csv_str, nb_lignes) """ records = await self.get_tracking_data( - experiment, well, planarian, record_type, uuid, start, stop) + experiment, + well, + uuid, + planarian=planarian, + record_type=record_type, + start=start, + stop=stop, + ) + if not records: return "", 0 records = self._convert_timestamps(records) diff --git a/test_tube_scanner/planarian/export_service.py b/test_tube_scanner/planarian/export_service.py new file mode 100644 index 0000000..a79a639 --- /dev/null +++ b/test_tube_scanner/planarian/export_service.py @@ -0,0 +1,68 @@ + +import logging +import asyncio + +from asgiref.sync import async_to_sync +from django.conf import settings +from modules.planarian_metrics import ReductStoreClient + +logger = logging.getLogger(__name__) + + +def _get_reduct_client() -> ReductStoreClient: + """Instancie le client ReductStore depuis les settings Django.""" + return ReductStoreClient(url=settings.REDUCTSTORE_URL, token=settings.REDUCTSTORE_TOKEN) + + +def export_csv_sync( + *, + experiment, + well, + uuid, + planarian, + record_type, + start=None, + stop=None, +): + #@async_to_sync + async def _run(): + client = _get_reduct_client() + + await client.connect() + return await client.export_csv_response( + experiment=experiment, + well=well, + uuid=uuid, + planarian=planarian, + record_type=record_type, + start=start, + stop=stop, + ) + #return _run() + return asyncio.run(_run()) + +def export_csv_file_sync( + *, + experiment, + well, + uuid, + planarian, + record_type, +): + async def _run(): + client = _get_reduct_client() + + await client.connect() + try: + return await client.export_csv( + experiment=experiment, + well=well, + uuid=uuid, + planarian=planarian, + record_type=record_type, + output_dir=settings.CSV_EXPORT_DIR, + ) + finally: + await client.close() + return asyncio.run(_run()) + diff --git a/test_tube_scanner/planarian/models.py b/test_tube_scanner/planarian/models.py index 7fa58e1..1abacd9 100644 --- a/test_tube_scanner/planarian/models.py +++ b/test_tube_scanner/planarian/models.py @@ -143,7 +143,8 @@ class ExperimentConfig(models.Model): def get_session(self): - return self.experiment_key.session_experiments.first() if self.experiment_key else None + se = self.experiment_key.session_experiments.first() if self.experiment_key else None + return se.session def to_params_dict(self) -> dict: diff --git a/test_tube_scanner/planarian/tasks.py b/test_tube_scanner/planarian/tasks.py index 90edbf9..aa69eeb 100644 --- a/test_tube_scanner/planarian/tasks.py +++ b/test_tube_scanner/planarian/tasks.py @@ -1,54 +1,80 @@ # tasks.py -from celery import shared_task +from celery import shared_task, group from celery.utils.log import get_task_logger -from django.conf import settings -from asgiref.sync import async_to_sync -from modules.planarian_metrics import ReductStoreClient -from scanner.models import SessionExperiment +from scanner.models import SessionExperiment, get_uuid_from_session, Experiment from .models import ExperimentConfig - +from .export_service import export_csv_file_sync logger = get_task_logger(__name__) -def _get_reduct_client() -> ReductStoreClient: - """Instancie le client ReductStore depuis les settings Django.""" - return ReductStoreClient(url=settings.REDUCTSTORE_URL, token=settings.REDUCTSTORE_TOKEN) + +@shared_task(bind=True, autoretry_for=(Exception,), retry_backoff=True, retry_kwargs={"max_retries": 3},) +def export_single_metric(self, *, experiment, well, uuid, planarian, record_type,): + """ + Exporte UN fichier CSV. + """ + logger.info( f"Export start {experiment} {well} {planarian} {record_type}") + f, n = export_csv_file_sync( + experiment=experiment, + well=well, + uuid=uuid, + planarian=planarian, + record_type=record_type, + ) + logger.info( f"Export done {f} ({n} rows)" ) + return {"file": f, "rows": n, } @shared_task -def export_experiment_metrics(experiment): +def export_experiment_metrics_task(experiment_id): + experiment = Experiment.objects.filter(pk=experiment_id).first() + if not experiment: + return - @async_to_sync - async def _do_export(well, planarian, record_type): - client = _get_reduct_client() - - await client.connect() - try: - f, n = await client.export_csv( - experiment = experiment.identifier, - well = well, - planarian = planarian, - record_type = record_type, - output_dir = settings.CSV_EXPORT_DIR, - ) - logger.warning(f"Export CSV {settings.CSV_EXPORT_DIR}/{f} done, {n} lignes") - except Exception as e: - logger.error(f"Erreur export CSV: {e}") - - experiment_configs = ExperimentConfig.objects.filter(experiment_key_id=experiment.id).order_by('well').all() - for conf in experiment_configs: + jobs = [] + configs = (ExperimentConfig.objects.filter(experiment_key_id=experiment.id).order_by("well")) + for conf in configs: well = conf.well count = conf.planarian_count - + session = conf.get_session() + uuid = get_uuid_from_session(session.id, conf.experiment_key.multiwell.position, well, ) + for record_type in ["frame", "summary"]: for planarian in range(count): - _do_export(well, planarian, record_type) - + ''' + jobs.append( + export_single_metric.s( + experiment=experiment.identifier, + well=well, + uuid=uuid, + planarian=planarian, + record_type=record_type, + ) + ) + + ''' + try: + f, n = export_csv_file_sync( + experiment=experiment.identifier, + well=well, + uuid=uuid, + planarian=planarian, + record_type=record_type, + ) + logger.info( f"Export done {f} ({n} rows)" ) + except Exception as e: + logger.exception(e) + ''' + result = group(jobs).apply_async() + logger.info(f"{len(jobs)} exports launched group_id={result.id}") + return {"group_id": result.id,"jobs": len(jobs), }''' + + @shared_task -def export_session_metrics(session): - experiments = SessionExperiment.experiment_by_session(session.id) +def export_session_metrics_task(session_id): + experiments = SessionExperiment.experiment_by_session(session_id, active=False) for experiment in experiments: - export_experiment_metrics(experiment) + export_experiment_metrics_task.delay(experiment.id) \ No newline at end of file diff --git a/test_tube_scanner/planarian/templates/planarian/base.html b/test_tube_scanner/planarian/templates/planarian/base.html index 2f58afa..c793344 100644 --- a/test_tube_scanner/planarian/templates/planarian/base.html +++ b/test_tube_scanner/planarian/templates/planarian/base.html @@ -27,6 +27,12 @@ {% trans "Exporter les metrics de l'expérience" %}
 $> {{ export_csv_destination }} + + + {% trans "Exporter les metrics de la session" %}
+  $> {{ export_csv_destination }} + + {% endif %} {% endblock %} {% endblock %} @@ -34,6 +40,7 @@ {% block js_footer %} {{ block.super }}