"""Sprint A14-S16 — sanity check inter-vues sur le cas BnF central. Vérifie qu'un même pipeline a une cohérence (et parfois une divergence intéressante) entre TextView, AltoView et SearchView. Cas démontrés : - Pipeline parfait → toutes vues maximisent. - Pipeline avec erreur sur une année → SearchView baisse fortement, TextView baisse légèrement (pattern "perte de données critiques invisible au CER global"). - Pipeline sans ALTO → AltoView l'OMET, autres vues l'évaluent. """ from __future__ import annotations from picarones.domain import Artifact, ArtifactType, MetricSpec from picarones.evaluation.metrics.alto_structural import ( compute_alto_validity, compute_line_count_ratio, compute_word_box_coverage, ) from picarones.evaluation.metrics.search import ( numerical_sequence_preservation, searchability_recall, ) from picarones.evaluation.projectors import ( AltoToText, CanonicalToText, PageToText, ProjectorRegistry, ) from picarones.evaluation.registry import MetricRegistry from picarones.evaluation.views import ( DefaultEvaluationViewExecutor, build_alto_view, build_search_view, build_text_view, ) from picarones.formats.alto.types import ( AltoBBox, AltoDocument, AltoLine, AltoPage, AltoString, AltoTextBlock, ) # ────────────────────────────────────────────────────────────────── # Stubs métriques texte (cer/wer simplifiés sans jiwer) # ────────────────────────────────────────────────────────────────── def _stub_cer(reference: str, hypothesis: str) -> float: if not reference: return 0.0 if not hypothesis else 1.0 common = sum(1 for a, b in zip(reference, hypothesis) if a == b) return 1.0 - (common / max(len(reference), len(hypothesis))) def _stub_wer(reference: str, hypothesis: str) -> float: ref_w = reference.split() hyp_w = hypothesis.split() if not ref_w: return 0.0 if not hyp_w else 1.0 common = sum(1 for a, b in zip(ref_w, hyp_w) if a == b) return 1.0 - (common / len(ref_w)) def _build_unified_executor(payloads: dict) -> DefaultEvaluationViewExecutor: """Executor configuré pour TextView + AltoView + SearchView.""" metrics = MetricRegistry() # TextView metrics for name, fn in ( ("cer", _stub_cer), ("wer", _stub_wer), ("mer", _stub_cer), ("wil", _stub_wer), ): metrics.register( MetricSpec( name=name, input_types=(ArtifactType.RAW_TEXT, ArtifactType.RAW_TEXT), ), fn, ) # AltoView metrics for name, fn in ( ("alto_validity", compute_alto_validity), ("alto_line_count_ratio", compute_line_count_ratio), ("alto_word_box_coverage", compute_word_box_coverage), ): metrics.register( MetricSpec( name=name, input_types=(ArtifactType.ALTO_XML, ArtifactType.ALTO_XML), higher_is_better=True, ), fn, ) # SearchView metrics metrics.register( MetricSpec( name="searchability_recall", input_types=(ArtifactType.RAW_TEXT, ArtifactType.RAW_TEXT), higher_is_better=True, ), searchability_recall, ) metrics.register( MetricSpec( name="numerical_sequence_preservation", input_types=(ArtifactType.RAW_TEXT, ArtifactType.RAW_TEXT), higher_is_better=True, ), numerical_sequence_preservation, ) projectors = ProjectorRegistry() projectors.register(AltoToText()) projectors.register(PageToText()) projectors.register(CanonicalToText()) def loader(art: Artifact): if art.id not in payloads: raise KeyError(art.id) return payloads[art.id] return DefaultEvaluationViewExecutor.from_registries(metrics, projectors, loader) # ────────────────────────────────────────────────────────────────── # Cas 1 — pipeline parfait # ────────────────────────────────────────────────────────────────── class TestPerfectPipelineAcrossViews: def test_perfect_text_pipeline_maximizes_text_and_search(self) -> None: """Un pipeline qui produit du texte parfait : - TextView : CER = 0 - SearchView : recall = 1.0, year preservation = 1.0 - AltoView : OMIS (pas d'ALTO produit). """ gt_text = "Bonjour Paris en 1789" payloads = {"cand": gt_text, "gt_text": gt_text} executor = _build_unified_executor(payloads) text_view = build_text_view() search_view = build_search_view() alto_view = build_alto_view() cand = Artifact(id="cand", document_id="d", type=ArtifactType.RAW_TEXT) gt = Artifact(id="gt_text", document_id="d", type=ArtifactType.RAW_TEXT) text_result = executor.evaluate(text_view, cand, gt, pipeline_name="test") search_result = executor.evaluate(search_view, cand, gt, pipeline_name="test") assert text_result.metric_values["cer"] == 0.0 assert search_result.metric_values["searchability_recall"] == 1.0 assert search_result.metric_values["numerical_sequence_preservation"] == 1.0 # AltoView OMIS : le caller doit filtrer. assert not alto_view.accepts(cand.type) # ────────────────────────────────────────────────────────────────── # Cas 2 — divergence TextView ↔ SearchView # ────────────────────────────────────────────────────────────────── class TestDivergencePattern: def test_year_corruption_invisible_to_cer_visible_to_search(self) -> None: """Pattern critique : une corruption d'année (1 caractère sur ~50) est invisible côté CER mais catastrophique côté recherchabilité numérique. C'est précisément ce que le rapport BnF doit rendre visible — les deux vues racontent des histoires complémentaires. """ gt_text = "Charte signée à Paris le 14 juillet 1789 en présence du roi" # Hypothèse : le LLM a "corrigé" 1789 en 1798 (faute grossière). # Le reste du texte est identique. cand_text = "Charte signée à Paris le 14 juillet 1798 en présence du roi" payloads = {"cand": cand_text, "gt": gt_text} executor = _build_unified_executor(payloads) cand = Artifact(id="cand", document_id="d", type=ArtifactType.RAW_TEXT) gt = Artifact(id="gt", document_id="d", type=ArtifactType.RAW_TEXT) text_result = executor.evaluate(build_text_view(), cand, gt, pipeline_name="test") search_result = executor.evaluate(build_search_view(), cand, gt, pipeline_name="test") # CER ≈ 0.03 (3 chars sur ~58) assert text_result.metric_values["cer"] < 0.1, "CER doit rester faible" # WER : 1 mot changé sur 11 → 1/11 ≈ 0.09 assert text_result.metric_values["wer"] < 0.15 # Mais SearchView : 1789 (GT) n'est PAS dans hyp_years = [1798] # → preservation = 0.0 (catastrophique pour un historien). assert search_result.metric_values["numerical_sequence_preservation"] == 0.0 # Searchability : "1789" GT n'est pas matché à "1798" (distance 2, # MAIS la longueur est égale, fuzziness ≤ 2 le matche). # On vérifie juste qu'il y a un signal mesurable. assert search_result.metric_values["searchability_recall"] >= 0.8 # ────────────────────────────────────────────────────────────────── # Cas 3 — pipeline ALTO évaluable dans les 3 vues # ────────────────────────────────────────────────────────────────── def _build_simple_alto(words: list[str], n_lines: int = 1) -> AltoDocument: """Construit un AltoDocument avec ``words`` répartis sur ``n_lines`` lignes, chaque mot avec une bbox.""" chunks = [words[i::n_lines] for i in range(n_lines)] lines = tuple( AltoLine(strings=tuple( AltoString( content=w, bbox=AltoBBox(hpos=0, vpos=0, width=10, height=10), ) for w in chunk )) for chunk in chunks ) return AltoDocument(pages=(AltoPage(blocks=(AltoTextBlock(lines=lines),),),),) class TestAltoPipelineEvaluatedInThreeViews: def test_alto_pipeline_has_text_alto_search_results(self, tmp_path) -> None: """Un pipeline qui produit ALTO_XML est évaluable dans les 3 vues : TextView (via projection), AltoView (direct), SearchView (via projection). """ from picarones.formats.alto import write_alto words_gt = "Charte signée Paris 14 juillet 1789".split() words_cand = "Charte signée Paris 14 juillet 1789".split() # identique # n_lines=1 pour préserver l'ordre des mots dans l'extraction # (sinon ``alto_document_to_text`` produit des sauts de ligne # qui font diverger le CER d'une comparaison ligne unique). gt_alto = _build_simple_alto(words_gt, n_lines=1) cand_alto = _build_simple_alto(words_cand, n_lines=1) cand_alto_path = tmp_path / "cand.alto.xml" cand_alto_path.write_bytes(write_alto(cand_alto)) # Payloads : raw text pour les payloads projetés depuis ALTO, # AltoDocument pour la GT et le candidat ALTO direct. from picarones.evaluation.projectors import alto_document_to_text payloads = { "gt_text": " ".join(words_gt), "gt_alto": gt_alto, "cand": cand_alto, # AltoDocument pour AltoView "cand:projected_text": alto_document_to_text(cand_alto), } executor = _build_unified_executor(payloads) gt_text_art = Artifact(id="gt_text", document_id="d", type=ArtifactType.RAW_TEXT) gt_alto_art = Artifact(id="gt_alto", document_id="d", type=ArtifactType.ALTO_XML) cand_art = Artifact( id="cand", document_id="d", type=ArtifactType.ALTO_XML, uri=str(cand_alto_path), ) # TextView : projette ALTO → texte, compare au gt_text. text_result = executor.evaluate(build_text_view(), cand_art, gt_text_art, pipeline_name="test") assert text_result.metric_values["cer"] == 0.0 # SearchView : projette ALTO → texte, mesure recall + années. search_result = executor.evaluate(build_search_view(), cand_art, gt_text_art, pipeline_name="test") assert search_result.metric_values["searchability_recall"] == 1.0 # AltoView : compare ALTO direct contre ALTO GT. alto_result = executor.evaluate(build_alto_view(), cand_art, gt_alto_art, pipeline_name="test") assert alto_result.metric_values["alto_validity"] == 1.0 assert alto_result.metric_values["alto_line_count_ratio"] == 1.0 assert alto_result.metric_values["alto_word_box_coverage"] == 1.0 # ────────────────────────────────────────────────────────────────── # Cohérence globale : projection report présent ssi projection appliquée # ────────────────────────────────────────────────────────────────── class TestProjectionReportConsistency: def test_text_search_views_share_projection_report_pattern(self) -> None: """Pour un même candidat ALTO_XML évalué dans TextView et SearchView, les deux ViewResult doivent porter un projection_report (les deux vues projettent vers texte).""" gt_text = "test" gt_alto = _build_simple_alto(["test"], n_lines=1) from picarones.evaluation.projectors import alto_document_to_text from picarones.formats.alto import write_alto # Pour ce test on n'a pas besoin du fichier réel — on simule # via le payload_loader qui retourne directement le texte # extrait pour l'id "cand:projected_text". payloads = { "gt_text": gt_text, "cand:projected_text": alto_document_to_text(gt_alto), } # Mais le projecteur a besoin d'un URI. On contourne en # créant un fichier temporaire dans pytest fixture. # Pour ce test simple on écrit dans /tmp. import tempfile with tempfile.NamedTemporaryFile(suffix=".alto.xml", delete=False) as f: f.write(write_alto(gt_alto)) cand_uri = f.name executor = _build_unified_executor(payloads) cand = Artifact( id="cand", document_id="d", type=ArtifactType.ALTO_XML, uri=cand_uri, ) gt = Artifact(id="gt_text", document_id="d", type=ArtifactType.RAW_TEXT) text_result = executor.evaluate(build_text_view(), cand, gt, pipeline_name="test") search_result = executor.evaluate(build_search_view(), cand, gt, pipeline_name="test") # Les deux doivent avoir un projection_report (même projecteur). assert text_result.projection_report is not None assert search_result.projection_report is not None assert text_result.projection_report.projector_name == "alto_to_text" assert search_result.projection_report.projector_name == "alto_to_text"