Centralized configuration for the traffic congestion pipeline.
All city definitions, time periods, directory conventions, and shared
constants live here so that every other module imports from one place.
Return city dict or raise KeyError with a helpful message.
Source code in src/trafficpipeline/config.py
| def get_city(code: str) -> dict:
"""Return city dict or raise ``KeyError`` with a helpful message."""
if code not in CITIES:
valid = ", ".join(sorted(CITIES))
raise KeyError(f"Unknown city code '{code}'. Choose from: {valid}")
return CITIES[code]
|
Map an hour (0-23) to its time-period name.
Source code in src/trafficpipeline/config.py
| def get_time_period(hour: int) -> str:
"""Map an hour (0-23) to its time-period name."""
for name, (start, end) in TIME_PERIOD_HOURS.items():
if start <= hour < end:
return name
raise ValueError(f"Hour {hour} does not fall in any time period")
|
traffic_data_path(city_code)
Absolute path to raw traffic snapshots for city_code.
Source code in src/trafficpipeline/config.py
| def traffic_data_path(city_code: str) -> Path:
"""Absolute path to raw traffic snapshots for *city_code*."""
return DATA_DIR / CITIES[city_code]["traffic_data_dir"]
|
traffic_output_path(city_code)
Absolute path to aggregated output for city_code.
Source code in src/trafficpipeline/config.py
| def traffic_output_path(city_code: str) -> Path:
"""Absolute path to aggregated output for *city_code*."""
return DATA_DIR / CITIES[city_code]["traffic_output_dir"]
|