denshooter
89 lines
3.6 KiB
Python
89 lines
3.6 KiB
Python
import httpx
|
|
import asyncio
|
|
|
|
# NOAA Space Weather Prediction Center - free, no key
|
|
NOAA_KP_URL = "https://services.swpc.noaa.gov/products/noaa-planetary-k-index.json"
|
|
NOAA_ALERTS_URL = "https://services.swpc.noaa.gov/products/alerts.json"
|
|
NOAA_SOLAR_URL = "https://services.swpc.noaa.gov/json/solar-cycle/observed-solar-cycle-indices.json"
|
|
|
|
KP_STATUS = {
|
|
(0, 2): ("QUIET", "No significant geomagnetic activity."),
|
|
(2, 4): ("UNSETTLED", "Minor geomagnetic activity."),
|
|
(4, 5): ("ACTIVE", "Active geomagnetic conditions."),
|
|
(5, 6): ("STORM-G1", "Minor geomagnetic storm (G1)."),
|
|
(6, 7): ("STORM-G2", "Moderate geomagnetic storm (G2). HF radio disruption possible."),
|
|
(7, 8): ("STORM-G3", "Strong geomagnetic storm (G3). Possible power grid fluctuations."),
|
|
(8, 9): ("STORM-G4", "Severe geomagnetic storm (G4). Widespread power disruption."),
|
|
(9, 10): ("STORM-G5", "Extreme geomagnetic storm (G5). Grid collapse risk."),
|
|
}
|
|
|
|
def classify_kp(kp: float) -> tuple[str, str]:
|
|
for (lo, hi), (status, desc) in KP_STATUS.items():
|
|
if lo <= kp < hi:
|
|
return status, desc
|
|
return "STORM-G5", "Extreme geomagnetic storm."
|
|
|
|
|
|
async def fetch_space_weather() -> dict:
|
|
"""
|
|
Fetches real space weather data from NOAA SWPC.
|
|
- Planetary K-index (geomagnetic storm indicator)
|
|
- Active space weather alerts
|
|
"""
|
|
result = {
|
|
"kp_index": 0.0,
|
|
"status": "UNKNOWN",
|
|
"description": "Awaiting space weather data...",
|
|
"alerts": [],
|
|
"timestamp": "",
|
|
}
|
|
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
# Fetch Kp index - returns list of [time_tag, Kp, ...]
|
|
kp_resp = await client.get(NOAA_KP_URL, timeout=10.0)
|
|
if kp_resp.status_code == 200:
|
|
kp_data = kp_resp.json()
|
|
# Skip header row [0], get latest reading from the end
|
|
if len(kp_data) > 1:
|
|
latest = kp_data[-1]
|
|
raw_kp = latest[1] if latest[1] not in ("-1", None, "") else "0"
|
|
kp = float(raw_kp)
|
|
status, description = classify_kp(kp)
|
|
result["kp_index"] = kp
|
|
result["status"] = status
|
|
result["description"] = description
|
|
result["timestamp"] = latest[0]
|
|
|
|
# Fetch active alerts
|
|
alerts_resp = await client.get(NOAA_ALERTS_URL, timeout=10.0)
|
|
if alerts_resp.status_code == 200:
|
|
alerts_raw = alerts_resp.json()
|
|
# Filter for active, non-cancelled alerts
|
|
active_alerts = []
|
|
for alert in (alerts_raw or [])[:10]:
|
|
msg = alert.get("message", "")
|
|
if "CANCEL" not in msg and "SUMMARY" not in msg:
|
|
# Extract first meaningful line as title
|
|
lines = [l.strip() for l in msg.split("\n") if l.strip()]
|
|
title = lines[0] if lines else "Space Weather Alert"
|
|
active_alerts.append({
|
|
"issue_time": alert.get("issue_datetime", ""),
|
|
"title": title[:100],
|
|
})
|
|
result["alerts"] = active_alerts[:5]
|
|
|
|
print(f"[SPACEWEATHER] Kp={result['kp_index']} — {result['status']} | {len(result['alerts'])} active alerts")
|
|
|
|
except Exception as e:
|
|
print(f"[SPACEWEATHER] Fetch error: {e}")
|
|
result["status"] = "OFFLINE"
|
|
|
|
return result
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import json
|
|
data = asyncio.run(fetch_space_weather())
|
|
print(json.dumps(data, indent=2))
|