feat(observatoire): calcul et rendu tuile considérations sociales/env

Ajoute compute_considerations_stats (dédoublonnage par uid, regex
Clause|Critère|Marché réservé, colonne absente → 0%) et
get_considerations_card_content (deux dbc.Progress colorés Safe).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Colin Maudry
2026-06-22 18:16:34 +02:00
parent 2ce762b6cc
commit a85baca408
2 changed files with 239 additions and 0 deletions
+80
View File
@@ -728,6 +728,86 @@ def get_dashboard_summary_table(dff, dff_per_uid, nb_marches):
return summary_table return summary_table
CONSIDERATIONS_REGEX = r"(?i)Clause|Critère|Marché réservé"
CONSIDERATIONS_COLUMNS = {
"sociales": "considerationsSociales",
"environnementales": "considerationsEnvironnementales",
}
def compute_considerations_stats(lff: pl.LazyFrame) -> dict[str, tuple[int, int]]:
"""Part des marchés (uid distincts) ayant au moins une considération.
Renvoie {"sociales": (count, pct), "environnementales": (count, pct)}.
Dénominateur = tous les uid distincts. Colonne absente -> (0, 0).
"""
names = lff.collect_schema().names()
present = {key: col for key, col in CONSIDERATIONS_COLUMNS.items() if col in names}
stats = {key: (0, 0) for key in CONSIDERATIONS_COLUMNS}
if not present:
return stats
agg = (
lff.select(["uid"] + list(present.values()))
.group_by("uid")
.agg([pl.col(col).first() for col in present.values()])
.collect(engine="streaming")
)
total = agg.height
if total == 0:
return stats
for key, col in present.items():
count = agg.filter(pl.col(col).str.contains(CONSIDERATIONS_REGEX)).height
pct = round(100 * count / total)
stats[key] = (count, pct)
return stats
CONSIDERATIONS_DISPLAY = [
# (clé, libellé, couleur Safe)
("sociales", "Sociales", "rgb(204, 102, 119)"),
("environnementales", "Environnementales", "rgb(17, 119, 51)"),
]
def get_considerations_card_content(lff: pl.LazyFrame) -> html.Div:
"""Deux barres de progression : part des marchés avec considération."""
stats = compute_considerations_stats(lff)
blocks = []
for key, label, color in CONSIDERATIONS_DISPLAY:
count, pct = stats[key]
blocks.append(
html.Div(
className="mb-3",
children=[
html.Div(
className="d-flex justify-content-between",
children=[
html.Span(label),
html.Span(
f"{format_number(count)} marchés",
className="text-muted",
),
],
),
dbc.Progress(
value=pct,
label=f"{pct} %",
style={"backgroundColor": color},
),
],
)
)
return html.Div(children=blocks)
def make_card( def make_card(
title: str, subtitle=None, fig=None, paragraphs=None, lg=6, xl=4 title: str, subtitle=None, fig=None, paragraphs=None, lg=6, xl=4
) -> dbc.Col: ) -> dbc.Col:
+159
View File
@@ -0,0 +1,159 @@
import polars as pl
def _make_lff(rows):
return pl.LazyFrame(rows)
def test_compute_considerations_stats_basic():
from src.figures import compute_considerations_stats
lff = _make_lff(
[
# uid u1 : social oui (Clause), env non (Sans objet)
{
"uid": "u1",
"considerationsSociales": "Clause sociale",
"considerationsEnvironnementales": "Sans objet",
},
# uid u2 : social non (Sans objet), env oui (Critère)
{
"uid": "u2",
"considerationsSociales": "Sans objet",
"considerationsEnvironnementales": "Critère environnemental",
},
# uid u3 : social oui (Marché réservé compte), env null
{
"uid": "u3",
"considerationsSociales": "Marché réservé",
"considerationsEnvironnementales": None,
},
# uid u4 : aucune considération
{
"uid": "u4",
"considerationsSociales": "Pas de considération sociale",
"considerationsEnvironnementales": "Sans objet",
},
]
)
stats = compute_considerations_stats(lff)
# 4 marchés au total. Social : u1, u3 -> 2/4 = 50%. Env : u2 -> 1/4 = 25%.
assert stats["sociales"] == (2, 50)
assert stats["environnementales"] == (1, 25)
def test_compute_considerations_stats_dedup_per_uid():
from src.figures import compute_considerations_stats
lff = _make_lff(
[
# uid u1 présent 2 fois (2 titulaires) -> compté une seule fois
{
"uid": "u1",
"considerationsSociales": "Clause sociale",
"considerationsEnvironnementales": "Sans objet",
},
{
"uid": "u1",
"considerationsSociales": "Clause sociale",
"considerationsEnvironnementales": "Sans objet",
},
{
"uid": "u2",
"considerationsSociales": "Sans objet",
"considerationsEnvironnementales": "Sans objet",
},
]
)
stats = compute_considerations_stats(lff)
# 2 marchés distincts. Social : u1 -> 1/2 = 50%.
assert stats["sociales"] == (1, 50)
assert stats["environnementales"] == (0, 0)
def test_compute_considerations_stats_missing_column():
from src.figures import compute_considerations_stats
lff = _make_lff(
[
{"uid": "u1", "considerationsSociales": "Clause sociale"},
{"uid": "u2", "considerationsSociales": "Sans objet"},
]
)
stats = compute_considerations_stats(lff)
# Colonne env absente -> (0, 0) sans exception. Social : 1/2 = 50%.
assert stats["sociales"] == (1, 50)
assert stats["environnementales"] == (0, 0)
def test_compute_considerations_stats_empty():
from src.figures import compute_considerations_stats
lff = pl.LazyFrame(
{
"uid": pl.Series([], dtype=pl.String),
"considerationsSociales": pl.Series([], dtype=pl.String),
"considerationsEnvironnementales": pl.Series([], dtype=pl.String),
}
)
stats = compute_considerations_stats(lff)
assert stats["sociales"] == (0, 0)
assert stats["environnementales"] == (0, 0)
def test_get_considerations_card_content_returns_two_progress_bars():
import dash_bootstrap_components as dbc
from dash import html
from src.figures import get_considerations_card_content
lff = pl.LazyFrame(
[
{
"uid": "u1",
"considerationsSociales": "Clause sociale",
"considerationsEnvironnementales": "Sans objet",
},
{
"uid": "u2",
"considerationsSociales": "Sans objet",
"considerationsEnvironnementales": "Critère environnemental",
},
]
)
div = get_considerations_card_content(lff)
assert isinstance(div, html.Div)
# Récupère récursivement tous les dbc.Progress
def find_progress(component, found):
children = getattr(component, "children", None)
if isinstance(component, dbc.Progress):
found.append(component)
if isinstance(children, (list, tuple)):
for c in children:
find_progress(c, found)
elif children is not None:
find_progress(children, found)
return found
bars = find_progress(div, [])
assert len(bars) == 2
# Sociales (rouge) : u1 -> 50%. Environnementales (vert) : u2 -> 50%.
social_bar, env_bar = bars[0], bars[1]
assert social_bar.value == 50
assert social_bar.label == "50 %"
assert social_bar.style["backgroundColor"] == "rgb(204, 102, 119)"
assert env_bar.value == 50
assert env_bar.label == "50 %"
assert env_bar.style["backgroundColor"] == "rgb(17, 119, 51)"