def compute_kde_scores(
nx_graph: Any,
crime_points: list[dict],
bandwidth_m: float | str | None = None,
arrays: GraphArrays | None = None,
) -> KDEResult:
"""
Calcule les scores KDE pour chaque arc du graphe.
Args:
nx_graph: graphe NetworkX (OSMnx MultiDiGraph)
crime_points: liste de {"lat", "lon", "weight"} (weight par défaut : 1.0)
bandwidth_m: écart-type du noyau en mètres. None → règle de Silverman (bornée à [200 ; 800] m,
qui sur-lisse sur Londres : voir docs/benchmarks.md) ; "auto" → choisi parmi
AUTO_CANDIDATES_M par log-vraisemblance de crimes mis de côté
arrays: colonnes déjà calculées de `nx_graph` (évite de reparcourir le graphe)
Returns:
KDEResult avec risk_map normalisé [0, 1] (le maximum sur les arcs vaut 1).
"""
t0 = time.perf_counter()
if not crime_points:
logger.warning("Aucun crime fourni — scores de risque tous à 0.0")
return KDEResult(
risk_map={}, bandwidth_deg=0, bandwidth_m=0, method="none", n_crimes=0, elapsed_s=0, stats={}
)
lats = np.array([c["lat"] for c in crime_points], dtype=np.float64)
lons = np.array([c["lon"] for c in crime_points], dtype=np.float64)
weights = np.array([c.get("weight", 1.0) for c in crime_points], dtype=np.float64)
valid = np.isfinite(lats) & np.isfinite(lons) & np.isfinite(weights) & (weights >= 0)
if not valid.all():
logger.warning("%d crime(s) ignoré(s) (coordonnée ou poids invalide)", int((~valid).sum()))
lats, lons, weights = lats[valid], lons[valid], weights[valid]
if lats.size == 0 or weights.sum() <= 0:
return KDEResult(
risk_map={}, bandwidth_deg=0, bandwidth_m=0, method="none",
n_crimes=len(crime_points), elapsed_s=0, stats={},
)
# ── Calibration du bandwidth ───────────────────────────────────────────────
cv_scores: dict[float, float] | None = None
clamp = True
if isinstance(bandwidth_m, str):
if bandwidth_m != "auto":
raise ValueError(f'bandwidth_m doit être un nombre, None ou "auto", reçu : {bandwidth_m!r}')
months_all = [c.get("month") for c in crime_points]
months = np.array([m for m, ok in zip(months_all, valid, strict=True) if ok], dtype=object)
best_m, cv_scores, cv_protocol = select_bandwidth(lats, lons, weights, months)
bw_deg, method, clamp = best_m / M_PER_DEG, f"auto_cv_{cv_protocol}", False
elif bandwidth_m is not None:
if not np.isfinite(bandwidth_m) or bandwidth_m <= 0:
raise ValueError(f"bandwidth_m doit être un nombre > 0, reçu : {bandwidth_m}")
# Une valeur choisie explicitement est respectée telle quelle (avant 0.1.2, 100 m devenait
# silencieusement 200 m). Seule la règle de Silverman est bornée à [200 ; 800] m.
bw_deg, method, clamp = bandwidth_m / M_PER_DEG, "fixed", False
else:
bw_deg, method = _calibrate_bandwidth(lats, lons, weights)
# Seule la règle de Silverman est bornée à [200 ; 800] m.
if clamp:
bw_deg = float(np.clip(bw_deg, BANDWIDTH_MIN_DEG, BANDWIDTH_MAX_DEG))
bw_m = bw_deg * M_PER_DEG
logger.info("KDE bandwidth : %.0f m (%.5f°) — méthode : %s — %d crimes", bw_m, bw_deg, method, lats.size)
# ── Évaluation aux milieux d'arcs ──────────────────────────────────────────
if arrays is None:
arrays = GraphArrays.from_networkx(nx_graph)
if arrays.n_edges == 0:
return KDEResult(
risk_map={}, bandwidth_deg=bw_deg, bandwidth_m=bw_m, method=method,
n_crimes=len(crime_points), elapsed_s=0, stats={},
)
mid_lat, mid_lon = arrays.midpoints()
lat0 = float(np.mean(mid_lat))
cos0 = float(np.cos(np.radians(lat0)))
lon0 = float(np.mean(mid_lon))
def to_xy(la: np.ndarray, lo: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
return (lo - lon0) * cos0 * M_PER_DEG, (la - lat0) * M_PER_DEG
cx, cy = to_xy(lats, lons)
qx, qy = to_xy(mid_lat, mid_lon)
raw = binned_gaussian_density(cx, cy, weights, qx, qy, bw_m)
# ── Normalisation [0, 1] par le maximum sur les arcs ───────────────────────
max_score = float(raw.max())
normalized = raw / max_score if max_score > 0 else raw
normalized = np.clip(normalized, 0.0, 1.0)
elapsed = time.perf_counter() - t0
stats: dict[str, Any] = {
"min": float(normalized.min()),
"max": float(normalized.max()),
"mean": float(normalized.mean()),
"std": float(normalized.std()),
"p50": float(np.percentile(normalized, 50)),
"p90": float(np.percentile(normalized, 90)),
"p99": float(np.percentile(normalized, 99)),
}
if cv_scores is not None:
stats["bandwidth_cv_loglik"] = {f"{k:.0f}": round(v, 4) for k, v in cv_scores.items()}
logger.info(
"KDE terminé en %.2fs — %d arcs scorés | mean=%.3f, p90=%.3f, p99=%.3f",
elapsed, arrays.n_edges, stats["mean"], stats["p90"], stats["p99"],
)
return KDEResult(
risk_map=RiskMap(arrays, normalized),
bandwidth_deg=bw_deg,
bandwidth_m=bw_m,
method=method,
n_crimes=len(crime_points),
elapsed_s=elapsed,
stats=stats,
edge_risk=normalized,
)