SafeVixAI/SafeVixAI-Dataset-Hub

1from __future__ import annotations2 3import argparse4import csv5import json6from pathlib import Path7 8 9ROOT_DIR = Path(__file__).resolve().parents[1]10DEFAULT_INPUT = ROOT_DIR / 'chatbot_service' / 'data' / 'accidents' / 'morth_2022'11DEFAULT_OUTPUT_CSV = ROOT_DIR / 'chatbot_service' / 'data' / 'accidents' / 'accident_blackspots_preview.csv'12DEFAULT_OUTPUT_GEOJSON = ROOT_DIR / 'frontend' / 'public' / 'offline-data' / 'accident-blackspots.geojson'13STATE_CENTROIDS = {14    'andhra pradesh': (15.9129, 79.74),15    'arunachal pradesh': (28.2180, 94.7278),16    'assam': (26.2006, 92.9376),17    'bihar': (25.0961, 85.3131),18    'chhattisgarh': (21.2787, 81.8661),19    'delhi': (28.7041, 77.1025),20    'goa': (15.2993, 74.1240),21    'gujarat': (22.2587, 71.1924),22    'haryana': (29.0588, 76.0856),23    'himachal pradesh': (31.1048, 77.1734),24    'jharkhand': (23.6102, 85.2799),25    'karnataka': (15.3173, 75.7139),26    'kerala': (10.8505, 76.2711),27    'madhya pradesh': (22.9734, 78.6569),28    'maharashtra': (19.7515, 75.7139),29    'manipur': (24.6637, 93.9063),30    'meghalaya': (25.4670, 91.3662),31    'mizoram': (23.1645, 92.9376),32    'nagaland': (26.1584, 94.5624),33    'odisha': (20.9517, 85.0985),34    'punjab': (31.1471, 75.3412),35    'rajasthan': (27.0238, 74.2179),36    'sikkim': (27.5330, 88.5122),37    'tamil nadu': (11.1271, 78.6569),38    'telangana': (18.1124, 79.0193),39    'tripura': (23.9408, 91.9882),40    'uttar pradesh': (26.8467, 80.9462),41    'uttarakhand': (30.0668, 79.0193),42    'west bengal': (22.9868, 87.8550),43}44STATE_FIELDS = ('state', 'state_name', 'state_ut', 'state/ut', 'state_ut_name')45CITY_FIELDS = ('city', 'city_name', 'district', 'district_name', 'location')46LAT_FIELDS = ('lat', 'latitude')47LON_FIELDS = ('lon', 'lng', 'longitude')48ACCIDENT_FIELDS = ('total_accidents', 'accidents', 'road_accidents', 'fatal_accidents')49DEATH_FIELDS = ('persons_killed', 'killed', 'deaths')50INJURY_FIELDS = ('persons_injured', 'injured')51 52 53def _first_value(row: dict[str, str], names: tuple[str, ...]) -> str:54    for name in names:55        value = (row.get(name) or '').strip()56        if value:57            return value58    return ''59 60 61def _parse_float(value: str) -> float | None:62    try:63        return float(value)64    except (TypeError, ValueError):65        return None66 67 68def _parse_int(value: str) -> int:69    try:70        return int(float(value))71    except (TypeError, ValueError):72        return 073 74 75def _discover_csvs(path: Path) -> list[Path]:76    if path.is_file():77        return [path]78    return sorted(candidate for candidate in path.rglob('*.csv') if candidate.is_file())79 80 81def _normalize_row(row: dict[str, str], *, source_file: str, index: int) -> dict | None:82    state = _first_value(row, STATE_FIELDS)83    city = _first_value(row, CITY_FIELDS)84    lat = _parse_float(_first_value(row, LAT_FIELDS))85    lon = _parse_float(_first_value(row, LON_FIELDS))86 87    if (lat is None or lon is None) and state.lower() in STATE_CENTROIDS:88        lat, lon = STATE_CENTROIDS[state.lower()]89 90    if lat is None or lon is None:91        return None92 93    accidents = _parse_int(_first_value(row, ACCIDENT_FIELDS))94    killed = _parse_int(_first_value(row, DEATH_FIELDS))95    injured = _parse_int(_first_value(row, INJURY_FIELDS))96    severity_score = accidents + (2 * killed) + injured97 98    return {99        'blackspot_id': f'{source_file}:{index}',100        'state': state,101        'city': city,102        'lat': f'{lat:.6f}',103        'lon': f'{lon:.6f}',104        'accidents': accidents,105        'killed': killed,106        'injured': injured,107        'severity_score': severity_score,108        'source_file': source_file,109    }110 111 112def _load_records(input_path: Path) -> list[dict]:113    records: list[dict] = []114    for csv_path in _discover_csvs(input_path):115        with csv_path.open('r', encoding='utf-8', newline='') as handle:116            reader = csv.DictReader(handle)117            for index, row in enumerate(reader, start=1):118                normalized = _normalize_row(row, source_file=csv_path.name, index=index)119                if normalized is not None:120                    records.append(normalized)121    return records122 123 124def _write_csv(path: Path, rows: list[dict]) -> None:125    path.parent.mkdir(parents=True, exist_ok=True)126    with path.open('w', encoding='utf-8', newline='') as handle:127        writer = csv.DictWriter(128            handle,129            fieldnames=['blackspot_id', 'state', 'city', 'lat', 'lon', 'accidents', 'killed', 'injured', 'severity_score', 'source_file'],130        )131        writer.writeheader()132        writer.writerows(rows)133 134 135def _write_geojson(path: Path, rows: list[dict]) -> None:136    path.parent.mkdir(parents=True, exist_ok=True)137    geojson = {138        'type': 'FeatureCollection',139        'features': [140            {141                'type': 'Feature',142                'geometry': {'type': 'Point', 'coordinates': [float(row['lon']), float(row['lat'])]},143                'properties': {144                    'blackspot_id': row['blackspot_id'],145                    'state': row['state'],146                    'city': row['city'],147                    'accidents': row['accidents'],148                    'killed': row['killed'],149                    'injured': row['injured'],150                    'severity_score': row['severity_score'],151                    'source_file': row['source_file'],152                },153            }154            for row in rows155        ],156    }157    path.write_text(json.dumps(geojson, indent=2), encoding='utf-8')158 159 160def main() -> None:161    parser = argparse.ArgumentParser(162        description='Normalize accident CSVs into a blackspot preview CSV and GeoJSON bundle.',163    )164    parser.add_argument('--input', type=Path, default=DEFAULT_INPUT, help=f'CSV file or directory. Defaults to {DEFAULT_INPUT}')165    parser.add_argument('--output-csv', type=Path, default=DEFAULT_OUTPUT_CSV, help=f'Normalized CSV output. Defaults to {DEFAULT_OUTPUT_CSV}')166    parser.add_argument(167        '--output-geojson',168        type=Path,169        default=DEFAULT_OUTPUT_GEOJSON,170        help=f'GeoJSON output for offline mapping. Defaults to {DEFAULT_OUTPUT_GEOJSON}',171    )172    args = parser.parse_args()173 174    if not args.input.exists():175        raise SystemExit(f'Input path not found: {args.input}')176 177    rows = _load_records(args.input)178    if not rows:179        raise SystemExit('No accident CSV rows could be normalized from the provided input.')180 181    rows.sort(key=lambda item: item['severity_score'], reverse=True)182    _write_csv(args.output_csv, rows)183    _write_geojson(args.output_geojson, rows)184    print(f'Wrote {len(rows)} normalized blackspot rows to {args.output_csv}')185    print(f'Wrote GeoJSON preview to {args.output_geojson}')186 187 188if __name__ == '__main__':189    main()190