aggregate

Time-period aggregation of raw traffic GeoPackage snapshots.

aggregate

Time-period aggregation of raw traffic GeoPackage snapshots.

Uses OSM-based segment matching for stable segment identity across time periods. Each HERE segment is mapped to an osm_composite_id via geometry hashing + spatial-join fallback, replacing the legacy row-index-based fid which was unstable across snapshots.

aggregate_city

aggregate_city(city_code, traffic_column=('jam_factor', 'speed', 'free_flow'), *, data_dir=None, output_dir=None, base_dir=None, verbose=True)

Aggregate raw traffic snapshots into per-time-period GeoPackages.

Uses OSM-based segment matching for stable segment identity.

Parameters:

Name	Type	Description	Default
city_code	str	One of 'smg', 'bdg', 'jkt'.	required
traffic_column	str	Column in the raw snapshots to aggregate (default 'jam_factor').	('jam_factor', 'speed', 'free_flow')
data_dir	path - like	Override for the raw-data folder.	None
output_dir	path - like	Override for the output folder.	None
base_dir	path - like	Project root (where osm_reference/ lives). Defaults to ".".	None
verbose	bool	Print progress messages.	True

Returns:

Type	Description
dict[str, GeoDataFrame]	Mapping from time-period name to the aggregated GeoDataFrame.

Source code in src/trafficpipeline/aggregate.py

def aggregate_city(
    city_code: str,
    traffic_column: str | list[str] | tuple[str, ...] = (
        "jam_factor", "speed", "free_flow",
    ),
    *,
    data_dir: str | Path | None = None,
    output_dir: str | Path | None = None,
    base_dir: str | Path | None = None,
    verbose: bool = True,
) -> dict[str, gpd.GeoDataFrame]:
    """Aggregate raw traffic snapshots into per-time-period GeoPackages.

    Uses OSM-based segment matching for stable segment identity.

    Parameters
    ----------
    city_code : str
        One of ``'smg'``, ``'bdg'``, ``'jkt'``.
    traffic_column : str
        Column in the raw snapshots to aggregate (default ``'jam_factor'``).
    data_dir : path-like, optional
        Override for the raw-data folder.
    output_dir : path-like, optional
        Override for the output folder.
    base_dir : path-like, optional
        Project root (where ``osm_reference/`` lives). Defaults to ``"."``.
    verbose : bool
        Print progress messages.

    Returns
    -------
    dict[str, GeoDataFrame]
        Mapping from time-period name to the aggregated GeoDataFrame.
    """
    city = CITIES[city_code]
    base = Path(base_dir) if base_dir else Path(".")

    src = Path(data_dir) if data_dir else traffic_data_path(city_code)
    dst = Path(output_dir) if output_dir else traffic_output_path(city_code)
    dst.mkdir(parents=True, exist_ok=True)

    # Normalise traffic_column → list[str] (preserves backward compat
    # for callers passing a single string).
    if isinstance(traffic_column, str):
        traffic_columns: list[str] = [traffic_column]
    else:
        traffic_columns = list(traffic_column)

    gpkg_files = sorted(glob.glob(str(src / "*.gpkg")))
    if not gpkg_files:
        raise FileNotFoundError(f"No .gpkg files in {src}")

    if verbose:
        print(f"[{city['name']}] Found {len(gpkg_files)} snapshots")
        print(f"  Range: {os.path.basename(gpkg_files[0])} → {os.path.basename(gpkg_files[-1])}")
        print(f"  Aggregating columns: {traffic_columns}")

    # Load OSM mapping and reference geometry
    if verbose:
        print(f"  Loading OSM mapping...")
    osm_ref_gdf, wkt_hash_mapping, create_geom_hash = _load_osm_mapping(city_code, base)
    if verbose:
        print(f"  OSM segments: {len(osm_ref_gdf)}, mapping entries: {len(wkt_hash_mapping)}")

    # Build WKB hash cache from first raw file
    if verbose:
        print(f"  Building geometry hash cache...")
    first_raw = gpd.read_file(gpkg_files[0])
    wkb_cache = _build_wkb_cache(first_raw, osm_ref_gdf, wkt_hash_mapping, create_geom_hash)
    if verbose:
        print(f"  Cache: {len(wkb_cache)} WKB→OSM mappings")

    # Read all snapshots with OSM ID assignment
    frames: list[pd.DataFrame] = []
    skip_reasons: dict = {}
    n_bad_timestamp = 0
    bad_timestamp_first: str | None = None
    for i, fp in enumerate(gpkg_files, 1):
        if verbose and i % 500 == 0:
            print(f"  Reading {i}/{len(gpkg_files)} …")
        df = _read_snapshot_osm(
            fp, traffic_columns, wkb_cache, osm_ref_gdf, skip_reasons=skip_reasons,
        )
        if df is None:
            continue
        ts = _extract_timestamp(fp)
        if ts is None:
            n_bad_timestamp += 1
            if bad_timestamp_first is None:
                bad_timestamp_first = os.path.basename(str(fp))
            continue
        df["timestamp"] = ts
        frames.append(df[["osm_composite_id", *traffic_columns, "timestamp"]])

    if verbose and (skip_reasons or n_bad_timestamp):
        print(f"  Skipped snapshots:")
        for reason in ("read_error", "missing_column", "no_osm_match",
                       "empty_after_dropna"):
            n = skip_reasons.get(reason, 0)
            if n:
                fname, detail = skip_reasons.get(reason + "_first", ("?", "?"))
                print(f"    {reason}: {n} (first: {os.path.basename(fname)} — {detail})")
        if n_bad_timestamp:
            print(f"    bad_filename_timestamp: {n_bad_timestamp} "
                  f"(first: {bad_timestamp_first})")

    if not frames:
        diag = [f"{r}={skip_reasons.get(r, 0)}" for r in
                ("read_error", "missing_column", "no_osm_match",
                 "empty_after_dropna")]
        if n_bad_timestamp:
            diag.append(f"bad_filename_timestamp={n_bad_timestamp}")
        # Surface the very first failure detail for fast diagnosis.
        for reason in ("missing_column", "read_error", "no_osm_match",
                       "empty_after_dropna"):
            first = skip_reasons.get(reason + "_first")
            if first:
                fname, detail = first
                raise RuntimeError(
                    f"No valid data read. Counts: {', '.join(diag)}. "
                    f"First {reason} in {os.path.basename(fname)}: {detail}"
                )
        raise RuntimeError(
            f"No valid data read. Counts: {', '.join(diag)}. "
            f"First bad-timestamp filename: {bad_timestamp_first}"
        )

    combined = pd.concat(frames, ignore_index=True)
    combined["hour"] = combined["timestamp"].dt.hour
    combined["time_period"] = combined["hour"].apply(get_time_period)

    if verbose:
        print(f"  Combined records: {len(combined):,}")
        print(f"  Date range: {combined['timestamp'].min()} → {combined['timestamp'].max()}")

    # Reference geometry for output
    osm_geom = osm_ref_gdf.set_index('osm_composite_id')[['geometry']]

    # Aggregate per time period (one combined GeoPackage per period,
    # with mean/std/count/min/max columns for every traffic column).
    results: dict[str, gpd.GeoDataFrame] = {}
    for period in sorted(combined["time_period"].unique()):
        subset = combined[combined["time_period"] == period]

        # Compute mean/std/count/min/max for each requested column.
        agg_spec = {col: ["mean", "std", "count", "min", "max"]
                    for col in traffic_columns}
        stats = (
            subset.groupby("osm_composite_id")
                  .agg(agg_spec)
                  .round(4)
                  .reset_index()
        )
        # Flatten MultiIndex columns: ('jam_factor', 'mean') → 'jam_factor_mean'
        flat_cols = ["osm_composite_id"]
        for col in traffic_columns:
            flat_cols.extend(f"{col}_{stat}" for stat in
                             ["mean", "std", "count", "min", "max"])
        stats.columns = flat_cols

        # Join with OSM geometry
        gdf = stats.merge(osm_geom, left_on='osm_composite_id', right_index=True, how='left')
        gdf = gpd.GeoDataFrame(gdf, geometry='geometry', crs=osm_ref_gdf.crs)
        gdf = gdf.dropna(subset=['geometry'])

        out_path = dst / f"{period}_{city_code}.gpkg"
        gdf.to_file(out_path, driver="GPKG")
        results[period] = gdf

        if verbose:
            primary = traffic_columns[0]
            means = stats[f"{primary}_mean"]
            extra = f" + {len(traffic_columns)-1} more cols" if len(traffic_columns) > 1 else ""
            print(
                f"  {period}: {len(subset):,} records → "
                f"{primary}_mean={means.mean():.4f}{extra}, "
                f"segments={len(gdf)}  ✓ saved"
            )

    if verbose:
        print(f"[{city['name']}] Aggregation complete — {len(results)} periods saved to {dst}/")

    return results

aggregate_all

aggregate_all(traffic_column=('jam_factor', 'speed', 'free_flow'), *, verbose=True)

Run :func:aggregate_city for every city in the config.

Returns nested dict {city_code: {period: GeoDataFrame}}.

Source code in src/trafficpipeline/aggregate.py

def aggregate_all(
    traffic_column: str | list[str] | tuple[str, ...] = (
        "jam_factor", "speed", "free_flow",
    ),
    *,
    verbose: bool = True,
) -> dict[str, dict[str, gpd.GeoDataFrame]]:
    """Run :func:`aggregate_city` for every city in the config.

    Returns nested dict ``{city_code: {period: GeoDataFrame}}``.
    """
    all_results: dict[str, dict[str, gpd.GeoDataFrame]] = {}
    for code in CITIES:
        try:
            all_results[code] = aggregate_city(code, traffic_column, verbose=verbose)
        except Exception as exc:
            print(f"[{CITIES[code]['name']}] FAILED: {exc}")
    return all_results