spothole/data/solar_conditions.py

import json
from dataclasses import dataclass

# Lookup tables for derived text descriptions.
# Each threshold-based table is a list of (min_value, description) pairs in descending order;
# the first entry whose threshold the value meets or exceeds is used.

XRAY_CLASS_DESCRIPTIONS = {
    "X": "Wide area HF radio blackout across sunlit side",
    "M": "Occasional loss of HF communications on sunlit side",
    "C": "Low absorption of HF signals on sunlit side",
    "B": "No significant radio blackout",
    "A": "No impact",
}

PROTON_FLUX_DESCRIPTIONS = [
    (1000000, "Complete HF blackout in polar regions"),
    (100000, "Partial HF blackout in polar regions"),
    (10000, "Degraded HF propagation in polar regions"),
    (1000, "Small effect on HF propagation in polar regions"),
    (100, "Minor effect on HF propagation in polar regions"),
    (10, "Very minor effect on HF propagation in polar regions"),
    (0, "No impact"),
]

SOLAR_STORM_SCALES = [
    (100000, 5),
    (10000, 4),
    (1000, 3),
    (100, 2),
    (10, 1),
    (0, 0),
]

GEOMAG_STORM_DESCRIPTIONS = [
    (9, "Complete HF blackout"),
    (8, "HF sporadic only"),
    (7, "HF intermittent"),
    (6, "HF fading at higher latitudes"),
    (5, "HF fading at higher latitudes"),
    (4, "Minor HF fading at higher latitudes"),
    (3, "Minor HF fading at higher latitudes"),
    (2, "No impact"),
    (1, "No impact"),
    (0, "No impact"),
]

GEOMAG_STORM_SCALES = [
    (9, 5),
    (8, 4),
    (7, 3),
    (6, 2),
    (5, 1),
    (0, 0),
]

BAND_CONDITIONS_DESCRIPTIONS = [
    (200, "Reliable conditions on all bands including 6m"),
    (150, "Excellent conditions on all bands up to 10m, occasional 6m openings"),
    (120, "Fair to good conditions on all bands up to 10m"),
    (90, "Fair conditions on bands up to 15m"),
    (70, "Poor to fair conditions on bands up to 20m"),
    (0, "Bands above 40m unusable"),
]

ELECTRON_FLUX_DESCRIPTIONS = [
    (1000, "Partial to complete HF blackout in polar regions"),
    (100, "Degraded HF propagation in polar regions"),
    (10, "Minor impact on HF in polar regions"),
    (0, "No impact"),
]


def _xray_blackout_scale(xray):
    """Return the NOAA Radio Blackout scale number (R0-R5) for the given X-ray flux class string
    (e.g. "M4.5", "X12")."""

    if not xray or len(xray) < 2:
        return 0
    letter = xray[0].upper()
    try:
        number = float(xray[1:])
    except ValueError:
        return 0
    if letter == 'M':
        return 1 if number < 5 else 2
    if letter == 'X':
        if number < 10:
            return 3
        if number < 20:
            return 4
        return 5
    return 0


def _lookup_by_threshold(value, table, default=None):
    """Return the description from a threshold table for the given numeric value.
    The table is a list of (min_value, description) pairs in descending order."""

    if value is None:
        return default
    for threshold, description in table:
        if value >= threshold:
            return description
    return default


@dataclass
class HFBandCondition:
    """Data class representing HF propagation conditions for certain bands and time of day."""

    # Band name, e.g. "80m-40m", "20m-17m", "10m-6m"
    band: str = None
    # Time of day: "day" or "night"
    time: str = None
    # Propagation condition: "Good", "Fair", or "Poor"
    condition: str = None


@dataclass
class SolarConditions:
    """Data class representing current solar and propagation conditions."""

    # Time the data was last updated at the source, UTC seconds since UNIX epoch
    updated: float = None
    # Solar Flux Index (SFI)
    sfi: int = None
    # A-index (daily geomagnetic activity)
    a_index: int = None
    # K-index (3-hour geomagnetic activity)
    k_index: int = None
    # X-ray flux class, e.g. "B2.3", "C1.0"
    xray: str = None
    # Proton flux
    proton_flux: int = None
    # Electron flux
    electron_flux: int = None
    # Aurora activity level
    aurora: int = None
    # Latitude in degrees of the aurora boundary
    aurora_latitude: float = None
    # Sunspot count
    sunspots: int = None
    # Solar wind speed in km/s
    solar_wind: float = None
    # Interplanetary magnetic field strength in nT
    magnetic_field: float = None
    # Geomagnetic field condition, e.g. "Quiet", "Unsettled", "Active", "Storm"
    geomag_field: str = None
    # Geomagnetic background noise level, e.g. "S0", "S1", "S2"
    geomag_noise: str = None
    # HF band propagation conditions, keyed by "{band}-{time}" e.g. "80m-40m-day"
    hf_conditions: dict = None
    # VHF propagation conditions, keyed by condition name
    vhf_conditions: dict = None
    # NOAA Kp index 3-day forecast, keyed by UNIX timestamp of the start of each 3-hour UTC period
    k_index_forecast: dict = None
    # NOAA Solar Radiation Storm (S1 or greater) probability forecast, keyed by UNIX timestamp of start of day UTC
    solar_storm_forecast: dict = None
    # NOAA Radio Blackout (R1-R2) probability forecast, keyed by UNIX timestamp of start of day UTC
    blackout_forecast_r1r2: dict = None
    # NOAA Radio Blackout (R3 or greater) probability forecast, keyed by UNIX timestamp of start of day UTC
    blackout_forecast_r3_or_greater: dict = None

    # Derived values (populated by infer_descriptions())
    # HF radio blackout risk description, derived from xray
    xray_desc: str = None
    # HF radio blackout scale number (R0-R5), derived from xray
    radio_blackout_scale: int = None
    # Solar radiation storm level description, derived from proton_flux
    proton_flux_desc: str = None
    # Solar radiation storm scale number (S0-S5), derived from proton_flux
    solar_storm_scale: int = None
    # Geomagnetic storm level description, derived from k_index
    geomag_storm_desc: str = None
    # Geomagnetic storm scale number (G0-G5), derived from k_index
    geomag_storm_scale: int = None
    # Overall HF band conditions summary, derived from sfi
    band_conditions_desc: str = None
    # Electron flux description, derived from electron_flux
    electron_flux_desc: str = None

    def infer_descriptions(self):
        """Populate derived text description fields from the current numeric/raw field values."""

        if self.xray and len(self.xray) > 0:
            self.xray_desc = XRAY_CLASS_DESCRIPTIONS.get(self.xray[0].upper())
            self.radio_blackout_scale = _xray_blackout_scale(self.xray)
        self.proton_flux_desc = _lookup_by_threshold(self.proton_flux, PROTON_FLUX_DESCRIPTIONS)
        self.solar_storm_scale = _lookup_by_threshold(self.proton_flux, SOLAR_STORM_SCALES)
        self.geomag_storm_desc = _lookup_by_threshold(self.k_index, GEOMAG_STORM_DESCRIPTIONS)
        self.geomag_storm_scale = _lookup_by_threshold(self.k_index, GEOMAG_STORM_SCALES)
        self.band_conditions_desc = _lookup_by_threshold(self.sfi, BAND_CONDITIONS_DESCRIPTIONS)
        self.electron_flux_desc = _lookup_by_threshold(self.electron_flux, ELECTRON_FLUX_DESCRIPTIONS)

    def to_json(self):
        """JSON serialise"""

        return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True)