372 lines
13 KiB
Python
372 lines
13 KiB
Python
|
"""Health check and system status diagnostics for PolyWeather."""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from typing import Any, Dict, Optional
|
||
|
|
|
||
|
|
from src.data_collection.country_networks import provider_coverage_summary
|
||
|
|
from src.utils.metrics import build_metrics_summary, gauge_set
|
||
|
|
|
||
|
|
|
||
|
|
def _env_bool(name: str, default: bool = False) -> bool:
|
||
|
|
raw = os.getenv(name)
|
||
|
|
if raw is None:
|
||
|
|
return default
|
||
|
|
return raw.strip().lower() in {"1", "true", "yes", "on"}
|
||
|
|
|
||
|
|
|
||
|
|
def _sqlite_health(account_db) -> Dict[str, Any]:
|
||
|
|
try:
|
||
|
|
from src.database.sqlite_connection import connect_sqlite
|
||
|
|
with connect_sqlite(account_db.db_path, timeout=0.05) as conn:
|
||
|
|
conn.execute("SELECT 1").fetchone()
|
||
|
|
return {"ok": True, "db_path": account_db.db_path}
|
||
|
|
except Exception as exc:
|
||
|
|
return {"ok": False, "db_path": account_db.db_path, "error": str(exc)}
|
||
|
|
|
||
|
|
|
||
|
|
def _cache_summary(weather_collector, analysis_cache, cache_entries: int) -> Dict[str, Any]:
|
||
|
|
from web.analysis_service import get_analysis_cache_stats
|
||
|
|
|
||
|
|
open_meteo_forecast_entries = len(getattr(weather_collector, "_open_meteo_cache", {}) or {})
|
||
|
|
open_meteo_ensemble_entries = len(getattr(weather_collector, "_ensemble_cache", {}) or {})
|
||
|
|
open_meteo_multi_model_entries = len(
|
||
|
|
getattr(weather_collector, "_multi_model_cache", {}) or {}
|
||
|
|
)
|
||
|
|
metar_entries = weather_collector.cache.store_size("metar") if hasattr(weather_collector, "cache") else 0
|
||
|
|
taf_entries = len(getattr(weather_collector, "_taf_cache", {}) or {})
|
||
|
|
settlement_entries = len(getattr(weather_collector, "_settlement_cache", {}) or {})
|
||
|
|
|
||
|
|
gauge_set("polyweather_api_cache_entries", cache_entries)
|
||
|
|
gauge_set("polyweather_open_meteo_forecast_cache_entries", open_meteo_forecast_entries)
|
||
|
|
gauge_set("polyweather_open_meteo_ensemble_cache_entries", open_meteo_ensemble_entries)
|
||
|
|
gauge_set(
|
||
|
|
"polyweather_open_meteo_multi_model_cache_entries",
|
||
|
|
open_meteo_multi_model_entries,
|
||
|
|
)
|
||
|
|
gauge_set("polyweather_metar_cache_entries", metar_entries)
|
||
|
|
gauge_set("polyweather_taf_cache_entries", taf_entries)
|
||
|
|
gauge_set("polyweather_settlement_cache_entries", settlement_entries)
|
||
|
|
return {
|
||
|
|
"api_cache_entries": cache_entries,
|
||
|
|
"open_meteo_forecast_entries": open_meteo_forecast_entries,
|
||
|
|
"open_meteo_ensemble_entries": open_meteo_ensemble_entries,
|
||
|
|
"open_meteo_multi_model_entries": open_meteo_multi_model_entries,
|
||
|
|
"metar_entries": metar_entries,
|
||
|
|
"taf_entries": taf_entries,
|
||
|
|
"settlement_entries": settlement_entries,
|
||
|
|
"analysis": get_analysis_cache_stats(),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _feature_flags_summary(config: dict) -> Dict[str, Any]:
|
||
|
|
from src.auth.supabase_entitlement import SUPABASE_ENTITLEMENT
|
||
|
|
from src.payments import PAYMENT_CHECKOUT
|
||
|
|
from src.database.runtime_state import get_state_storage_mode
|
||
|
|
|
||
|
|
_SUPABASE_AUTH_REQUIRED = _env_bool(
|
||
|
|
"POLYWEATHER_AUTH_REQUIRED",
|
||
|
|
SUPABASE_ENTITLEMENT.enabled,
|
||
|
|
)
|
||
|
|
_ENTITLEMENT_GUARD_ENABLED = _env_bool("POLYWEATHER_REQUIRE_ENTITLEMENT", False)
|
||
|
|
|
||
|
|
return {
|
||
|
|
"auth_enabled": bool(SUPABASE_ENTITLEMENT.enabled),
|
||
|
|
"auth_required": bool(_SUPABASE_AUTH_REQUIRED),
|
||
|
|
"entitlement_guard_enabled": bool(_ENTITLEMENT_GUARD_ENABLED),
|
||
|
|
"payment_enabled": bool(getattr(PAYMENT_CHECKOUT, "enabled", False)),
|
||
|
|
"state_storage_mode": get_state_storage_mode(),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _integration_summary(config: dict) -> Dict[str, Any]:
|
||
|
|
from src.auth.supabase_entitlement import SUPABASE_ENTITLEMENT
|
||
|
|
|
||
|
|
weather_cfg = config.get("weather", {}) if isinstance(config, dict) else {}
|
||
|
|
return {
|
||
|
|
"supabase_configured": bool(SUPABASE_ENTITLEMENT.configured),
|
||
|
|
"telegram_bot_configured": bool(
|
||
|
|
(config.get("telegram", {}) or {}).get("bot_token")
|
||
|
|
),
|
||
|
|
"walletconnect_configured": bool(
|
||
|
|
os.getenv("NEXT_PUBLIC_WALLETCONNECT_PROJECT_ID")
|
||
|
|
),
|
||
|
|
"weather_sources": {
|
||
|
|
"openweather": bool(weather_cfg.get("openweather_api_key")),
|
||
|
|
"wunderground": bool(weather_cfg.get("wunderground_api_key")),
|
||
|
|
"visualcrossing": bool(weather_cfg.get("visualcrossing_api_key")),
|
||
|
|
},
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _probability_summary() -> Dict[str, Any]:
|
||
|
|
return {
|
||
|
|
"engine_mode": "legacy",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _read_json_file(path: str) -> Optional[Dict[str, Any]]:
|
||
|
|
if not os.path.exists(path):
|
||
|
|
return None
|
||
|
|
try:
|
||
|
|
with open(path, "r", encoding="utf-8") as fh:
|
||
|
|
payload = json.load(fh)
|
||
|
|
if isinstance(payload, dict):
|
||
|
|
return payload
|
||
|
|
except Exception:
|
||
|
|
return None
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def _table_date_summary(conn, table_name: str) -> Dict[str, Any]:
|
||
|
|
try:
|
||
|
|
row = conn.execute(
|
||
|
|
f"""
|
||
|
|
SELECT COUNT(*) AS row_count,
|
||
|
|
COUNT(DISTINCT city) AS cities_count,
|
||
|
|
MIN(target_date) AS min_date,
|
||
|
|
MAX(target_date) AS max_date
|
||
|
|
FROM {table_name}
|
||
|
|
"""
|
||
|
|
).fetchone()
|
||
|
|
except Exception as exc:
|
||
|
|
return {"ok": False, "error": str(exc), "row_count": 0, "cities_count": 0}
|
||
|
|
|
||
|
|
return {
|
||
|
|
"ok": True,
|
||
|
|
"row_count": int(row["row_count"] or 0),
|
||
|
|
"cities_count": int(row["cities_count"] or 0),
|
||
|
|
"min_date": row["min_date"],
|
||
|
|
"max_date": row["max_date"],
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _truth_source_counts(conn) -> Dict[str, int]:
|
||
|
|
try:
|
||
|
|
rows = conn.execute(
|
||
|
|
"""
|
||
|
|
SELECT COALESCE(NULLIF(TRIM(settlement_source), ''), 'unknown') AS settlement_source,
|
||
|
|
COUNT(*) AS row_count
|
||
|
|
FROM truth_records_store
|
||
|
|
GROUP BY COALESCE(NULLIF(TRIM(settlement_source), ''), 'unknown')
|
||
|
|
ORDER BY row_count DESC, settlement_source ASC
|
||
|
|
"""
|
||
|
|
).fetchall()
|
||
|
|
except Exception:
|
||
|
|
return {}
|
||
|
|
return {str(row["settlement_source"]): int(row["row_count"] or 0) for row in rows}
|
||
|
|
|
||
|
|
|
||
|
|
def _truth_revisions_summary(conn) -> Dict[str, Any]:
|
||
|
|
try:
|
||
|
|
row = conn.execute(
|
||
|
|
"""
|
||
|
|
SELECT COUNT(*) AS row_count,
|
||
|
|
MAX(updated_at) AS last_updated_at
|
||
|
|
FROM truth_revisions_store
|
||
|
|
"""
|
||
|
|
).fetchone()
|
||
|
|
except Exception as exc:
|
||
|
|
return {"ok": False, "error": str(exc), "row_count": 0}
|
||
|
|
return {
|
||
|
|
"ok": True,
|
||
|
|
"row_count": int(row["row_count"] or 0),
|
||
|
|
"last_updated_at": row["last_updated_at"],
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _city_coverage_summary(conn, city_registry) -> Dict[str, Any]:
|
||
|
|
truth_rows = conn.execute(
|
||
|
|
"""
|
||
|
|
SELECT city, COUNT(*) AS row_count, MIN(target_date) AS min_date, MAX(target_date) AS max_date
|
||
|
|
FROM truth_records_store
|
||
|
|
GROUP BY city
|
||
|
|
"""
|
||
|
|
).fetchall()
|
||
|
|
feature_rows = conn.execute(
|
||
|
|
"""
|
||
|
|
SELECT city, COUNT(*) AS row_count, MIN(target_date) AS min_date, MAX(target_date) AS max_date
|
||
|
|
FROM training_feature_records_store
|
||
|
|
GROUP BY city
|
||
|
|
"""
|
||
|
|
).fetchall()
|
||
|
|
|
||
|
|
truth_index = {
|
||
|
|
str(row["city"]): {
|
||
|
|
"truth_rows": int(row["row_count"] or 0),
|
||
|
|
"truth_min_date": row["min_date"],
|
||
|
|
"truth_max_date": row["max_date"],
|
||
|
|
}
|
||
|
|
for row in truth_rows
|
||
|
|
}
|
||
|
|
feature_index = {
|
||
|
|
str(row["city"]): {
|
||
|
|
"feature_rows": int(row["row_count"] or 0),
|
||
|
|
"feature_min_date": row["min_date"],
|
||
|
|
"feature_max_date": row["max_date"],
|
||
|
|
}
|
||
|
|
for row in feature_rows
|
||
|
|
}
|
||
|
|
|
||
|
|
entries = []
|
||
|
|
for city, meta in city_registry.items():
|
||
|
|
truth_payload = truth_index.get(city, {})
|
||
|
|
feature_payload = feature_index.get(city, {})
|
||
|
|
entries.append(
|
||
|
|
{
|
||
|
|
"city": city,
|
||
|
|
"name": str(meta.get("name") or city),
|
||
|
|
"settlement_source": str(meta.get("settlement_source") or "metar"),
|
||
|
|
"settlement_station_code": str(
|
||
|
|
meta.get("settlement_station_code") or meta.get("icao") or ""
|
||
|
|
),
|
||
|
|
"truth_rows": int(truth_payload.get("truth_rows") or 0),
|
||
|
|
"feature_rows": int(feature_payload.get("feature_rows") or 0),
|
||
|
|
"truth_min_date": truth_payload.get("truth_min_date"),
|
||
|
|
"truth_max_date": truth_payload.get("truth_max_date"),
|
||
|
|
"feature_min_date": feature_payload.get("feature_min_date"),
|
||
|
|
"feature_max_date": feature_payload.get("feature_max_date"),
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
highlighted = [
|
||
|
|
entry for entry in entries if entry["city"] in {"taipei", "shenzhen"}
|
||
|
|
]
|
||
|
|
gaps = sorted(
|
||
|
|
entries,
|
||
|
|
key=lambda entry: (
|
||
|
|
entry["feature_rows"] > 0,
|
||
|
|
entry["truth_rows"] > 0,
|
||
|
|
entry["truth_rows"],
|
||
|
|
entry["feature_rows"],
|
||
|
|
entry["city"],
|
||
|
|
),
|
||
|
|
)[:10]
|
||
|
|
return {
|
||
|
|
"total_cities": len(entries),
|
||
|
|
"with_truth_rows": sum(1 for entry in entries if entry["truth_rows"] > 0),
|
||
|
|
"with_feature_rows": sum(1 for entry in entries if entry["feature_rows"] > 0),
|
||
|
|
"entries": entries,
|
||
|
|
"highlighted": highlighted,
|
||
|
|
"top_gaps": gaps,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _model_city_coverage_summary(city_entries) -> Dict[str, Any]:
|
||
|
|
rows = []
|
||
|
|
for entry in city_entries or []:
|
||
|
|
city = str(entry.get("city") or "").strip().lower()
|
||
|
|
rows.append(
|
||
|
|
{
|
||
|
|
"city": city,
|
||
|
|
"name": entry.get("name") or city,
|
||
|
|
"settlement_source": entry.get("settlement_source"),
|
||
|
|
"truth_rows": int(entry.get("truth_rows") or 0),
|
||
|
|
"feature_rows": int(entry.get("feature_rows") or 0),
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
weakest = sorted(
|
||
|
|
rows,
|
||
|
|
key=lambda row: (
|
||
|
|
row["truth_rows"] > 0,
|
||
|
|
row["truth_rows"],
|
||
|
|
row["city"],
|
||
|
|
),
|
||
|
|
)[:12]
|
||
|
|
strongest = sorted(
|
||
|
|
rows,
|
||
|
|
key=lambda row: (
|
||
|
|
-row["truth_rows"],
|
||
|
|
row["city"],
|
||
|
|
),
|
||
|
|
)[:8]
|
||
|
|
return {
|
||
|
|
"weakest": weakest,
|
||
|
|
"strongest": strongest,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _training_data_summary(account_db, city_registry) -> Dict[str, Any]:
|
||
|
|
from src.database.sqlite_connection import connect_sqlite
|
||
|
|
import sqlite3
|
||
|
|
|
||
|
|
db_path = account_db.db_path
|
||
|
|
truth_records = {"ok": False, "row_count": 0, "cities_count": 0}
|
||
|
|
truth_revisions = {"ok": False, "row_count": 0}
|
||
|
|
training_features = {"ok": False, "row_count": 0, "cities_count": 0}
|
||
|
|
try:
|
||
|
|
with connect_sqlite(db_path, row_factory=sqlite3.Row) as conn:
|
||
|
|
truth_records = _table_date_summary(conn, "truth_records_store")
|
||
|
|
if truth_records.get("ok"):
|
||
|
|
truth_records["source_counts"] = _truth_source_counts(conn)
|
||
|
|
truth_revisions = _truth_revisions_summary(conn)
|
||
|
|
training_features = _table_date_summary(
|
||
|
|
conn, "training_feature_records_store"
|
||
|
|
)
|
||
|
|
city_coverage = _city_coverage_summary(conn, city_registry)
|
||
|
|
except Exception as exc:
|
||
|
|
return {
|
||
|
|
"db_path": db_path,
|
||
|
|
"db_ok": False,
|
||
|
|
"error": str(exc),
|
||
|
|
"truth_records": truth_records,
|
||
|
|
"truth_revisions": truth_revisions,
|
||
|
|
"training_features": training_features,
|
||
|
|
"city_coverage": {},
|
||
|
|
"model_city_coverage": {},
|
||
|
|
}
|
||
|
|
|
||
|
|
return {
|
||
|
|
"db_path": db_path,
|
||
|
|
"db_ok": True,
|
||
|
|
"truth_records": truth_records,
|
||
|
|
"truth_revisions": truth_revisions,
|
||
|
|
"training_features": training_features,
|
||
|
|
"city_coverage": city_coverage,
|
||
|
|
"model_city_coverage": _model_city_coverage_summary(
|
||
|
|
city_coverage.get("entries") or [],
|
||
|
|
),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def build_health_payload(account_db, cities_count: int) -> Dict[str, Any]:
|
||
|
|
from src.database.runtime_state import get_state_storage_mode
|
||
|
|
|
||
|
|
db = _sqlite_health(account_db)
|
||
|
|
return {
|
||
|
|
"status": "ok" if db.get("ok") else "degraded",
|
||
|
|
"time_utc": datetime.now(timezone.utc).isoformat(),
|
||
|
|
"db": db,
|
||
|
|
"state_storage_mode": get_state_storage_mode(),
|
||
|
|
"cities_count": cities_count,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def build_system_status_payload(
|
||
|
|
account_db,
|
||
|
|
config: dict,
|
||
|
|
weather_collector,
|
||
|
|
analysis_cache,
|
||
|
|
cache_entries: int,
|
||
|
|
cities_count: int,
|
||
|
|
city_registry,
|
||
|
|
) -> Dict[str, Any]:
|
||
|
|
from src.database.runtime_state import get_state_storage_mode
|
||
|
|
|
||
|
|
return {
|
||
|
|
"status": build_health_payload(account_db, cities_count)["status"],
|
||
|
|
"time_utc": datetime.now(timezone.utc).isoformat(),
|
||
|
|
"state_storage_mode": get_state_storage_mode(),
|
||
|
|
"db": _sqlite_health(account_db),
|
||
|
|
"features": _feature_flags_summary(config),
|
||
|
|
"integrations": _integration_summary(config),
|
||
|
|
"cache": _cache_summary(weather_collector, analysis_cache, cache_entries),
|
||
|
|
"metrics": build_metrics_summary(),
|
||
|
|
"probability": _probability_summary(),
|
||
|
|
"training_data": _training_data_summary(account_db, city_registry),
|
||
|
|
"station_networks": provider_coverage_summary(),
|
||
|
|
"cities_count": cities_count,
|
||
|
|
}
|