CA/backend/utils/geojson.py

"""
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=8)
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