Spaces:
Sleeping
Sleeping
File size: 3,957 Bytes
93e2220 | 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 | """TransitPulse configuration — paths, thresholds, and feature flags."""
from __future__ import annotations
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Literal
@dataclass(frozen=True)
class DataScale:
"""Defines a data-generation scale tier."""
name: str
num_routes: int
num_days: int
ping_interval_sec: int
approx_rows: str # human-readable label
# Pre-defined scale tiers for benchmarking
SCALES: dict[str, DataScale] = {
"small": DataScale("small", num_routes=20, num_days=30, ping_interval_sec=30, approx_rows="~1.1M"),
"medium": DataScale("medium", num_routes=80, num_days=30, ping_interval_sec=30, approx_rows="~25M"),
"full": DataScale("full", num_routes=200, num_days=30, ping_interval_sec=30, approx_rows="~150M"),
}
@dataclass
class TransitPulseConfig:
"""Master configuration for the TransitPulse pipeline."""
# ── Paths ──────────────────────────────────────────────────────────
project_root: Path = field(default_factory=lambda: Path(__file__).resolve().parent)
@property
def data_dir(self) -> Path:
return self.project_root / "data"
@property
def pings_dir(self) -> Path:
return self.data_dir / "pings"
@property
def output_dir(self) -> Path:
return self.data_dir / "output"
# ── Mode ───────────────────────────────────────────────────────────
mode: Literal["local", "cloud"] = "local"
# ── GCP settings (cloud mode) ──────────────────────────────────────
gcp_project: str = field(default_factory=lambda: os.getenv("GCP_PROJECT", ""))
gcs_bucket: str = field(default_factory=lambda: os.getenv("GCS_BUCKET", "transitpulse-data"))
bq_dataset: str = field(default_factory=lambda: os.getenv("BQ_DATASET", "transitpulse"))
# ── Gemini ─────────────────────────────────────────────────────────
gemini_api_key: str = field(default_factory=lambda: os.getenv("GEMINI_API_KEY", ""))
gemini_model: str = "gemini-2.5-flash"
# ── Reliability scoring thresholds ─────────────────────────────────
bunching_threshold: float = 0.25 # headway < 25 % of scheduled → bunching
gap_threshold: float = 2.0 # headway > 200 % of scheduled → gap
weight_headway_cov: float = 0.40
weight_bunching: float = 0.30
weight_gap: float = 0.30
# ── Server ─────────────────────────────────────────────────────────
host: str = "0.0.0.0"
port: int = 8000
# ── Delhi bounding box (for synthetic lat/lon) ─────────────────────
delhi_lat_min: float = 28.40
delhi_lat_max: float = 28.88
delhi_lon_min: float = 76.84
delhi_lon_max: float = 77.35
# ── Synthetic generator tuning ─────────────────────────────────────
stops_per_route_min: int = 15
stops_per_route_max: int = 45
buses_per_route_min: int = 8
buses_per_route_max: int = 25
peak_hours: tuple[int, ...] = (8, 9, 17, 18)
def ensure_dirs(self) -> None:
"""Create data directories if they don't exist."""
self.pings_dir.mkdir(parents=True, exist_ok=True)
self.output_dir.mkdir(parents=True, exist_ok=True)
# Singleton
CFG = TransitPulseConfig()
|