matt/meshai
1
0
Fork 0
mirror of https://github.com/zvx-echo6/meshai.git synced 2026-05-21 23:24:44 +02:00
Code Issues Projects Releases Packages Wiki Activity

fix(swpc): handle new NOAA API format for Kp index

Browse source 
NOAA changed the planetary k-index endpoint from array of arrays
to array of objects. Updated parser to handle both formats.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
zvx-echo6 2026-05-12 11:40:57 -06:00
commit d1511a74b9
1 changed files with 17 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

35
meshai/env/swpc.py vendored
View file

@ -139,31 +139,30 @@ class SWPCAdapter:
def _parse_kp(self, data): def _parse_kp(self, data):
"""Parse noaa-planetary-k-index.json. """Parse noaa-planetary-k-index.json.
Data format: array of arrays Data format: array of objects with time_tag, Kp, a_running, station_count
First row is header: ["time_tag", "Kp", "a_running", "station_count"] Last entry is most recent.
Last row is most recent.
""" """
if not data or len(data) < 2: if not data:
return return
# Find Kp column index from header # Get last entry (most recent)
header = data[0] last_entry = data[-1]
try:
kp_idx = header.index("Kp")
except ValueError:
kp_idx = 1
# Get last row # Handle both dict format (new API) and list format (legacy)
last_row = data[-1] if isinstance(last_entry, dict):
if len(last_row) > kp_idx:
try: try:
self._status["kp_current"] = float(last_row[kp_idx]) self._status["kp_current"] = float(last_entry.get("Kp", 0))
except (ValueError, TypeError): except (ValueError, TypeError):
pass pass
self._status["kp_timestamp"] = last_entry.get("time_tag", "")
# Get timestamp elif isinstance(last_entry, list) and len(last_entry) > 1:
if len(last_row) > 0: # Legacy array format fallback
self._status["kp_timestamp"] = last_row[0] try:
self._status["kp_current"] = float(last_entry[1])
except (ValueError, TypeError):
pass
if len(last_entry) > 0:
self._status["kp_timestamp"] = last_entry[0]
def _parse_alerts(self, data): def _parse_alerts(self, data):
"""Parse alerts.json. """Parse alerts.json.