"""Boucle de feedback : collecte des corrections pompiste et conversion en données d'entraînement (voir docs/MONITORING.md). Principe : chaque photo traitée par `/analyze` est déjà sauvegardée dans `photos/.jpg` (voir `app/main.py`). Une correction envoyée via `POST /feedback` référence cette photo par son `photo_reference` et fournit les valeurs réellement correctes. `convert_feedback_to_annotations` transforme ensuite ces corrections en entrées au même format que `annotator/annotations/annotations.json`, réutilisables par le pipeline d'entraînement existant (`split_lines.py` -> `prepare_doctr_dataset.py` -> `finetune_doctr.py`). """ import json import os import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional from app.preprocessing import detect_screen_region MODULE_ROOT = Path(__file__).resolve().parent.parent # Même variable YELY_DATA_DIR que `app/main.py`/`monitoring/metrics.py` : si # un stockage persistant est monté (ex. /data sur un Hugging Face Space), # feedback.jsonl et les crops doivent survivre aux redémarrages du Space # comme les photos et les logs, sinon la boucle de feedback perd tout à # chaque redéploiement. `DEFAULT_ANNOTATIONS_PATH` reste dans le dépôt # (`annotator/`) volontairement : la conversion en données d'entraînement # est un script lancé en local par un développeur, pas sur le Space (voir # docs/MONITORING.md — l'annotateur n'est de toute façon pas poussé sur HF). DATA_DIR = Path(os.environ.get("YELY_DATA_DIR", str(MODULE_ROOT))) DEFAULT_FEEDBACK_PATH = DATA_DIR / "monitoring" / "feedback.jsonl" DEFAULT_PHOTOS_DIR = DATA_DIR / "photos" DEFAULT_ANNOTATIONS_PATH = MODULE_ROOT.parent / "annotator" / "annotations" / "annotations.json" DEFAULT_CROPS_DIR = DATA_DIR / "monitoring" / "feedback_crops" CORRECTABLE_FIELDS = ("prix", "volume", "prix_litre") def record_feedback(photo_reference: str, corrected_fields: Dict[str, Any], corrected_by: Optional[str] = None, feedback_path: Optional[Path] = None) -> Dict[str, Any]: """Ajoute une correction pompiste au journal `feedback.jsonl` (append-only). `corrected_fields` : sous-ensemble de {"prix", "volume", "prix_litre"} -> valeur correcte (les champs non fournis restent inconnus, pas déduits). """ unknown = set(corrected_fields) - set(CORRECTABLE_FIELDS) if unknown: raise ValueError(f"Champs de correction inconnus : {sorted(unknown)}") if not corrected_fields: raise ValueError("Aucune correction fournie.") feedback_path = feedback_path or DEFAULT_FEEDBACK_PATH feedback_path.parent.mkdir(parents=True, exist_ok=True) entry = { "feedback_id": str(uuid.uuid4()), "photo_reference": photo_reference, "corrected_fields": corrected_fields, "corrected_by": corrected_by, "created_at": datetime.now(timezone.utc).isoformat(), "converted": False, } with open(feedback_path, "a", encoding="utf-8") as f: f.write(json.dumps(entry, ensure_ascii=False) + "\n") return entry def load_feedback_entries(feedback_path: Optional[Path] = None) -> List[Dict[str, Any]]: feedback_path = feedback_path or DEFAULT_FEEDBACK_PATH if not feedback_path.exists(): return [] entries = [] with open(feedback_path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue try: entries.append(json.loads(line)) except json.JSONDecodeError: continue return entries def convert_feedback_to_annotations(photos_dir: Optional[Path] = None, feedback_path: Optional[Path] = None, annotations_path: Optional[Path] = None, crops_dir: Optional[Path] = None) -> int: """Convertit les corrections non encore traitées en entrées `annotations.json`. La détection d'écran (`detect_screen_region`) est automatique, donc pas fiable à 100% (voir docs/LIMITATIONS.md, point 2) : les entrées générées sont marquées `status="pending_review"` plutôt que `"annotated"`, pour qu'une relecture humaine (même rapide, via le visualiseur de l'annotateur) précède leur utilisation dans un ré-entraînement. Retourne le nombre d'entrées converties. """ import cv2 photos_dir = photos_dir or DEFAULT_PHOTOS_DIR feedback_path = feedback_path or DEFAULT_FEEDBACK_PATH annotations_path = annotations_path or DEFAULT_ANNOTATIONS_PATH crops_dir = crops_dir or DEFAULT_CROPS_DIR entries = load_feedback_entries(feedback_path) pending = [e for e in entries if not e.get("converted")] if not pending: return 0 annotations_path.parent.mkdir(parents=True, exist_ok=True) if annotations_path.exists(): with open(annotations_path, "r", encoding="utf-8") as f: annotations = json.load(f) else: annotations = {} crops_dir.mkdir(parents=True, exist_ok=True) converted_count = 0 for entry in pending: photo_path = photos_dir / entry["photo_reference"] if not photo_path.exists(): continue img = cv2.imread(str(photo_path)) if img is None: continue crop, bbox = detect_screen_region(img) crop_name = f"feedback_{entry['feedback_id']}_lcd.jpg" cv2.imwrite(str(crops_dir / crop_name), crop) fields = {name: "" for name in CORRECTABLE_FIELDS} fields.update({"unite_prix": "", "unite_vol": "", "notes": "issu du feedback pompiste"}) for name, value in entry["corrected_fields"].items(): fields[name] = str(value) annotations[entry["photo_reference"]] = { "status": "pending_review", "annotated_at": entry["created_at"], "image_path": str(photo_path), "lcd_bbox": list(bbox), "lcd_crop": str(crops_dir / crop_name), "fields": fields, } entry["converted"] = True converted_count += 1 with open(annotations_path, "w", encoding="utf-8") as f: json.dump(annotations, f, ensure_ascii=False, indent=2) with open(feedback_path, "w", encoding="utf-8") as f: for entry in entries: f.write(json.dumps(entry, ensure_ascii=False) + "\n") return converted_count if __name__ == "__main__": n = convert_feedback_to_annotations() print(f"{n} correction(s) converties en entrées à relire dans annotations.json")