Metadata-Version: 2.4
Name: swx-advisories
Version: 0.3.0
Summary: Parse and process space weather advisories from ICAO and NOAA/SWPC
Project-URL: Homepage, https://gitlab.com/KNMI-OSS/spaceweather/swx-advisories
Project-URL: Repository, https://gitlab.com/KNMI-OSS/spaceweather/swx-advisories
Author-email: KNMI <eelco.doornbos@knmi.nl>
License-Expression: Apache-2.0
Keywords: ICAO,NOAA,SWPC,advisories,alerts,aviation,space-weather
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Science/Research
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Scientific/Engineering :: Atmospheric Science
Requires-Python: >=3.10
Requires-Dist: httpx>=0.25
Requires-Dist: pandas>=2.0
Requires-Dist: pydantic>=2.0
Description-Content-Type: text/markdown

# 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

```bash
pip install swx-advisories
```

Or install from source:

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

## Quick Start

### ICAO Advisories

```python
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

```python
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

```python
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:

```python
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

```python
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:

```python
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`:

```python
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:

```python
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:**

```python
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

```python
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

```python
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

```python
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

```python
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:

```python
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`:

```python
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:

```python
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

```python
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:

```python
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

```python
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](https://tools.ietf.org/html/rfc7946).

```python
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:

```python
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:

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

## Data Models

### Enums

```python
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

```python
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

```bash
# 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
