Akiba So
fc468464b2
Context: Build a spatial risk assessment system correlating air quality data with children's respiratory disease incidence across Wuhan. Approach: FastAPI backend serving PostGIS spatial queries, React frontend with Deck.gl maps, and a PyTorch SpatialTemporalGCN pipeline for multi-day (1d/3d/7d) risk prediction. Changes: - backend/ — FastAPI API with auth (JWT), alerts, risk analysis, geocoded case data, grid statistics, and report endpoints - frontend/ — React dashboard with interactive risk maps, alert monitoring, district comparison charts, and timeline player - models/ — SpatialTemporalGCN model with trained weights and ONNX export for inference - scripts/ — ETL pipeline for weather + medical data, grid generation, feature engineering, training, and daily inference - deploy/ — Docker Compose configs for backend, frontend, and MLflow - docs/ — API docs, deployment guide, user guide, and code review Impact: Enables spatial risk visualization, alert monitoring, and ML-driven health risk forecasting for environmental health teams.
57 lines
1.4 KiB
Python
57 lines
1.4 KiB
Python
"""
|
|
Risk level classification and trend calculation utilities.
|
|
"""
|
|
from typing import Literal
|
|
|
|
from config import (
|
|
RISK_HIGH,
|
|
RISK_MEDIUM_HIGH,
|
|
RISK_MEDIUM,
|
|
RISK_MEDIUM_LOW,
|
|
TREND_SLOPE_THRESHOLD,
|
|
)
|
|
|
|
|
|
def risk_value_to_level(risk_value: float) -> str:
|
|
"""Convert risk value (0-1) to risk level string."""
|
|
if risk_value >= RISK_HIGH:
|
|
return "high"
|
|
elif risk_value >= RISK_MEDIUM_HIGH:
|
|
return "medium_high"
|
|
elif risk_value >= RISK_MEDIUM:
|
|
return "medium"
|
|
elif risk_value >= RISK_MEDIUM_LOW:
|
|
return "medium_low"
|
|
else:
|
|
return "low"
|
|
|
|
|
|
def calculate_trend(values: list[float]) -> Literal["up", "down", "stable"]:
|
|
"""Calculate trend direction from a series of values using linear regression slope."""
|
|
if len(values) < 2:
|
|
return "stable"
|
|
|
|
n = len(values)
|
|
x_mean = (n - 1) / 2
|
|
y_mean = sum(values) / n
|
|
|
|
numerator = sum((i - x_mean) * (values[i] - y_mean) for i in range(n))
|
|
denominator = sum((i - x_mean) ** 2 for i in range(n))
|
|
|
|
if denominator == 0:
|
|
return "stable"
|
|
|
|
slope = numerator / denominator
|
|
|
|
if y_mean == 0:
|
|
return "stable"
|
|
|
|
relative_slope = slope / y_mean
|
|
|
|
if relative_slope > TREND_SLOPE_THRESHOLD:
|
|
return "up"
|
|
elif relative_slope < -TREND_SLOPE_THRESHOLD:
|
|
return "down"
|
|
else:
|
|
return "stable"
|