"""Registre typé des hooks de métriques document-level et corpus-level. Chantier 2 du plan d'évolution post-Sprint 97. Pourquoi ce module ------------------ Avant ce chantier, ``picarones.core.runner._compute_document_result`` contenait **11 imports tardifs codés en dur** vers ``picarones.core.confusion``, ``char_scores``, ``taxonomy``, ``structure``, ``image_quality``, ``line_metrics``, ``hallucination``, ``philological_runner``, ``searchability_runner``, ``numerical_sequences_runner``, ``readability_runner`` — chacun enrobé dans un ``try/except Exception`` qui logue un warning. Symétriquement, la phase d'agrégation contenait 11 fonctions ``_aggregate_*`` ou ``aggregate_*``. Ajouter une nouvelle métrique exigeait de patcher ``runner.py`` à deux endroits, ce qui rendait le fichier monolithique (1322 lignes) et fragile. Ce module centralise le mécanisme : - **Profils** (``minimal`` / ``standard`` / ``philological`` / ``diagnostics`` / ``pipeline`` / ``full``) — l'utilisateur choisit quel sous-ensemble de métriques calculer selon son use case. - **Hooks document-level** (:class:`DocumentMetricHook`) enregistrés via :func:`register_document_metric` — fonctions appelées pour chaque document, leur retour remplit un attribut nommé du ``DocumentResult``. - **Agrégateurs corpus-level** (:class:`CorpusMetricAggregator`) enregistrés via :func:`register_corpus_aggregator` — fonctions appelées une fois par moteur pour synthétiser les ``DocumentResult`` en attributs ``aggregated_*`` du ``EngineReport``. Rétrocompat stricte ------------------- Le profil ``standard`` (défaut) active exactement les 11 hooks et 11 agrégateurs historiques. Comportement, ordre d'exécution, gestion d'erreurs et octets de sortie : strictement identiques à avant le chantier 2. La preuve est dans ``tests/test_metric_hooks.py`` (cas-tests qui comparent profil ``standard`` vs comportement legacy sur fixtures). Comment ajouter un hook ----------------------- Pour ajouter une métrique document-level : >>> from picarones.core.metric_hooks import ( ... register_document_metric, PROFILE_STANDARD, PROFILE_FULL, ... ) >>> >>> @register_document_metric( ... name="my_metric", ... attribute="my_metric", # nom du champ dans DocumentResult ... profiles=(PROFILE_STANDARD, PROFILE_FULL), ... requires_success=True, ... ) ... def my_hook(*, ground_truth, hypothesis, image_path, corpus_lang, ... ocr_result): ... # Imports tardifs OK ici — le coût n'est payé que si le hook ... # est dans le profil actif. ... from my_pkg import compute_my_metric ... return compute_my_metric(ground_truth, hypothesis) Pour un nouveau profil, l'ajouter à :data:`KNOWN_PROFILES` (et référencer dans la doc utilisateur ``docs/profiles/``). """ from __future__ import annotations import logging from dataclasses import dataclass, field from typing import Any, Callable, Iterable, Optional logger = logging.getLogger(__name__) # ────────────────────────────────────────────────────────────────────────── # Profils # ────────────────────────────────────────────────────────────────────────── PROFILE_MINIMAL = "minimal" """Profil le plus léger : juste CER/WER de ``compute_metrics``. Aucun hook document-level ni agrégateur ne s'exécute. Cible : tests rapides ou bench massif où seul le CER global compte.""" PROFILE_STANDARD = "standard" """Profil par défaut. Active tous les hooks et agrégateurs historiques de Picarones (Sprints 5 + 10 + 39+42 + 55-60 + 84-87). Comportement strictement identique au runner pré-chantier-2.""" PROFILE_PHILOLOGICAL = "philological" """Profil orienté édition critique : standard + emphase philologique. Aujourd'hui équivalent à standard ; réservé pour des hooks futurs spécifiques aux corpus médiévaux et imprimés anciens.""" PROFILE_DIAGNOSTICS = "diagnostics" """Profil orienté diagnostic : standard + leviers d'amélioration, prédiction de complexité, baseline historique. Réservé pour des hooks futurs (chantiers 3-4).""" PROFILE_ECONOMICS = "economics" """Profil orienté décision budget : minimal + métriques économiques (throughput effectif, coût marginal). Réservé pour des hooks futurs.""" PROFILE_PIPELINE = "pipeline" """Profil pour les benchmarks de pipelines composées (axe B). Active les hooks pertinents aux jonctions du DAG. Réservé pour des hooks futurs spécifiques aux pipelines.""" PROFILE_FULL = "full" """Profil exhaustif : tous les hooks de tous les profils. Coût maximal mais reproductibilité scientifique maximale.""" KNOWN_PROFILES: frozenset[str] = frozenset({ PROFILE_MINIMAL, PROFILE_STANDARD, PROFILE_PHILOLOGICAL, PROFILE_DIAGNOSTICS, PROFILE_ECONOMICS, PROFILE_PIPELINE, PROFILE_FULL, }) def validate_profile(profile: str) -> None: """Lève ``ValueError`` si ``profile`` n'est pas connu. Le runner appelle cette fonction au démarrage pour rejeter rapidement une faute de frappe utilisateur (``--profile philolagic``). """ if profile not in KNOWN_PROFILES: raise ValueError( f"profil inconnu : {profile!r}. " f"Profils valides : {sorted(KNOWN_PROFILES)}" ) # ────────────────────────────────────────────────────────────────────────── # Modèles de hook # ────────────────────────────────────────────────────────────────────────── @dataclass(frozen=True) class DocumentMetricHook: """Hook calculé pour chaque document. Attributs --------- name: Identifiant lisible utilisé dans les logs et les warnings. attribute: Nom du champ du :class:`DocumentResult` à remplir (par exemple ``"confusion_matrix"`` ou ``"taxonomy"``). Doit correspondre exactement à un attribut existant — le runner passe le résultat via ``setattr``. profiles: Ensemble des profils dans lesquels ce hook s'active. func: Fonction calculant la métrique. Signature attendue : ``func(*, ground_truth, hypothesis, image_path, corpus_lang, ocr_result) -> Any``. Tous les arguments sont passés en keyword pour que les hooks puissent ignorer ceux qu'ils n'utilisent pas avec ``**_``. requires_success: Si ``True``, le hook n'est appelé que quand ``ocr_result.success`` (texte hyp non-vide). Évite de gaspiller du temps sur des documents en erreur OCR. requires_token_confidences: Si ``True``, le hook n'est appelé que quand ``ocr_result.token_confidences`` est non-vide. Réservé à la calibration (Sprint 42). """ name: str attribute: str profiles: frozenset[str] func: Callable[..., Any] requires_success: bool = False requires_token_confidences: bool = False @dataclass(frozen=True) class CorpusMetricAggregator: """Agrégateur calculé une fois par moteur sur tous les documents. Attributs --------- name: Identifiant lisible. attribute: Nom du champ du :class:`EngineReport` à remplir (par exemple ``"aggregated_confusion"``). profiles: Profils dans lesquels l'agrégateur s'active. func: ``func(document_results: list[DocumentResult]) -> Any``. """ name: str attribute: str profiles: frozenset[str] func: Callable[..., Any] # ────────────────────────────────────────────────────────────────────────── # Registres globaux # ────────────────────────────────────────────────────────────────────────── _DOCUMENT_HOOKS: list[DocumentMetricHook] = [] _CORPUS_AGGREGATORS: list[CorpusMetricAggregator] = [] def _check_profiles(profiles: Iterable[str]) -> frozenset[str]: frozen = frozenset(profiles) unknown = frozen - KNOWN_PROFILES if unknown: raise ValueError( f"profils inconnus : {sorted(unknown)}. " f"Profils valides : {sorted(KNOWN_PROFILES)}" ) return frozen def register_document_metric( *, name: str, attribute: str, profiles: Iterable[str], requires_success: bool = False, requires_token_confidences: bool = False, ) -> Callable[[Callable[..., Any]], Callable[..., Any]]: """Décorateur d'enregistrement d'un hook document-level.""" profiles_set = _check_profiles(profiles) def decorator(func: Callable[..., Any]) -> Callable[..., Any]: # Idempotence : si le même func est ré-enregistré (ré-import # du module en test), on ignore silencieusement. Si un autre # func tente le même ``name``, on lève. for existing in _DOCUMENT_HOOKS: if existing.name == name: if existing.func is func: return func raise ValueError( f"hook document '{name}' déjà enregistré par " f"{existing.func.__module__}.{existing.func.__qualname__}" ) _DOCUMENT_HOOKS.append(DocumentMetricHook( name=name, attribute=attribute, profiles=profiles_set, func=func, requires_success=requires_success, requires_token_confidences=requires_token_confidences, )) return func return decorator def register_corpus_aggregator( *, name: str, attribute: str, profiles: Iterable[str], ) -> Callable[[Callable[..., Any]], Callable[..., Any]]: """Décorateur d'enregistrement d'un agrégateur corpus-level.""" profiles_set = _check_profiles(profiles) def decorator(func: Callable[..., Any]) -> Callable[..., Any]: for existing in _CORPUS_AGGREGATORS: if existing.name == name: if existing.func is func: return func raise ValueError( f"agrégateur corpus '{name}' déjà enregistré par " f"{existing.func.__module__}.{existing.func.__qualname__}" ) _CORPUS_AGGREGATORS.append(CorpusMetricAggregator( name=name, attribute=attribute, profiles=profiles_set, func=func, )) return func return decorator # ────────────────────────────────────────────────────────────────────────── # Sélection + exécution selon profil # ────────────────────────────────────────────────────────────────────────── def select_document_hooks(profile: str) -> list[DocumentMetricHook]: """Retourne les hooks document-level actifs pour ``profile``. L'ordre d'enregistrement est préservé pour garantir que les warnings et logs apparaissent dans le même ordre qu'avant le chantier 2 (cf. test de rétrocompat). """ validate_profile(profile) return [h for h in _DOCUMENT_HOOKS if profile in h.profiles] def select_corpus_aggregators(profile: str) -> list[CorpusMetricAggregator]: """Retourne les agrégateurs corpus-level actifs pour ``profile``.""" validate_profile(profile) return [a for a in _CORPUS_AGGREGATORS if profile in a.profiles] def run_document_hooks( profile: str, *, ground_truth: str, hypothesis: str, image_path: str, corpus_lang: str, ocr_result: Any, ) -> dict[str, Any]: """Exécute tous les hooks document-level actifs pour ``profile``. Retourne un dict ``{attribute_name: value}`` que le runner peut appliquer au ``DocumentResult`` via ``setattr`` ou ``**kwargs``. Pré-conditions : - les hooks à ``requires_success=True`` ne tournent que si ``ocr_result.success`` ; - les hooks à ``requires_token_confidences=True`` ne tournent que si ``ocr_result.token_confidences`` est non vide. Toute exception levée par un hook est loggée en warning et le hook est sauté (``attribute`` absent du dict retourné). Aucun hook ne fait jamais échouer le calcul des autres — discipline historique préservée. """ out: dict[str, Any] = {} for hook in select_document_hooks(profile): if hook.requires_success and not getattr(ocr_result, "success", False): continue if hook.requires_token_confidences and not getattr( ocr_result, "token_confidences", None, ): continue try: value = hook.func( ground_truth=ground_truth, hypothesis=hypothesis, image_path=image_path, corpus_lang=corpus_lang, ocr_result=ocr_result, ) except Exception as exc: # noqa: BLE001 logger.warning( "[%s] fonctionnalité dégradée : %s", hook.name, exc, ) continue if value is not None: out[hook.attribute] = value return out def run_corpus_aggregators( profile: str, document_results: list, ) -> dict[str, Any]: """Exécute tous les agrégateurs corpus-level pour ``profile``. Retourne un dict ``{attribute_name: value}`` à appliquer au ``EngineReport``. Comme pour les hooks doc-level, une exception dans un agrégateur est loggée et l'agrégateur sauté. """ out: dict[str, Any] = {} for agg in select_corpus_aggregators(profile): try: value = agg.func(document_results) except Exception as exc: # noqa: BLE001 logger.warning( "[aggregate_%s] fonctionnalité dégradée : %s", agg.name, exc, ) continue if value is not None: out[agg.attribute] = value return out # ────────────────────────────────────────────────────────────────────────── # Helpers test-only # ────────────────────────────────────────────────────────────────────────── def _reset_for_tests() -> None: """Vide les registres globaux. **Réservé aux tests** — désactive toutes les métriques en production.""" _DOCUMENT_HOOKS.clear() _CORPUS_AGGREGATORS.clear() def _all_document_hook_names() -> list[str]: return [h.name for h in _DOCUMENT_HOOKS] def _all_corpus_aggregator_names() -> list[str]: return [a.name for a in _CORPUS_AGGREGATORS] __all__ = [ "PROFILE_MINIMAL", "PROFILE_STANDARD", "PROFILE_PHILOLOGICAL", "PROFILE_DIAGNOSTICS", "PROFILE_ECONOMICS", "PROFILE_PIPELINE", "PROFILE_FULL", "KNOWN_PROFILES", "validate_profile", "DocumentMetricHook", "CorpusMetricAggregator", "register_document_metric", "register_corpus_aggregator", "select_document_hooks", "select_corpus_aggregators", "run_document_hooks", "run_corpus_aggregators", ]