Spaces:
Sleeping
Sleeping
File size: 5,535 Bytes
d641f6e 1e8b84c d641f6e 1e8b84c d641f6e 1e8b84c d641f6e 1e8b84c d641f6e 1e8b84c d641f6e 1e8b84c | 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 | """Front Pareto coût/qualité (Sprint 19).
Construit trois fronts Pareto avec des axes alternatifs :
- ``cost`` — CER vs coût € / 1000 pages.
- ``speed`` — CER vs durée moyenne par page.
- ``co2`` — CER vs empreinte carbone (g CO₂ / 1000 pages, expérimental).
API
---
Deux fonctions séparées pour rendre le contrat explicite :
1. :func:`attach_engine_costs` — **mute en place** ``engines_summary``
en y ajoutant ``mean_duration_seconds`` et ``cost`` (extraits du
benchmark et de la table de pricing). Le nom dit clairement qu'il
y a mutation.
2. :func:`build_pareto_section` — **fonction pure**, lit les coûts
déjà attachés à ``engines_summary``. Retourne le dict ``pareto``
prêt pour le template.
L'orchestrateur (``__init__.py``) appelle les deux dans l'ordre.
Cette séparation rend possible :
- Tester :func:`build_pareto_section` indépendamment avec un
``engines_summary`` pré-fabriqué.
- Réutiliser les coûts attachés sans recalculer Pareto.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from picarones.measurements.pricing import (
build_costs_for_benchmark,
load_pricing_database,
)
from picarones.measurements.statistics import compute_pareto_front
if TYPE_CHECKING:
from picarones.core.results import BenchmarkResult
def attach_engine_costs(
engines_summary: list[dict], benchmark: "BenchmarkResult",
) -> None:
"""Annote chaque entrée de ``engines_summary`` avec son coût.
**Mute en place** : ajoute deux champs à chaque dict moteur :
- ``mean_duration_seconds`` (float ou ``None`` si pas de durée).
- ``cost`` : dict de la forme ``{cost_per_1k_pages_eur: ...,
co2_per_1k_pages_g: ..., ...}`` ou ``None`` si pricing
indisponible.
Doit être appelée AVANT :func:`build_pareto_section`, qui lit
ces deux champs.
"""
durations_by_engine: dict[str, float] = {}
for report in benchmark.engine_reports:
durs = [
dr.duration_seconds
for dr in report.document_results
if dr.duration_seconds is not None
]
if durs:
durations_by_engine[report.engine_name] = sum(durs) / len(durs)
costs_by_engine = build_costs_for_benchmark(
engines_summary, durations_by_engine,
)
for entry in engines_summary:
name = entry["name"]
entry["mean_duration_seconds"] = (
round(durations_by_engine.get(name, 0.0), 4)
if name in durations_by_engine else None
)
entry["cost"] = costs_by_engine.get(name)
def build_pareto_section(engines_summary: list[dict]) -> dict:
"""Construit le bloc ``pareto`` du dict de rapport.
**Fonction pure** : ne mute rien. Lit ``mean_duration_seconds``
et ``cost`` qui doivent avoir été attachés en amont par
:func:`attach_engine_costs`. Si ces champs sont absents, le
moteur est silencieusement omis du front (cohérent avec un
moteur qui n'a pas de prix connu).
Retour
------
dict
Trois fronts Pareto (``cost``, ``speed``, ``co2``) plus
``pricing_meta`` (table de pricing utilisée).
"""
pricing_defaults, _ = load_pricing_database()
pareto_points = []
for entry in engines_summary:
cer = entry.get("cer")
cost = (entry.get("cost") or {}).get("cost_per_1k_pages_eur")
if cer is None or cost is None:
continue
pareto_points.append({"engine": entry["name"], "cer": cer, "cost": cost})
pareto_front_engines = compute_pareto_front(
pareto_points, objectives=("cer", "cost"),
)
pareto_speed_points = []
for entry in engines_summary:
cer = entry.get("cer")
dur = entry.get("mean_duration_seconds")
if cer is None or dur is None:
continue
pareto_speed_points.append({"engine": entry["name"], "cer": cer, "dur": dur})
pareto_front_speed = compute_pareto_front(
pareto_speed_points, objectives=("cer", "dur"),
)
pareto_co2_points = []
for entry in engines_summary:
cer = entry.get("cer")
co2 = (entry.get("cost") or {}).get("co2_per_1k_pages_g")
if cer is None or co2 is None:
continue
pareto_co2_points.append({"engine": entry["name"], "cer": cer, "co2": co2})
pareto_front_co2 = compute_pareto_front(
pareto_co2_points, objectives=("cer", "co2"),
)
return {
"cost": {
"points": pareto_points,
"front": pareto_front_engines,
"axis_label": "Coût (€ / 1000 pages)",
},
"speed": {
"points": pareto_speed_points,
"front": pareto_front_speed,
"axis_label": "Temps moyen (s / page)",
},
"co2": {
"points": pareto_co2_points,
"front": pareto_front_co2,
"axis_label": (
"Empreinte carbone (g CO₂ / 1000 pages, expérimental)"
),
},
"pricing_meta": {
"last_updated": pricing_defaults.last_updated,
"currency": pricing_defaults.currency,
"hourly_rate_local_cpu_eur": pricing_defaults.hourly_rate_local_cpu_eur,
"hourly_rate_local_gpu_eur": pricing_defaults.hourly_rate_local_gpu_eur,
"grid_intensity_local": pricing_defaults.grid_intensity_local,
"grid_intensity_cloud": pricing_defaults.grid_intensity_cloud,
},
}
__all__ = ["attach_engine_costs", "build_pareto_section"]
|