Spaces:
Sleeping
Sleeping
File size: 1,718 Bytes
c8a8b27 | 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 | """Aggregate cleaned records into per-H3-cell hotspot statistics."""
from collections import Counter
import pandas as pd
from src import config
from src.features import cell_to_latlng
def _mode_or_blank(series):
s = series.dropna()
return s.mode().iat[0] if not s.empty else ""
def _main_junction(series):
s = series.fillna("No Junction")
s = s[s.str.strip().str.lower() != "no junction"]
return s.mode().iat[0] if not s.empty else ""
def _top_violation(series):
c = Counter(v for lst in series for v in lst if v in config.PARKING_SEVERITY)
return c.most_common(1)[0][0] if c else ""
def build_cell_stats(df, total_days):
"""One row per H3 cell with the components the CII is built from."""
g = df.groupby("h3")
stats = g.agg(
n_violations=("id", "size"),
weighted_volume=("severity", "sum"),
active_days=("date", "nunique"),
peak_violations=("is_peak", "sum"),
junction_share=("has_junction", "mean"),
)
stats["persistence"] = stats["active_days"] / float(total_days)
stats["peak_share"] = stats["peak_violations"] / stats["n_violations"]
# representative centroid for each hexagon (for map centring / scatter)
centroids = {c: cell_to_latlng(c) for c in stats.index}
stats["lat"] = [centroids[c][0] for c in stats.index]
stats["lon"] = [centroids[c][1] for c in stats.index]
# human-readable context
stats["location"] = g["location"].agg(_mode_or_blank)
stats["police_station"] = g["police_station"].agg(_mode_or_blank)
stats["junction_name"] = g["junction_name"].agg(_main_junction)
stats["top_violation"] = g["violations"].agg(_top_violation)
return stats.reset_index()
|