Akiba So
e95e2f1338
Ship a new app version with broader analytics, restructured dashboards, and a server-rendered risk map. Frontend: - Add Overview, Demographic, Disease, and Environmental Health analysis pages - Add AnomalyMarkers, CalendarHeatmap, and MetricHeatmapTable components - Rebuild Alerts map onto server-rendered raster risk tiles; expand Monitoring, Trend, and District Comparison views - Extend API client, stores, and TypeScript types Backend: - Add environment router (pollutants, lag correlations) - Add risk_raster util serving XYZ 100m risk tiles - Expand cases endpoints (demographics, seasonality, diagnoses) and insights; harden auth and file-based loaders Data & tooling: - Add processed outpatient/inpatient/combined case parquet (LFS) - Add nested CLAUDE.md guides, pyrightconfig, and test updates
328 lines
15 KiB
Python
328 lines
15 KiB
Python
"""
|
|
Pydantic models for CBPOA risk assessment API
|
|
Aligned with frontend types from CBPOA/frontend/src/types/index.ts
|
|
"""
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
from typing import Optional, List, Literal
|
|
from datetime import datetime
|
|
|
|
|
|
class GridRisk(BaseModel):
|
|
"""Grid risk data for map visualization"""
|
|
grid_id: str = Field(..., description="Grid identifier")
|
|
latitude: float = Field(..., description="Latitude coordinate")
|
|
longitude: float = Field(..., description="Longitude coordinate")
|
|
risk_value: float = Field(..., description="Risk value (0-1)")
|
|
risk_level: Literal['high', 'medium_high', 'medium', 'medium_low', 'low'] = Field(..., description="Risk level classification")
|
|
|
|
|
|
class GridDetail(GridRisk):
|
|
"""Detailed grid information with environmental factors"""
|
|
region: str = Field(..., description="Administrative region")
|
|
street: str = Field(..., description="Street name")
|
|
population_density: float = Field(..., description="Population density per km²")
|
|
nearby_schools: int = Field(..., description="Number of nearby schools")
|
|
nearby_schools_distance: float = Field(..., description="Distance to nearest school (km)")
|
|
nearby_hospitals: int = Field(..., description="Number of nearby hospitals")
|
|
nearby_hospitals_distance: float = Field(..., description="Distance to nearest hospital (km)")
|
|
traffic_flow: str = Field(..., description="Traffic flow level")
|
|
green_coverage: float = Field(..., description="Green coverage percentage")
|
|
building_density: float = Field(..., description="Building density percentage")
|
|
air_quality: str = Field(..., description="Air quality description")
|
|
humidity: float = Field(..., description="Humidity percentage")
|
|
wind_speed: float = Field(..., description="Wind speed (m/s)")
|
|
temperature: float = Field(..., description="Temperature (°C)")
|
|
trend: str = Field(..., description="Risk trend")
|
|
forecast_1day: float = Field(..., description="1-day forecast risk value")
|
|
forecast_3day: float = Field(..., description="3-day forecast risk value")
|
|
forecast_7day: float = Field(..., description="7-day forecast risk value")
|
|
timestamp: str = Field(..., description="Data timestamp")
|
|
|
|
|
|
class RiskMapResponse(BaseModel):
|
|
"""Response for risk map data"""
|
|
grids: List[GridRisk] = Field(..., description="List of grid risk data")
|
|
total_count: int = Field(..., description="Total number of grids")
|
|
timestamp: str = Field(..., description="Response timestamp")
|
|
|
|
|
|
class GridDetailResponse(BaseModel):
|
|
"""Response for grid detail with history"""
|
|
grid: GridDetail = Field(..., description="Grid detail information")
|
|
history_risk: List[dict[str, str | float]] = Field(..., description="Historical risk data")
|
|
|
|
|
|
class Alert(BaseModel):
|
|
"""Health alert for high-risk area"""
|
|
alert_id: str = Field(..., description="Alert identifier")
|
|
grid_id: str = Field(..., description="Grid identifier")
|
|
region: str = Field(..., description="Administrative region")
|
|
street: str = Field(..., description="Street name")
|
|
latitude: float = Field(..., description="Latitude coordinate")
|
|
longitude: float = Field(..., description="Longitude coordinate")
|
|
risk_value: float = Field(..., description="Risk value")
|
|
risk_level: Literal['high', 'medium_high', 'medium', 'medium_low', 'low'] = Field(..., description="Risk level")
|
|
priority: Literal['P1', 'P2'] = Field(..., description="Alert priority")
|
|
reason: str = Field(..., description="Alert reason")
|
|
timestamp: str = Field(..., description="Alert timestamp")
|
|
forecast_time: str = Field(..., description="Forecast time")
|
|
|
|
|
|
class AlertResponse(BaseModel):
|
|
"""Response for alerts list"""
|
|
alerts: List[Alert] = Field(..., description="List of alerts")
|
|
total: int = Field(..., description="Total number of alerts")
|
|
timestamp: str = Field(..., description="Response timestamp")
|
|
|
|
|
|
class Stats(BaseModel):
|
|
"""Risk statistics summary"""
|
|
total_grids: int = Field(..., description="Total number of grids")
|
|
avg_risk: float = Field(..., description="Average risk value")
|
|
distribution: dict[str, int] = Field(..., description="Risk level distribution")
|
|
high_risk_count: int = Field(..., description="Count of high risk grids")
|
|
timestamp: str = Field(..., description="Stats timestamp")
|
|
|
|
|
|
class HistoryPoint(BaseModel):
|
|
"""Single point in risk history"""
|
|
date: str = Field(..., description="Date string")
|
|
risk_value: float = Field(..., description="Risk value")
|
|
|
|
|
|
class RiskHistoryResponse(BaseModel):
|
|
"""Response for risk history"""
|
|
grid_id: str = Field(..., description="Grid identifier")
|
|
history: List[HistoryPoint] = Field(..., description="Historical risk data")
|
|
|
|
|
|
ForecastDay = Literal[0, 1, 3, 7]
|
|
|
|
|
|
# ============================================================================
|
|
# Insights Models
|
|
# ============================================================================
|
|
|
|
class InsightTrendItem(BaseModel):
|
|
"""Single trend data point for insights"""
|
|
date: str = Field(..., description="Date string")
|
|
value: float = Field(..., description="Risk value")
|
|
change: float = Field(default=0, description="Change from previous day")
|
|
|
|
|
|
class InsightTrend(BaseModel):
|
|
"""Trend analysis for insights"""
|
|
period: str = Field(..., description="Time period (e.g., '7d', '30d')")
|
|
data: List[InsightTrendItem] = Field(..., description="Trend data points")
|
|
direction: Literal["up", "down", "stable"] = Field(..., description="Overall trend direction")
|
|
avg_change: float = Field(..., description="Average daily change percentage")
|
|
|
|
|
|
class InsightHotspot(BaseModel):
|
|
"""Hotspot area for insights"""
|
|
grid_id: str = Field(..., description="Grid identifier")
|
|
latitude: float = Field(..., description="Latitude coordinate")
|
|
longitude: float = Field(..., description="Longitude coordinate")
|
|
risk_value: float = Field(..., description="Current risk value")
|
|
risk_level: Literal['high', 'medium_high', 'medium', 'medium_low', 'low'] = Field(..., description="Risk level")
|
|
region: str = Field(..., description="Administrative region")
|
|
street: str = Field(..., description="Street name")
|
|
population_density: float = Field(..., description="Population density")
|
|
days_in_high_risk: int = Field(..., description="Consecutive days in high risk")
|
|
|
|
|
|
class InsightCorrelation(BaseModel):
|
|
"""Correlation factor for insights"""
|
|
factor: str = Field(..., description="Factor name (e.g., 'temperature', 'PM2.5')")
|
|
correlation: float = Field(..., description="Correlation coefficient (-1 to 1)")
|
|
significance: Literal["high", "medium", "low"] = Field(..., description="Statistical significance")
|
|
description: str = Field(..., description="Factor description")
|
|
impact: Literal["positive", "negative", "neutral"] = Field(..., description="Impact direction")
|
|
|
|
|
|
class InsightDemographic(BaseModel):
|
|
"""Demographic breakdown for insights"""
|
|
age_group: str = Field(..., description="Age group (e.g., '0-14', '15-64', '65+')")
|
|
case_count: int = Field(..., description="Number of cases")
|
|
percentage: float = Field(..., description="Percentage of total cases")
|
|
risk_ratio: float = Field(..., description="Risk ratio compared to baseline")
|
|
|
|
|
|
class InsightsResponse(BaseModel):
|
|
"""Response for comprehensive insights"""
|
|
trend: InsightTrend = Field(..., description="Risk trend analysis")
|
|
hotspots: List[InsightHotspot] = Field(..., description="Top hotspot areas")
|
|
correlations: List[InsightCorrelation] = Field(..., description="Key correlation factors")
|
|
demographics: List[InsightDemographic] = Field(..., description="Demographic breakdown")
|
|
summary: str = Field(..., description="AI-generated summary of insights")
|
|
timestamp: str = Field(..., description="Response timestamp")
|
|
|
|
|
|
# ============================================================================
|
|
# Reports Models
|
|
# ============================================================================
|
|
|
|
class DiagnosisBreakdown(BaseModel):
|
|
"""Diagnosis breakdown for reports"""
|
|
diagnosis: str = Field(..., description="Diagnosis name")
|
|
outpatient: int = Field(..., description="Outpatient count")
|
|
inpatient: int = Field(..., description="Inpatient count")
|
|
total: int = Field(..., description="Total cases")
|
|
|
|
|
|
class ReportSection(BaseModel):
|
|
"""Single section of a report"""
|
|
title: str = Field(..., description="Section title")
|
|
content: str = Field(..., description="Section content")
|
|
charts: List[str] = Field(default=[], description="Chart identifiers for this section")
|
|
|
|
|
|
class ReportMetadata(BaseModel):
|
|
"""Metadata for a report"""
|
|
report_id: str = Field(..., description="Report identifier")
|
|
title: str = Field(..., description="Report title")
|
|
type: Literal["daily", "weekly", "monthly", "custom"] = Field(..., description="Report type")
|
|
generated_at: str = Field(..., description="Generation timestamp")
|
|
period_start: str = Field(..., description="Report period start date")
|
|
period_end: str = Field(..., description="Report period end date")
|
|
author: str = Field(default="CBPOA System", description="Report author")
|
|
|
|
|
|
class ReportSummary(BaseModel):
|
|
"""Summary statistics for a report"""
|
|
total_cases: int = Field(..., description="Total cases in period")
|
|
avg_risk: float = Field(..., description="Average risk level")
|
|
peak_risk_date: str = Field(..., description="Date of peak risk")
|
|
peak_risk_value: float = Field(..., description="Peak risk value")
|
|
high_risk_areas: int = Field(..., description="Number of high risk areas")
|
|
trend_direction: Literal["improving", "stable", "worsening"] = Field(..., description="Overall trend")
|
|
|
|
|
|
class ReportRecommendation(BaseModel):
|
|
"""Recommendation from report"""
|
|
priority: Literal["high", "medium", "low"] = Field(..., description="Recommendation priority")
|
|
category: Literal["prevention", "monitoring", "intervention", "resource_allocation"] = Field(..., description="Recommendation category")
|
|
title: str = Field(..., description="Recommendation title")
|
|
description: str = Field(..., description="Detailed recommendation")
|
|
target_areas: List[str] = Field(default=[], description="Target grid IDs or regions")
|
|
|
|
|
|
class ReportResponse(BaseModel):
|
|
"""Response for full report"""
|
|
metadata: ReportMetadata = Field(..., description="Report metadata")
|
|
summary: ReportSummary = Field(..., description="Report summary")
|
|
sections: List[ReportSection] = Field(..., description="Report sections")
|
|
recommendations: List[ReportRecommendation] = Field(..., description="Recommendations")
|
|
attachments: List[str] = Field(default=[], description="Attachment file paths")
|
|
timestamp: str = Field(..., description="Response timestamp")
|
|
diagnosis_breakdown: List[DiagnosisBreakdown] = Field(default=[], description="Diagnosis breakdown data")
|
|
|
|
|
|
class ReportListResponse(BaseModel):
|
|
"""Response for list of reports"""
|
|
reports: List[ReportMetadata] = Field(..., description="List of report metadata")
|
|
total: int = Field(..., description="Total number of reports")
|
|
timestamp: str = Field(..., description="Response timestamp")
|
|
|
|
|
|
# ============================================================================
|
|
# Grid Data Models (Wave 2 - Task 9)
|
|
# ============================================================================
|
|
|
|
class GridFeature(BaseModel):
|
|
"""Single grid cell with features for model input"""
|
|
grid_id: str = Field(..., description="Grid identifier (e.g., 'r100_c200')")
|
|
latitude: float = Field(..., description="Center latitude")
|
|
longitude: float = Field(..., description="Center longitude")
|
|
dem: float = Field(..., description="Digital elevation model (meters)")
|
|
population_density: float = Field(..., description="Population density per km²")
|
|
district: Optional[str] = Field(None, description="District name")
|
|
|
|
|
|
class WeatherFeature(BaseModel):
|
|
"""Weather features for a grid cell"""
|
|
grid_id: str
|
|
AQI: float
|
|
PM25: float
|
|
PM10: float
|
|
SO2: float
|
|
NO2: float
|
|
O3: float
|
|
CO: float
|
|
|
|
|
|
class CaseFeature(BaseModel):
|
|
"""Case features for a grid cell"""
|
|
grid_id: str
|
|
outpatient_count: int = Field(default=0, description="Outpatient count")
|
|
inpatient_count: int = Field(default=0, description="Inpatient count")
|
|
total_cases: int = Field(default=0, description="Total case count")
|
|
|
|
|
|
class GridPrediction(BaseModel):
|
|
"""Prediction result for a single grid cell"""
|
|
grid_id: str
|
|
latitude: float
|
|
longitude: float
|
|
risk_1day: float = Field(..., description="1-day risk prediction (0-1)")
|
|
risk_3day: float = Field(..., description="3-day risk prediction (0-1)")
|
|
risk_7day: float = Field(..., description="7-day risk prediction (0-1)")
|
|
risk_level: Literal['high', 'medium_high', 'medium', 'medium_low', 'low']
|
|
confidence: Optional[float] = Field(None, description="Prediction confidence")
|
|
|
|
|
|
class MultiDayPredictionRequest(BaseModel):
|
|
"""Request for multi-day grid predictions"""
|
|
date: str = Field(..., description="Start date (YYYY-MM-DD)")
|
|
days: int = Field(default=7, ge=1, le=14, description="Number of days to predict")
|
|
district: Optional[str] = Field(None, description="Filter by district")
|
|
|
|
|
|
class MultiDayPredictionResponse(BaseModel):
|
|
"""Response for multi-day grid predictions"""
|
|
# `model_version` collides with Pydantic's protected `model_` namespace; opt out.
|
|
model_config = ConfigDict(protected_namespaces=())
|
|
|
|
predictions: List[GridPrediction] = Field(..., description="Grid predictions")
|
|
total_grids: int = Field(..., description="Total grids predicted")
|
|
date_range: tuple[str, str] = Field(..., description="Prediction date range")
|
|
model_version: str = Field(default="1.3.7", description="Model version")
|
|
timestamp: str = Field(..., description="Response timestamp")
|
|
partial: bool = Field(default=False, description="True if some dates failed to generate")
|
|
warnings: List[str] = Field(default_factory=list, description="Warnings from partial failures")
|
|
|
|
|
|
class HistoricalAggregationRequest(BaseModel):
|
|
"""Request for historical data aggregation"""
|
|
start_date: str = Field(..., description="Start date (YYYY-MM-DD)")
|
|
end_date: str = Field(..., description="End date (YYYY-MM-DD)")
|
|
aggregation: Literal['daily', 'weekly', 'monthly'] = Field(default='daily', description="Aggregation level")
|
|
district: Optional[str] = Field(None, description="Filter by district")
|
|
|
|
|
|
class DistrictAggregation(BaseModel):
|
|
"""Aggregated data for a district"""
|
|
district: str
|
|
date: str
|
|
total_cases: int
|
|
outpatient_count: int
|
|
inpatient_count: int
|
|
avg_AQI: float
|
|
avg_PM25: float
|
|
avg_PM10: float
|
|
|
|
|
|
class HistoricalAggregationResponse(BaseModel):
|
|
"""Response for historical data aggregation"""
|
|
aggregations: List[DistrictAggregation] = Field(..., description="Aggregated data")
|
|
total_records: int = Field(..., description="Total records")
|
|
date_range: tuple[str, str] = Field(..., description="Data date range")
|
|
timestamp: str = Field(..., description="Response timestamp")
|
|
|
|
|
|
class GridGeoJSONResponse(BaseModel):
|
|
"""Response for grid data as GeoJSON"""
|
|
type: Literal['FeatureCollection'] = 'FeatureCollection'
|
|
features: List[dict] = Field(..., description="GeoJSON features")
|
|
timestamp: str = Field(..., description="Response timestamp")
|