File size: 8,928 Bytes
03e7c21
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
979f3c3
03e7c21
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
"""Rendu HTML « Absorption d'erreur » — Sprint 94 (B.3).

Suite directe ``picarones/core/error_absorption.py``.  Pattern
identique aux autres rendus : server-side, pas de JS, anti-
injection systématique.

Vue
---
Tableau résumé des jonctions du pipeline ; chaque ligne décrit
un module post-correction et présente :

- erreurs en entrée vs en sortie ;
- nb corrigées (gradient vert), nb introduites (gradient rouge) ;
- taux de correction (gradient vert), taux d'introduction
  (gradient rouge) ;
- amélioration nette (n_corrected - n_introduced) — coloré.
- éventuellement un échantillon de tokens corrigés/introduits.

Adaptive : ``""`` si la liste est vide.

Note d'intégration
------------------
Module pur — la liste ``junctions`` est composée par
l'utilisateur depuis son benchmark de pipeline composée :

.. code-block:: python

    from picarones.measurements.error_absorption import (
        compute_error_absorption, aggregate_error_absorption,
    )
    from picarones.report.error_absorption_render import (
        build_error_absorption_html,
    )

    junctions = []
    for step in pipeline.steps_with_text_output:
        per_doc = [
            compute_error_absorption(doc.gt_text, doc.before_text,
                                     doc.after_text)
            for doc in benchmark.docs
        ]
        agg = aggregate_error_absorption(per_doc)
        if agg is not None:
            agg["junction_name"] = step.name
            junctions.append(agg)
    html = build_error_absorption_html(junctions, labels)
"""

from __future__ import annotations

from html import escape as _e
from typing import Optional


def _color_for_correction(rate: float) -> str:
    """Faible (rouge) → élevé (vert) — bon = beaucoup corrigées."""
    f = max(0.0, min(1.0, rate))
    if f < 0.5:
        t = f / 0.5
        r = 235
        g = int(70 + (200 - 70) * t)
        b = 70
    else:
        t = (f - 0.5) / 0.5
        r = int(235 + (60 - 235) * t)
        g = int(200 + (160 - 200) * t)
        b = int(70 + (90 - 70) * t)
    return f"#{r:02x}{g:02x}{b:02x}"


def _color_for_introduction(rate: float) -> str:
    """Faible (vert) → élevé (rouge) — bon = peu introduites."""
    f = max(0.0, min(1.0, rate))
    if f < 0.5:
        t = f / 0.5
        r = int(60 + (235 - 60) * t)
        g = int(160 + (180 - 160) * t)
        b = int(90 + (60 - 90) * t)
    else:
        t = (f - 0.5) / 0.5
        r = int(235 + (220 - 235) * t)
        g = int(180 + (50 - 180) * t)
        b = int(60 + (50 - 60) * t)
    return f"#{r:02x}{g:02x}{b:02x}"


def _color_for_net(net: int, max_abs: int) -> str:
    """Vert si positif, rouge si négatif. Saturation à max_abs."""
    if max_abs <= 0 or net == 0:
        return "#a7f0a7"
    f = max(-1.0, min(1.0, net / max_abs))
    if f >= 0:
        # vert clair → vert profond
        r = int(167 + (90 - 167) * f)
        g = int(240 + (200 - 240) * f)
        b = int(167 + (90 - 167) * f)
    else:
        f = -f
        r = int(167 + (220 - 167) * f)
        g = int(240 + (50 - 240) * f)
        b = int(167 + (50 - 167) * f)
    return f"#{r:02x}{g:02x}{b:02x}"


def build_error_absorption_html(
    junctions: Optional[list],
    labels: Optional[dict[str, str]] = None,
    *,
    sample_max: int = 8,
) -> str:
    """Construit la vue HTML « Absorption d'erreur ».

    Parameters
    ----------
    junctions:
        Liste de dicts (un par jonction de pipeline), enrichis
        d'un ``junction_name``.  Si vide ou ``None``, retourne
        ``""``.
    labels:
        Dict i18n.  Clés sous le préfixe ``absorption_*``.
    sample_max:
        Nombre maximal de tokens corrigés/introduits affichés
        en cellule échantillon.
    """
    if not junctions:
        return ""
    rows = [
        j for j in junctions
        if isinstance(j, dict) and j.get("junction_name")
    ]
    if not rows:
        return ""
    labels = labels or {}
    title = labels.get(
        "absorption_title", "Absorption d'erreur par jonction",
    )
    note = labels.get(
        "absorption_note",
        "À chaque jonction du pipeline, deux flux sont mesurés "
        "indépendamment : combien d'erreurs sont corrigées et "
        "combien sont introduites. Une jonction qui corrige "
        "beaucoup mais introduit aussi beaucoup absorbe les "
        "différences amont au lieu de les améliorer.",
    )
    h_junction = labels.get("absorption_junction", "Jonction")
    h_errors_before = labels.get("absorption_errors_before", "Erreurs avant")
    h_errors_after = labels.get("absorption_errors_after", "Erreurs après")
    h_corrected = labels.get("absorption_corrected", "Corrigées")
    h_introduced = labels.get("absorption_introduced", "Introduites")
    h_corr_rate = labels.get("absorption_corr_rate", "% corrigées")
    h_intro_rate = labels.get("absorption_intro_rate", "% introduites")
    h_net = labels.get("absorption_net", "Amélioration nette")
    h_sample = labels.get("absorption_sample", "Échantillon (intro)")

    # Saturation pour le gradient « net »
    max_abs_net = max(
        (abs(int(r.get("net_improvement") or 0)) for r in rows), default=1,
    ) or 1

    parts = [
        '<section class="absorption-section" style="margin:1rem 0">',
        f'<h3 style="margin:0 0 .3rem 0">{_e(title)}</h3>',
        f'<div style="font-size:.85rem;opacity:.75;margin-bottom:.6rem">'
        f'{_e(note)}</div>',
        '<table style="border-collapse:collapse;width:100%;'
        'font-size:.9rem">',
        '<thead><tr>',
    ]
    for col in (h_junction, h_errors_before, h_errors_after,
                h_corrected, h_introduced, h_corr_rate,
                h_intro_rate, h_net, h_sample):
        parts.append(
            f'<th style="padding:.4rem .6rem;text-align:left;'
            f'border-bottom:1px solid #ccc;font-weight:600">'
            f'{_e(col)}</th>'
        )
    parts.append("</tr></thead><tbody>")
    for entry in rows:
        name = str(entry.get("junction_name") or "?")
        n_eb = int(entry.get("n_errors_before") or 0)
        n_ea = int(entry.get("n_errors_after") or 0)
        n_corr = int(entry.get("n_corrected") or 0)
        n_intro = int(entry.get("n_introduced") or 0)
        net = int(entry.get("net_improvement") or 0)
        corr_rate = entry.get("correction_rate")
        intro_rate = entry.get("introduction_rate")
        if isinstance(corr_rate, (int, float)):
            corr_rate_str = f"{corr_rate * 100:.1f}%"
            corr_color = _color_for_correction(float(corr_rate))
            corr_cell = (
                f'<td style="padding:.4rem .6rem;text-align:right;'
                f'background:{corr_color};font-family:monospace;'
                f'font-weight:600">{corr_rate_str}</td>'
            )
        else:
            corr_cell = (
                '<td style="padding:.4rem .6rem;text-align:right;'
                'opacity:.4">—</td>'
            )
        if isinstance(intro_rate, (int, float)):
            intro_rate_str = f"{intro_rate * 100:.1f}%"
            intro_color = _color_for_introduction(float(intro_rate))
            intro_cell = (
                f'<td style="padding:.4rem .6rem;text-align:right;'
                f'background:{intro_color};font-family:monospace;'
                f'font-weight:600">{intro_rate_str}</td>'
            )
        else:
            intro_cell = (
                '<td style="padding:.4rem .6rem;text-align:right;'
                'opacity:.4">—</td>'
            )
        net_color = _color_for_net(net, max_abs_net)
        intro_sample = entry.get("introduced_tokens_sample") or []
        sample_cell_text = ", ".join(
            _e(str(t)) for t in intro_sample[:sample_max]
        ) or "—"
        if len(intro_sample) > sample_max:
            sample_cell_text += " …"
        parts.append(
            f'<tr>'
            f'<td style="padding:.4rem .6rem">{_e(name)}</td>'
            f'<td style="padding:.4rem .6rem;text-align:right;'
            f'font-family:monospace">{n_eb}</td>'
            f'<td style="padding:.4rem .6rem;text-align:right;'
            f'font-family:monospace">{n_ea}</td>'
            f'<td style="padding:.4rem .6rem;text-align:right;'
            f'font-family:monospace">{n_corr}</td>'
            f'<td style="padding:.4rem .6rem;text-align:right;'
            f'font-family:monospace">{n_intro}</td>'
            f'{corr_cell}'
            f'{intro_cell}'
            f'<td style="padding:.4rem .6rem;text-align:right;'
            f'background:{net_color};font-family:monospace;'
            f'font-weight:600">{net:+d}</td>'
            f'<td style="padding:.4rem .6rem;font-family:monospace;'
            f'font-size:.8rem">{sample_cell_text}</td>'
            f'</tr>'
        )
    parts.append("</tbody></table></section>")
    return "".join(parts)


__all__ = ["build_error_absorption_html"]