swx-advisories

A Python package for parsing and analyzing space weather advisories from ICAO and NOAA/SWPC sources.

Features

• ICAO Advisory Parsing: Parse ICAO space weather advisories (fnxx01/fnxx02 files) with support for:

  • Blank-line and ZCZC/NNNN delimited formats
  • Latitude bands (HNH, MNH, EQN, EQS, MSH, HSH) and polygon coordinates
  • Multi-region parsing: OBS/FCST fields with multiple severity-region pairs (e.g., MOD north + SEV south)
  • Non-contiguous latitude band splitting (e.g., HNH + HSH becomes two separate regions)
  • Event chain building via NR RPLC linking
  • HF COM, GNSS, and RADIATION effect types with MOD/SEV severity
  • Data validation to detect date/time errors and inconsistencies

• NOAA Alert Parsing: Parse NOAA Space Weather Prediction Center alerts with:

  • Live API fetching from services.swpc.noaa.gov
  • G/S/R scale extraction (geomagnetic, solar radiation, radio blackout)
  • Message type classification (ALERT, WARNING, WATCH, SUMMARY)
  • K-index, X-ray class, and proton flux metrics

• DataFrame Adapters: Convert advisories to pandas DataFrames (one row per advisory/event)

• GeoJSON Output: Convert advisory regions to GeoJSON FeatureCollections for map visualization

Installation

pip install swx-advisories

Or install from source:

git clone https://gitlab.com/KNMI-OSS/spaceweather/swx-advisories.git
cd swx-advisories
pip install -e .

Quick Start

ICAO Advisories

from swx_advisories import ICAOFetcher

# Load advisories from a file or directory
fetcher = ICAOFetcher("/path/to/advisories/")
result = fetcher.fetch()

# Access advisories and events directly from the result
for advisory in result.advisories:
    print(f"{advisory.advisory_id}: {advisory.effect}")

for event in result.events:
    print(f"{event.event_id}: {event.effect} ({event.peak_severity})")
    print(f"  Duration: {event.duration}")
    print(f"  Advisories: {event.num_advisories}")

# Convert to DataFrames — three tiers of detail
df_advisories = result.to_dataframe()                  # tier 1: one row per advisory
df_events = result.to_dataframe(detail="event")        # tier 2: one row per event
df_regions = result.to_dataframe(detail="region")      # tier 3: one row per region/timestep

# Convert to GeoJSON for map visualization
geojson = result.to_geojson()

NOAA Alerts

from swx_advisories import NOAAFetcher

# Fetch from live NOAA API
fetcher = NOAAFetcher()
result = fetcher.fetch()

for alert in result:
    print(f"{alert.advisory_id}: {alert.event_type.value}")
    print(f"  Scale: {alert.noaa_scale}")
    print(f"  Severity: {alert.severity}")

# Convert to DataFrame
df = result.to_dataframe()

ICAO Advisories

ICAO space weather advisories are issued by designated Space Weather Centers (SWXCs) to warn aviation of space weather impacts on HF communications and GNSS navigation.

Fetching Advisories

from swx_advisories import ICAOFetcher

# From a single file
fetcher = ICAOFetcher("/path/to/fnxx01_advisory.txt")
result = fetcher.fetch()

# From a directory (searches for fnxx01*.txt and fnxx02*.txt)
fetcher = ICAOFetcher("/path/to/advisory_directory/")
result = fetcher.fetch()

# Access advisories
for advisory in result:
    print(f"{advisory.advisory_id}: {advisory.effect} - {advisory.severity}")

# Check for errors
if not result.success:
    print(f"Errors: {result.errors}")

Building Event Chains

Advisories are linked via the NR RPLC field. The fetcher automatically builds these into event chains:

from swx_advisories import ICAOFetcher

fetcher = ICAOFetcher("/path/to/advisories/")
result = fetcher.fetch()

# Events are automatically built and available on result.events
for event in result.events:
    print(f"\nEvent: {event.event_id}")
    print(f"  Effect: {event.effect}")
    print(f"  Peak severity: {event.peak_severity}")
    print(f"  Start: {event.issue_start}")
    print(f"  End: {event.issue_end}")
    print(f"  Duration: {event.duration}")
    print(f"  Centers: {', '.join(event.centers)}")
    print(f"  Advisories in chain: {event.num_advisories}")

# Skip event building if not needed
result = fetcher.fetch(build_events=False)
assert result.events is None

Filtering by Date Range

from datetime import datetime, timezone
from swx_advisories import ICAOFetcher

fetcher = ICAOFetcher("/path/to/advisories/")

# Filter by date range
since = datetime(2024, 1, 1, tzinfo=timezone.utc)
until = datetime(2024, 6, 30, tzinfo=timezone.utc)
result = fetcher.fetch(since=since, until=until)

# Only advisories with issue_time in range are returned
print(f"Found {len(result.advisories)} advisories in date range")

Advisory Structure

Each ICAOAdvisory contains:

advisory.advisory_id      # e.g., "2026/0042"
advisory.center           # e.g., "PECASUS", "ACFJ"
advisory.effect           # "HF COM", "GNSS", or "RADIATION"
advisory.severity         # "MOD", "SEV", or None (max across regions)
advisory.issue_time       # datetime
advisory.observation      # TemporalObservation with time, location, and regions
advisory.forecasts        # List of 4 Forecast objects (+6h, +12h, +18h, +24h)
advisory.replaces_id      # ID of advisory this replaces, or None if opening
advisory.is_opening       # True if this starts a new event chain
advisory.remark           # Free-text remarks

# Multi-region data (observation and each forecast)
advisory.observation.regions   # list[SeverityRegion] - per-region detail
advisory.observation.location  # GeographicRegion - bounding box (backward compat)
advisory.forecasts[0].regions  # list[SeverityRegion] - per-region detail

Multi-Region Parsing

Since late November 2025, ICAO advisories can contain multiple geographic regions per time step, each with its own severity level. For example:

OBS SWX: 19/1118Z MOD N50 W180...N90 W180 SEV S55 W180...S90 E180

This encodes two regions: northern polar cap at MOD and southern polar cap at SEV. The parser splits these into individual SeverityRegion objects, each with its own severity and GeographicRegion:

from swx_advisories import ICAOFetcher

result = ICAOFetcher("/path/to/advisories/").fetch()

for adv in result.advisories:
    for region in adv.observation.regions:
        print(f"  {region.severity}: lat {region.location.lat_min} to {region.location.lat_max}")

Non-contiguous latitude bands (e.g., SEV HNH HSH where mid-latitudes are unaffected) are automatically split into separate regions.

Backward compatibility is maintained: advisory.severity returns the max severity across all regions, and advisory.observation.location returns the bounding box of all regions.

Data Validation

ICAO advisories from the early operational period (2019-2021) sometimes contain data entry errors such as switched dates or typos. The validation system automatically detects these issues:

from swx_advisories import ICAOFetcher, ValidationIssue

fetcher = ICAOFetcher("/path/to/advisories/")
result = fetcher.fetch()  # Validation runs automatically

# Filter to only clean data
clean_advisories = [a for a in result.advisories if a.is_valid]
clean_events = [e for e in result.events if e.is_valid]

print(f"Valid: {len(clean_advisories)}/{len(result.advisories)} advisories")

# Inspect validation issues on a specific advisory
for adv in result.advisories:
    if not adv.is_valid:
        print(f"{adv.advisory_id}: {adv.validation_issues}")
        for msg in adv.validation_messages:
            print(f"  - {msg}")

Validation issues detected:

Issue	Level	Description
NEGATIVE_LATENCY	Advisory	Observation time is after issue time
EXCESSIVE_LATENCY	Advisory	Observation time is >6h before issue time
FORECAST_TIME_MISMATCH	Advisory	Forecast time deviates >2h from expected
ORPHANED_ADVISORY	Advisory	References non-existent parent advisory
DUPLICATE_ADVISORY_ID	Advisory	Same ID used multiple times for same effect
OBSERVATION_TIME_REVERSAL	Event	Observation times go backwards in chain
ISSUE_TIME_REVERSAL	Event	Issue times go backwards in chain
EVENT_EXCESSIVE_DURATION	Event	Event duration exceeds 7 days

Customizing validation:

from swx_advisories import ICAOFetcher
from swx_advisories.validators import ICAOValidator, validate_advisory

# Disable automatic validation
result = fetcher.fetch(validate=False)

# Run validation manually with custom thresholds
validator = ICAOValidator(
    max_latency_hours=4.0,        # Stricter latency threshold
    forecast_tolerance_hours=1.0,  # Stricter forecast tolerance
    max_event_duration_days=5,     # Stricter duration threshold
)
for event in result.events:
    validator.validate_event(event)

# Check validation summary
print(validator.summary())

Parsing Raw Text

from swx_advisories import ICAOFetcher

text = """FNXX01 EFKL 201549
SWX ADVISORY
DTG:              20260120/1549Z
SWXC:             PECASUS
SWX EFFECT:       GNSS
ADVISORY NR:      2026/99
OBS SWX:          20/1542Z SEV N55 E055 - N85 E055 - N85 E005 - N55 E005 - N55 E055
FCST SWX +6 HR:   20/2200Z NOT AVBL
FCST SWX +12 HR:  21/0400Z NOT AVBL
FCST SWX +18 HR:  21/1000Z NOT AVBL
FCST SWX +24 HR:  21/1600Z NOT AVBL
RMK:              SPACE WEATHER EVENT IN PROGRESS
NXT ADVISORY:     NO FURTHER ADVISORIES=
"""

result = ICAOFetcher.from_text(text)
for adv in result:
    print(f"{adv.advisory_id}: {adv.effect}")

NOAA Alerts

NOAA Space Weather Prediction Center (SWPC) issues alerts, warnings, and watches for space weather events using the G/S/R scale system.

Fetching from API

from swx_advisories import NOAAFetcher

# Fetch all recent alerts from NOAA API
fetcher = NOAAFetcher()
result = fetcher.fetch()

print(f"Fetched {result.count} alerts")
for alert in result:
    print(f"{alert.advisory_id}: {alert.event_type.value} ({alert.noaa_scale})")

Fetching from Local File

from swx_advisories import NOAAFetcher

# Load from a local JSON file (same format as NOAA API)
fetcher = NOAAFetcher(from_file="/path/to/alerts.json")
result = fetcher.fetch()

Filtering Alerts

from datetime import datetime, timezone
from swx_advisories import NOAAFetcher, EventType, MessageType

fetcher = NOAAFetcher()

# Filter by event type
geomag = fetcher.fetch_by_type(EventType.GEOMAGNETIC_STORM)
radio = fetcher.fetch_by_type(EventType.RADIO_BLACKOUT)

# Filter by message type
alerts = fetcher.fetch_active_alerts()    # ALERT messages only
warnings = fetcher.fetch_warnings()        # WARNING messages only
watches = fetcher.fetch_watches()          # WATCH messages only

# Filter by date range
since = datetime(2024, 1, 1, tzinfo=timezone.utc)
until = datetime(2024, 12, 31, tzinfo=timezone.utc)
result = fetcher.fetch(since=since, until=until)

Alert Structure

Each SpaceWeatherAdvisory contains:

alert.advisory_id        # e.g., "ALTK06-690"
alert.event_type         # EventType enum (GEOMAGNETIC_STORM, RADIO_BLACKOUT, etc.)
alert.message_type       # MessageType enum (ALERT, WARNING, WATCH, SUMMARY)
alert.severity           # Severity enum (MINOR, MODERATE, SEVERE, EXTREME)
alert.noaa_scale         # e.g., "G2", "S1", "R3"
alert.issue_time         # datetime
alert.begin_time         # datetime (if available)
alert.end_time           # datetime (if available)
alert.short_description  # Brief description of the event
alert.potential_impacts  # Description of potential impacts

NOAA Scales

The package extracts NOAA space weather scales:

Scale	Type	Levels
G	Geomagnetic Storm	G1 (Minor) to G5 (Extreme)
S	Solar Radiation Storm	S1 (Minor) to S5 (Extreme)
R	Radio Blackout	R1 (Minor) to R5 (Extreme)

DataFrame Adapters

Convert advisories and events to pandas DataFrames for analysis. The recommended approach is to call to_dataframe() directly on the FetchResult:

from swx_advisories import ICAOFetcher, NOAAFetcher

# ICAO — three tiers of detail
result = ICAOFetcher("/path/to/advisories/").fetch()
df_advisories = result.to_dataframe()                  # tier 1: one row per advisory
df_events = result.to_dataframe(detail="event")        # tier 2: one row per event
df_regions = result.to_dataframe(detail="region")      # tier 3: one row per region/timestep

# NOAA — tier 1 (per-alert)
result = NOAAFetcher().fetch()
df = result.to_dataframe()

The free function to_dataframe() is also available for direct use with model objects:

from swx_advisories import to_dataframe

df_advisories = to_dataframe(result.advisories)            # tier 1 (auto-detected)
df_events = to_dataframe(result.events)                    # tier 2 (auto-detected)
df_regions = to_dataframe(result.advisories, detail="region")  # tier 3 (explicit)

ICAO DataFrames

from swx_advisories import icao_advisories_to_dataframe, icao_events_to_dataframe

# Advisories DataFrame (one row per advisory)
df = icao_advisories_to_dataframe(advisories)
# Columns: advisory_id, center, effect, severity, issue_time, obs_time,
#          lat_min, lat_max, lon_min, lon_max, replaces_id, ...,
#          is_valid, validation_issues

# Events DataFrame (one row per event)
df = icao_events_to_dataframe(events)
# Columns: event_id, effect, peak_severity, num_advisories, duration_hours,
#          issue_start, issue_end, centers, ..., is_valid, validation_issues

# Filter to valid data
clean_df = df[df["is_valid"]]

Regions DataFrame

A flattened DataFrame with one row per region per time step, designed for map visualization and time-series analysis:

from swx_advisories import ICAOFetcher, icao_regions_to_dataframe

result = ICAOFetcher("/path/to/advisories/").fetch()
df = icao_regions_to_dataframe(result.advisories)

# Each row has start_time/end_time defining the validity interval
print(df[["advisory_id", "severity", "region_index",
          "lat_min", "lat_max", "start_time", "end_time"]])

Key features:

• One row per affected region per time step (obs, +6h, +12h, +18h, +24h)
• start_time / end_time define each region's validity interval
• Per-region severity (MOD/SEV), not the advisory-level max
• geojson column with GeoJSON dict for each region (None for dayside/no-impact)
• Time steps with NO SWX EXP are excluded (absence = no impact)

Column	Description
advisory_id	Advisory identifier
center, center_country	Issuing center
effect	HF COM, GNSS, RADIATION
time_type	"obs" or "fcst"
start_time, end_time	Validity interval
hours_ahead	0 for obs, 6/12/18/24 for fcst
severity	Per-region severity (MOD/SEV)
region_index	0-based index within the time step
lat_min, lat_max, lon_min, lon_max	Region bounds
bands	Comma-joined latitude band codes
is_daylight_side	True for DAYSIDE/DAYLIGHT SIDE regions
above_fl	Flight level if specified
geojson	GeoJSON Feature dict for this region

NOAA DataFrames

from swx_advisories import noaa_alerts_to_dataframe

df = noaa_alerts_to_dataframe(alerts)
# Columns: advisory_id, event_type, message_type, severity, noaa_scale,
#          issue_time, begin_time, end_time, short_description, ...

GeoJSON Output

Convert advisory regions to GeoJSON FeatureCollections for map visualization. Each region becomes a GeoJSON Feature with [lon, lat] coordinates per RFC 7946.

from swx_advisories import ICAOFetcher

result = ICAOFetcher("/path/to/advisories/").fetch()

# All advisories -> single FeatureCollection
geojson = result.to_geojson()

# Each Feature includes properties:
# advisory_id, center, effect, severity, time_type, time, hours_ahead, region_index

The free functions advisory_to_geojson() and advisories_to_geojson() are also available for direct use:

from swx_advisories import advisory_to_geojson, advisories_to_geojson

geojson = advisory_to_geojson(result.advisories[0])    # single advisory
geojson = advisories_to_geojson(result.advisories)     # multiple advisories

Regions that are daylight-side, no-impact, or undefined produce no Feature. For DAYSIDE regions the is_daylight_side property is set to True on the GeographicRegion, but no polygon geometry is generated since the daylit hemisphere depends on the observation time.

Individual regions also support GeoJSON conversion directly:

region = advisory.observation.regions[0]
feature = region.location.to_geojson()  # GeoJSON Feature dict, or None

Data Models

Enums

from swx_advisories import EventType, MessageType, Severity, Source, ValidationIssue

# Event types
EventType.HF_BLACKOUT           # ICAO HF COM
EventType.GNSS_DEGRADATION      # ICAO GNSS
EventType.GEOMAGNETIC_STORM     # NOAA G-scale
EventType.SOLAR_RADIATION_STORM # NOAA S-scale
EventType.RADIO_BLACKOUT        # NOAA R-scale
EventType.SOLAR_FLARE
EventType.CME

# Message types (NOAA)
MessageType.ALERT
MessageType.WARNING
MessageType.WATCH
MessageType.SUMMARY

# Severity levels
Severity.MINOR
Severity.MODERATE
Severity.SEVERE
Severity.EXTREME

# Validation issues (ICAO)
ValidationIssue.NEGATIVE_LATENCY
ValidationIssue.EXCESSIVE_LATENCY
ValidationIssue.FORECAST_TIME_MISMATCH
ValidationIssue.ORPHANED_ADVISORY
ValidationIssue.DUPLICATE_ADVISORY_ID
ValidationIssue.OBSERVATION_TIME_REVERSAL
ValidationIssue.ISSUE_TIME_REVERSAL
ValidationIssue.EVENT_EXCESSIVE_DURATION

Location Models

from swx_advisories import GeographicRegion, LatitudeBand, SeverityRegion

# Latitude bands used in ICAO advisories
LatitudeBand.HNH  # High Northern Hemisphere (60-90N)
LatitudeBand.MNH  # Middle Northern Hemisphere (30-60N)
LatitudeBand.EQN  # Equatorial Northern (0-30N)
LatitudeBand.EQS  # Equatorial Southern (0-30S)
LatitudeBand.MSH  # Middle Southern Hemisphere (30-60S)
LatitudeBand.HSH  # High Southern Hemisphere (60-90S)

# Geographic region with bounds
region = GeographicRegion(
    lat_min=-90, lat_max=90,
    lon_min=-180, lon_max=180,
    text="Global coverage"
)

# Convert to GeoJSON Feature (RFC 7946, [lon, lat] order)
feature = region.to_geojson()  # dict or None for daylight-side/no-impact

# Severity region pairs a severity level with a geographic region
sev_region = SeverityRegion(severity="MOD", location=region)

API Reference

Fetchers

Class	Description
ICAOFetcher(path)	Load ICAO advisories from file or directory
NOAAFetcher(from_file=None)	Fetch NOAA alerts from API or local file
NOAAArchiveFetcher	Fetch NOAA alerts from HTML/FTP archives
FetchResult	Container with advisories, metadata, errors, and output methods

Key fetch() parameters:

• build_events (bool): Build event chains (default: True)
• validate (bool): Run data validation (default: True, ICAO only)
• since (datetime): Filter by minimum issue_time
• until (datetime): Filter by maximum issue_time

Parsers

Class	Description
ICAOTextParser	Parse raw ICAO advisory text
NOAAParser	Parse NOAA JSON alert data

Validators

Class/Function	Description
ValidationIssue	Enum of validation issue types
ICAOValidator	Batch validator with configurable thresholds
validate_advisory(advisory)	Validate single advisory
validate_event(event)	Validate event and its advisories

FetchResult Methods

Method	Description
result.to_dataframe()	Tier 1: one row per advisory/alert
result.to_dataframe(detail="event")	Tier 2: one row per event chain (ICAO only)
result.to_dataframe(detail="region")	Tier 3: one row per region per timestep (ICAO only)
result.to_geojson()	GeoJSON FeatureCollection (ICAO only)

Adapters (free functions)

Function	Description
to_dataframe(data, detail=None)	Auto-detect and convert to DataFrame
icao_advisories_to_dataframe(advisories)	ICAO advisories to DataFrame (one row per advisory)
icao_events_to_dataframe(events)	ICAO events to DataFrame (one row per event)
icao_regions_to_dataframe(advisories)	ICAO regions to DataFrame (one row per region per time step)
noaa_alerts_to_dataframe(alerts)	NOAA alerts to DataFrame
advisory_to_geojson(advisory)	Single advisory to GeoJSON FeatureCollection
advisories_to_geojson(advisories)	Multiple advisories to GeoJSON FeatureCollection

Note: The detail parameter on to_dataframe() selects the output tier: None (default, auto-detected), "event" (tier 2), or "region" (tier 3). The data_type parameter allows creating typed empty DataFrames with expected columns when input is empty.

Development

Running Tests

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest tests/ -v

Test Coverage

The package includes 146 tests covering:

• ICAO text parsing and event building
• Multi-region parsing and non-contiguous band splitting
• NOAA message parsing and alert extraction
• Fetcher functionality
• DataFrame adapter conversions (advisory, event, and region DataFrames)
• GeoJSON output
• Data validation

License

MIT License - see LICENSE file for details.

Acknowledgments

• ICAO for the space weather advisory format specification
• NOAA Space Weather Prediction Center for the alerts API