Akiba So
e22b004f9e
Migrate maps to @geoscene/core, polish monitoring/alerts UX, fix timeline basemap flicker and district alert regions, and ship compose/nginx Docker deploy assets with CBPOA_ROOT data mounts. Co-authored-by: Cursor <cursoragent@cursor.com>
82 lines
2.3 KiB
Python
82 lines
2.3 KiB
Python
"""Fast daily city-wide mean risk_1d with on-disk cache.
|
|
|
|
Avoids re-parsing ~45MB GeoJSON on every /analysis/trend request.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
from config import DATA_DIR, PROJECT_ROOT
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_CACHE_PATH = PROJECT_ROOT / "processed" / "daily_avg_risk.json"
|
|
|
|
|
|
def _read_disk_cache() -> dict[str, float]:
|
|
if not _CACHE_PATH.exists():
|
|
return {}
|
|
try:
|
|
raw = json.loads(_CACHE_PATH.read_text(encoding="utf-8"))
|
|
return {str(k): float(v) for k, v in raw.items()}
|
|
except (OSError, json.JSONDecodeError, TypeError, ValueError):
|
|
return {}
|
|
|
|
|
|
def _write_disk_cache(cache: dict[str, float]) -> None:
|
|
try:
|
|
_CACHE_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
_CACHE_PATH.write_text(
|
|
json.dumps(cache, ensure_ascii=False, separators=(",", ":")),
|
|
encoding="utf-8",
|
|
)
|
|
except OSError as e:
|
|
logger.warning("Failed to persist daily avg risk cache: %s", e)
|
|
|
|
|
|
def _compute_mean_risk_1d(filepath: Path) -> float:
|
|
"""Parse one risk GeoJSON and return mean risk_1d (0 if empty/missing)."""
|
|
try:
|
|
with open(filepath, "r", encoding="utf-8") as f:
|
|
geojson = json.load(f)
|
|
except (OSError, json.JSONDecodeError) as e:
|
|
logger.warning("Failed to parse %s: %s", filepath, e)
|
|
return 0.0
|
|
|
|
total = 0.0
|
|
n = 0
|
|
for feature in geojson.get("features", []):
|
|
props = feature.get("properties") or {}
|
|
r = props.get("risk_1d")
|
|
if r is None:
|
|
continue
|
|
total += float(r)
|
|
n += 1
|
|
return round(total / n, 4) if n else 0.0
|
|
|
|
|
|
@lru_cache(maxsize=64)
|
|
def daily_avg_risk(date_yyyymmdd: str) -> float:
|
|
"""Mean risk_1d for YYYYMMDD. Memory + disk cached."""
|
|
disk = _read_disk_cache()
|
|
if date_yyyymmdd in disk:
|
|
return disk[date_yyyymmdd]
|
|
|
|
filepath = DATA_DIR / f"risk_{date_yyyymmdd}.geojson"
|
|
if not filepath.exists():
|
|
return 0.0
|
|
|
|
avg = _compute_mean_risk_1d(filepath)
|
|
disk[date_yyyymmdd] = avg
|
|
_write_disk_cache(disk)
|
|
return avg
|
|
|
|
|
|
def warm_daily_avg_risk(dates: list[str]) -> None:
|
|
"""Precompute missing dates into the disk cache (blocking)."""
|
|
for d in dates:
|
|
daily_avg_risk(d)
|