""" GeoJSON file parsing utilities. """ import json import logging from functools import lru_cache from pathlib import Path from typing import Any from config import WUHAN_BOUNDARY_PATH from utils.risk import risk_value_to_level logger = logging.getLogger(__name__) @lru_cache(maxsize=16) def parse_geojson_file(filepath: Path) -> list[dict[str, Any]]: """Parse GeoJSON file and extract grid data with standard fields.""" try: with open(filepath, "r", encoding="utf-8") as f: geojson = json.load(f) except (json.JSONDecodeError, OSError) as e: logger.warning("Failed to parse GeoJSON file %s: %s", filepath, e) return [] grids: list[dict[str, Any]] = [] for feature in geojson.get("features", []): props = feature.get("properties", {}) coords = feature.get("geometry", {}).get("coordinates", [0, 0]) risk_1d = props.get("risk_1d", 0) grids.append({ "grid_id": str(props.get("node_id", "")), "latitude": props.get("lat", coords[1] if len(coords) > 1 else 0), "longitude": props.get("lon", coords[0] if len(coords) > 0 else 0), "risk_value": risk_1d, "risk_3d": props.get("risk_3d", 0), "risk_7d": props.get("risk_7d", 0), "risk_level": risk_value_to_level(risk_1d), }) return grids @lru_cache(maxsize=1) def load_districts() -> list[dict[str, Any]]: """Load Wuhan district boundaries from GeoJSON.""" if not WUHAN_BOUNDARY_PATH.exists(): return [] with open(WUHAN_BOUNDARY_PATH, "r", encoding="utf-8") as f: geojson = json.load(f) districts = [] for feature in geojson.get("features", []): props = feature.get("properties", {}) districts.append({ "name": props.get("name", ""), "adcode": props.get("adcode", ""), "coordinates": feature.get("geometry", {}).get("coordinates", []), }) return districts