"""Text/markdown renderers + unicode bar & radar charts (zero dependencies).""" import math from typing import Dict, List, Union from ...eval.record import EvalReport from .. import register_renderer def _bars(value: float, width: int = 30, char: str = '█') -> str: filled = int(round(max(0.0, min(1.0, value)) * width)) return char * filled + '·' * (width - filled) def _pct(value: float) -> str: return f'{value * 100:.1f}%' @register_renderer('text') def text_table(target: Union[EvalReport, List[EvalReport]], opts: Dict) -> str: reports = target if isinstance(target, list) else [target] out: List[str] = [] for rep in reports: # headline: dedupe dataset/recipe when identical; join facts compactly title = rep.dataset if rep.dataset == rep.recipe else f'{rep.dataset} [{rep.recipe}]' facts = [f'model={rep.model or "?"}', f'n={rep.num_samples}'] info = rep.metric_groups.get('run_info', {}) or {} secs = sum(float((s.usage or {}).get('latency_s', 0) or 0) for s in rep.samples) if secs >= 3600: facts.append(f'time={secs / 3600:.2f}h') elif secs: facts.append(f'time={secs:.0f}s') if info.get('gen_total_tokens'): facts.append(f'tokens={info["gen_total_tokens"]}') head = f'{title} · ' + ' '.join(facts) out.append(head) out.append('=' * max(len(head), 40)) if rep.num_failed_extractions: warn = (f'!! {rep.num_failed_extractions}/{rep.num_samples} extractions failed ' f'({_pct(rep.metrics.get("extraction_failure_rate", 0))}) -- check recipe/model fit') out.append(warn) metrics = [(m, v) for m, v in rep.metrics.items() if m != 'extraction_failure_rate' and isinstance(v, (int, float))] w = max([len(m) for m, _ in metrics] + [12]) # adaptive, long names survive for metric, value in metrics: out.append(f'{metric:<{w}} {_pct(value):>8} {_bars(value)}') for group_name, groups in rep.metric_groups.items(): if group_name == 'run_info' or group_name.startswith('agg_error'): continue numeric = {g: v for g, v in groups.items() if isinstance(v, (int, float))} if not numeric: continue gw = max([len(str(g)) for g in numeric] + [12]) out.append(f'-- {group_name} ' + '-' * max(0, 30 - len(group_name))) for g, v in numeric.items(): out.append(f' {str(g):<{gw}} {_pct(v):>8} {_bars(v, 20)}') out.append('') return '\n'.join(out) @register_renderer('md') def markdown(target: Union[EvalReport, List[EvalReport]], opts: Dict) -> str: reports = target if isinstance(target, list) else [target] out = ['# Eval Report', ''] for rep in reports: out += [f'## {rep.dataset} (`{rep.recipe}`)', '', f'- model: `{rep.model or "?"}` samples: {rep.num_samples} ' f'created: {rep.created_at}', ''] rows = ['| metric | value |', '|---|---|'] for metric, value in rep.metrics.items(): rows.append(f'| {metric} | {_pct(value) if metric != "extraction_failure_rate" else _pct(value)} |') out += rows + [''] for gname, groups in rep.metric_groups.items(): if gname in ('run_info',) or gname.startswith('agg_error') or not isinstance(groups, dict): continue out += [f'### {gname}', '', '| group | value |', '|---|---|'] out += [f'| {g} | {_pct(v) if isinstance(v, (int, float)) else v} |' for g, v in groups.items()] out.append('') return '\n'.join(out) @register_renderer('md_compare') def md_compare(target: List[EvalReport], opts: Dict) -> str: """Side-by-side metric table for N reports (models on one bench, or many benches of one model -- the shape is the same: one row per benchmark). The first report is the baseline; later columns get a delta marker.""" if not isinstance(target, list) or len(target) < 1: raise ValueError('md_compare needs a list of reports') # many DIFFERENT benchmarks, each with its own primary metric: a compact # one-row-per-bench table beats a sparse metric-x-bench grid datasets = {r.dataset for r in target} primary_metrics = {next((m for m in r.metrics if m != 'extraction_failure_rate'), '') for r in target} if len(datasets) > 1 and len(target) == len(datasets) and len(primary_metrics) > 1: out = ['# Comparison', '', '| benchmark | metric | score | n |', '|---|---|---:|---:|'] best = max((next((v for m, v in r.metrics.items() if m != 'extraction_failure_rate'), 0.0) for r in target)) for r in target: m = next((m for m in r.metrics if m != 'extraction_failure_rate'), '') v = r.metrics.get(m, 0.0) cell = f'**{_pct(v)}**' if v == best and len(target) > 1 else _pct(v) out.append(f'| {r.dataset} | {m} | {cell} | {r.num_samples} |') return '\n'.join(out + ['']) metrics: List[str] = [] for rep in target: for m in rep.metrics: if m not in metrics and m != 'extraction_failure_rate': metrics.append(m) cols = [] for rep in target: # models on one bench -> show the model; many benches -> show the bench same_dataset = len({r.dataset for r in target}) == 1 name = rep.model or '?' if same_dataset else rep.dataset cols.append(name) out = ['# Comparison', '', '| metric | ' + ' | '.join(cols) + ' |', '|---' * (len(cols) + 1) + '|'] for m in metrics: cells = [] base = target[0].metrics.get(m, 0.0) for rep, col_i in zip(target, range(len(target))): v = rep.metrics.get(m, 0.0) cell = _pct(v) best = max(r.metrics.get(m, 0.0) for r in target) if v == best and len(target) > 1: cell = f'**{cell}**' if rep is not target[0] and isinstance(base, (int, float)): d = v - base if d > 0.0005: cell += f' ▲{d * 100:+.1f}' elif d < -0.0005: cell += f' ▼{d * 100:+.1f}' cells.append(cell) out.append(f'| {m} | ' + ' | '.join(cells) + ' |') out.append('') return '\n'.join(out) @register_renderer('radar') def radar(target: Union[EvalReport, List[EvalReport]], opts: Dict) -> str: """Unicode radar chart over each report's metric_groups entries. opts: group (default: first non-run_info group), top (default 10 axes). """ reports = target if isinstance(target, list) else [target] group = opts.get('group') axes: List[str] = [] series: List[Dict[str, float]] = [] for rep in reports: gname = group or next((k for k in rep.metric_groups if k != 'run_info' and not k.startswith('agg_error') and isinstance(rep.metric_groups[k], dict)), None) data = rep.metric_groups.get(gname, {}) if gname else {} data = {k: v for k, v in data.items() if isinstance(v, (int, float))} if not data: return f'(no grouped metrics to chart for {rep.dataset})' top = opts.get('top', 10) picked = sorted(data.items(), key=lambda kv: -kv[1])[:top] axes = [k for k, _ in picked] series.append(dict(picked)) # simple ascii radar: axis list + per-report bars side by side W = 24 header = 'axis'.ljust(26) + ''.join((rep.dataset[:12] or '?').rjust(14) for rep in reports) lines = [header, '-' * len(header)] for ax in axes: row = ax[:24].ljust(26) for rep, s in zip(reports, series): v = s.get(ax, 0.0) row += (_bars(v, W // 2)[:W // 2] + f'{v * 100:5.1f}%').rjust(14) lines.append(row) return '\n'.join(lines) @register_renderer('errors') def errors(target: Union[EvalReport, List[EvalReport]], opts: Dict) -> str: """Failed samples browser: worst-N samples with extraction + target.""" rep = target[0] if isinstance(target, list) else target n = opts.get('n', 10) only_failed = opts.get('only_failed', True) rows = [r for r in rep.samples if r.error or (not r.extraction_ok if only_failed else False)] rows = sorted(rows, key=lambda r: sum(r.scores.values()))[:n] out = [f'# Errors / failed extractions: {rep.dataset} ({len(rows)} shown)', ''] for r in rows: out.append(f'## sample {r.sample_id} scores={r.scores}') out.append(f'- extraction_ok={r.extraction_ok} note={r.extraction_note!r}') out.append(f'- target: {str(r.target)[:120]!r}') out.append(f'- extracted: {r.extracted_prediction[:120]!r}') if r.error: out.append(f'- error: {r.error[:300]}') out.append(f'- raw: {r.raw_prediction[:200]!r}') out.append('') return '\n'.join(out)