#!/usr/bin/env python3 from __future__ import annotations import argparse import ast import json import re import sys from pathlib import Path import tkinter as tk from tkinter import filedialog, messagebox import matplotlib.pyplot as plt import pandas as pd import numpy as np from scipy.stats import mannwhitneyu from statsmodels.nonparametric.smoothers_lowess import lowess # --------------------------------------------------------------------- # Defaults # --------------------------------------------------------------------- DATE_DEFAULT = "2026-04-02" BAND_LABEL_DEFAULT = "20m" BAND_MIN_HZ_DEFAULT = 14_095_000 BAND_MAX_HZ_DEFAULT = 14_099_000 WINDOW_DEFAULT_MIN = 15 LOWESS_FRAC_DEFAULT = 0.30 TOP_RX_PREVIEW = 10 MAIN_CLUSTER_HALF_WIDTH_HZ = 50 USE_PER_RX_NORMALISATION = True DIST_NEAR_MAX_KM = 400 DIST_MID_MAX_KM = 1500 BZ_SMOOTH_MINUTES = 30 GOES_SMOOTH_MINUTES = 30 SOLAR_LINE_ALPHA = 0.95 SOLAR_GRID_ALPHA = 0.25 # --------------------------------------------------------------------- # CLI # --------------------------------------------------------------------- def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Analyse WSPR reports from a CSV dump and build " "windowed median-offset / IQR trend outputs, with optional " "solar context parsed from PSK HTML day pages and/or JSON tails. " "Run with --gui for an interactive picker." ) ) parser.add_argument( "csv_file", nargs="?", default="wspr_logs.csv", help="Input CSV file (default: wspr_logs.csv in current directory)", ) parser.add_argument( "--gui", action="store_true", help="Launch tkinter GUI instead of running directly from CLI.", ) parser.add_argument( "--start-utc", default=f"{DATE_DEFAULT} 00:00", help=f"UTC start datetime in 'YYYY-MM-DD HH:MM' form " f"(default: {DATE_DEFAULT} 00:00)", ) parser.add_argument( "--end-utc", default=f"{DATE_DEFAULT} 23:59", help=f"UTC end datetime in 'YYYY-MM-DD HH:MM' form " f"(default: {DATE_DEFAULT} 23:59)", ) parser.add_argument( "--band-label", default=BAND_LABEL_DEFAULT, help=f"Band label for titles/output (default: {BAND_LABEL_DEFAULT})", ) parser.add_argument( "--band-min-hz", type=int, default=BAND_MIN_HZ_DEFAULT, help=f"Minimum frequency in Hz (default: {BAND_MIN_HZ_DEFAULT})", ) parser.add_argument( "--band-max-hz", type=int, default=BAND_MAX_HZ_DEFAULT, help=f"Maximum frequency in Hz (default: {BAND_MAX_HZ_DEFAULT})", ) parser.add_argument( "--window-minutes", type=int, default=WINDOW_DEFAULT_MIN, help=f"Window size in minutes (default: {WINDOW_DEFAULT_MIN})", ) parser.add_argument( "--lowess-frac", type=float, default=LOWESS_FRAC_DEFAULT, help=f"LOWESS smoothing fraction (default: {LOWESS_FRAC_DEFAULT})", ) parser.add_argument( "--top-rx", type=int, default=0, help=( "If > 0, also run a second analysis using only the N most frequent " "reporting stations for the selected time/frequency slice." ), ) parser.add_argument( "--outdir", default="atmos_wobble_outputs", help="Output directory (default: atmos_wobble_outputs)", ) parser.add_argument( "--solar-html", nargs="*", default=[], help=( "One or more PSK HTML day pages containing GOES / Bt / Bz / By data. " "Example: day=2026-04-05.html day=2026-04-06.html" ), ) parser.add_argument( "--solar-wind-json", default="", help=( "Optional solar_wind_mag_1day.json tail file to extend Bt/Bz/By coverage." ), ) parser.add_argument( "--goes-json", default="", help=( "Optional goes_xray_1day.json tail file to extend GOES coverage." ), ) parser.add_argument( "--bz-smooth-minutes", type=int, default=BZ_SMOOTH_MINUTES, help="Smoothing window (minutes) for Bz in solar plots", ) parser.add_argument( "--goes-smooth-minutes", type=int, default=GOES_SMOOTH_MINUTES, help="Smoothing window (minutes) for GOES X-ray flux in solar plots", ) return parser.parse_args() # --------------------------------------------------------------------- # Data loading / filtering # --------------------------------------------------------------------- def parse_utc_datetime(text: str) -> pd.Timestamp: ts = pd.Timestamp(text) if ts.tzinfo is None: ts = ts.tz_localize("UTC") else: ts = ts.tz_convert("UTC") return ts def _clean_text(val: object) -> str: if pd.isna(val): return "" return str(val).replace("\xa0", " ").strip() def load_csv(csv_path: Path) -> pd.DataFrame: """ Handles multiple CSV styles. Output dataframe columns used downstream: - time_utc - freqHz - rxCall - mode - snr - distance_km (optional but useful) """ raw = pd.read_csv(csv_path, header=None, dtype=str) if raw.empty: raise ValueError("CSV appears to be empty.") # ----------------------------------------------------------------- # Case 1: older awkward export with 21 columns and 7 leading blanks # ----------------------------------------------------------------- if raw.shape[1] >= 21: cols = [ "blank0", "blank1", "blank2", "blank3", "blank4", "blank5", "blank6", "timestamp", "txCall", "freqMHz", "snr", "drift", "txGrid", "power", "dt", "rxCall", "rxGrid", "distance_km", "azimuth_deg", "mode", "reporter" ] work = raw.iloc[:, :21].copy() work.columns = cols header_as_row = pd.DataFrame([raw.columns[:21]], columns=cols) work = pd.concat([header_as_row, work], ignore_index=True) for col in work.columns: work[col] = work[col].map(_clean_text) work["time_utc"] = pd.to_datetime(work["timestamp"], utc=True, errors="coerce") work["freqHz"] = pd.to_numeric(work["freqMHz"], errors="coerce") * 1_000_000 work["snr"] = pd.to_numeric(work["snr"], errors="coerce") work["rxCall"] = work["rxCall"].astype(str).str.strip() work["mode"] = work["mode"].astype(str).str.strip() work["distance_km"] = pd.to_numeric(work["distance_km"], errors="coerce") out = work.dropna(subset=["time_utc", "freqHz"]).copy() out = out[ (out["rxCall"] != "") & (out["mode"] != "") ].copy() return out.sort_values("time_utc").reset_index(drop=True) # ----------------------------------------------------------------- # Case 2: cleaner CSV / newer WSPR export with normal headers # ----------------------------------------------------------------- norm = pd.read_csv(csv_path, dtype=str) norm.columns = [str(c).replace("\xa0", " ").strip() for c in norm.columns] for col in norm.columns: norm[col] = norm[col].map(_clean_text) time_candidates = ["time_utc", "timestamp", "time", "utc_time", "date_time", "Date"] freq_candidates = ["freqHz", "freq_hz", "frequency_hz", "freq", "Frequency"] rx_candidates = ["rxCall", "rx_call", "rx", "reporter_call", "Reported by"] mode_candidates = ["mode", "Mode"] snr_candidates = ["snr", "SNR"] dist_candidates = ["distance_km", "Distance_km", "Reported, km", "reported_km"] def pick(candidates: list[str], required: bool = True) -> str | None: for c in candidates: if c in norm.columns: return c if required: raise ValueError(f"CSV is missing one of the required columns: {candidates}") return None time_col = pick(time_candidates) freq_col = pick(freq_candidates) rx_col = pick(rx_candidates) mode_col = pick(mode_candidates) snr_col = pick(snr_candidates, required=False) dist_col = pick(dist_candidates, required=False) out = norm.copy() out["time_utc"] = pd.to_datetime(out[time_col], utc=True, errors="coerce") freq_series = pd.to_numeric(out[freq_col], errors="coerce") if freq_series.dropna().median() < 1000: out["freqHz"] = freq_series * 1_000_000 else: out["freqHz"] = freq_series out["rxCall"] = out[rx_col].astype(str).str.strip() out["mode"] = out[mode_col].astype(str).str.strip() out["snr"] = pd.to_numeric(out[snr_col], errors="coerce") if snr_col else pd.NA out["distance_km"] = ( pd.to_numeric(out[dist_col], errors="coerce") if dist_col else pd.NA ) out = out.dropna(subset=["time_utc", "freqHz"]).copy() return out.sort_values("time_utc").reset_index(drop=True) def filter_wspr_band_timerange( df: pd.DataFrame, start_utc: str, end_utc: str, band_min_hz: int, band_max_hz: int, ) -> pd.DataFrame: start_ts = parse_utc_datetime(start_utc) end_ts = parse_utc_datetime(end_utc) if end_ts <= start_ts: raise ValueError("End time must be later than start time.") mask = ( (df["mode"].str.upper().str.contains("WSPR", na=False)) & (df["freqHz"] >= band_min_hz) & (df["freqHz"] <= band_max_hz) & (df["time_utc"] >= start_ts) & (df["time_utc"] <= end_ts) ) out = df.loc[mask].copy() if out.empty: raise ValueError("No WSPR records found in the chosen time/frequency range.") return out # --------------------------------------------------------------------- # Solar parsing / merging # --------------------------------------------------------------------- def extract_js_array(text: str, var_name: str): m = re.search(rf"var\s+{re.escape(var_name)}\s*=\s*(\[[\s\S]*?\]);", text) if not m: return None arr_txt = m.group(1).replace("null", "None") return ast.literal_eval(arr_txt) def _safe_series_df(time_values, value_values, time_col: str = "time_utc", value_col: str = "value") -> pd.DataFrame: if time_values is None or value_values is None: return pd.DataFrame(columns=[time_col, value_col]) if len(time_values) != len(value_values): return pd.DataFrame(columns=[time_col, value_col]) df = pd.DataFrame({time_col: pd.to_datetime(time_values, utc=True, errors="coerce"), value_col: pd.to_numeric(value_values, errors="coerce")}) df = df.dropna(subset=[time_col, value_col]).copy() return df def load_solar_html_file(html_path: Path) -> dict[str, pd.DataFrame]: text = html_path.read_text(encoding="utf-8", errors="ignore") # GOES 0.1–0.8 nm (long) goes_x = extract_js_array(text, "goes_x_long") goes_y = extract_js_array(text, "goes_y_long") goes = _safe_series_df(goes_x, goes_y, value_col="goes_long") # IMF / solar wind magnetic field components mag_x_bt = extract_js_array(text, "mag_x_bt") mag_y_bt = extract_js_array(text, "mag_y_bt") bt = _safe_series_df(mag_x_bt, mag_y_bt, value_col="bt") mag_x_bz = extract_js_array(text, "mag_x_bz") mag_y_bz = extract_js_array(text, "mag_y_bz") bz = _safe_series_df(mag_x_bz, mag_y_bz, value_col="bz") mag_x_by = extract_js_array(text, "mag_x_by") mag_y_by = extract_js_array(text, "mag_y_by") by = _safe_series_df(mag_x_by, mag_y_by, value_col="by") return {"goes": goes, "bt": bt, "bz": bz, "by": by} def load_solar_html_files(html_paths: list[Path]) -> dict[str, pd.DataFrame]: tables: dict[str, list[pd.DataFrame]] = {"goes": [], "bt": [], "bz": [], "by": []} for path in html_paths: if not path.exists(): continue parsed = load_solar_html_file(path) for key, df in parsed.items(): if not df.empty: tables[key].append(df) out: dict[str, pd.DataFrame] = {} for key, parts in tables.items(): if parts: merged = pd.concat(parts, ignore_index=True) merged = merged.drop_duplicates("time_utc").sort_values("time_utc").reset_index(drop=True) out[key] = merged else: val_col = "goes_long" if key == "goes" else key out[key] = pd.DataFrame(columns=["time_utc", val_col]) return out def load_solar_wind_json(path: Path) -> dict[str, pd.DataFrame]: if not path or not path.exists(): return { "bt": pd.DataFrame(columns=["time_utc", "bt"]), "bz": pd.DataFrame(columns=["time_utc", "bz"]), "by": pd.DataFrame(columns=["time_utc", "by"]), } raw = json.loads(path.read_text(encoding="utf-8")) if not raw: return { "bt": pd.DataFrame(columns=["time_utc", "bt"]), "bz": pd.DataFrame(columns=["time_utc", "bz"]), "by": pd.DataFrame(columns=["time_utc", "by"]), } if isinstance(raw, list) and raw and isinstance(raw[0], list): df = pd.DataFrame(raw[1:], columns=raw[0]) else: df = pd.DataFrame(raw) if "time_tag" not in df.columns: return { "bt": pd.DataFrame(columns=["time_utc", "bt"]), "bz": pd.DataFrame(columns=["time_utc", "bz"]), "by": pd.DataFrame(columns=["time_utc", "by"]), } df["time_utc"] = pd.to_datetime(df["time_tag"], utc=True, errors="coerce") out = {} for src_col, dst_col in [("bt", "bt"), ("bz_gsm", "bz"), ("by_gsm", "by")]: if src_col in df.columns: tmp = df[["time_utc", src_col]].copy() tmp[dst_col] = pd.to_numeric(tmp[src_col], errors="coerce") tmp = tmp.drop(columns=[src_col]).dropna(subset=["time_utc", dst_col]) out[dst_col] = tmp else: out[dst_col] = pd.DataFrame(columns=["time_utc", dst_col]) return out def load_goes_json(path: Path) -> pd.DataFrame: if not path or not path.exists(): return pd.DataFrame(columns=["time_utc", "goes_long"]) raw = json.loads(path.read_text(encoding="utf-8")) df = pd.DataFrame(raw) if df.empty or "time_tag" not in df.columns: return pd.DataFrame(columns=["time_utc", "goes_long"]) df["time_utc"] = pd.to_datetime(df["time_tag"], utc=True, errors="coerce") energy_col = None for cand in ["energy", "channel", "band"]: if cand in df.columns: energy_col = cand break flux_col = None for cand in ["flux", "observed_flux", "value"]: if cand in df.columns: flux_col = cand break if flux_col is None: return pd.DataFrame(columns=["time_utc", "goes_long"]) if energy_col is not None: df = df[df[energy_col].astype(str) == "0.1-0.8nm"].copy() df["goes_long"] = pd.to_numeric(df[flux_col], errors="coerce") df = df.dropna(subset=["time_utc", "goes_long"]).copy() return df[["time_utc", "goes_long"]] def merge_solar_sources( html_paths: list[Path], solar_wind_json: Path | None, goes_json: Path | None, ) -> dict[str, pd.DataFrame]: html_data = load_solar_html_files(html_paths) wind_json_data = load_solar_wind_json(solar_wind_json) if solar_wind_json else { "bt": pd.DataFrame(columns=["time_utc", "bt"]), "bz": pd.DataFrame(columns=["time_utc", "bz"]), "by": pd.DataFrame(columns=["time_utc", "by"]), } goes_json_df = load_goes_json(goes_json) if goes_json else pd.DataFrame(columns=["time_utc", "goes_long"]) def _merge_value(key: str, col: str) -> pd.DataFrame: parts = [] if key in html_data and not html_data[key].empty: parts.append(html_data[key][["time_utc", col]]) if key in wind_json_data and not wind_json_data[key].empty: parts.append(wind_json_data[key][["time_utc", col]]) if parts: out = pd.concat(parts, ignore_index=True) out = out.drop_duplicates("time_utc").sort_values("time_utc").reset_index(drop=True) return out return pd.DataFrame(columns=["time_utc", col]) bt = _merge_value("bt", "bt") bz = _merge_value("bz", "bz") by = _merge_value("by", "by") goes_parts = [] if not html_data.get("goes", pd.DataFrame()).empty: goes_parts.append(html_data["goes"][["time_utc", "goes_long"]]) if not goes_json_df.empty: goes_parts.append(goes_json_df[["time_utc", "goes_long"]]) if goes_parts: goes = pd.concat(goes_parts, ignore_index=True) goes = goes.drop_duplicates("time_utc").sort_values("time_utc").reset_index(drop=True) goes = goes[goes["goes_long"] > 0].copy() # ignore zero dropouts else: goes = pd.DataFrame(columns=["time_utc", "goes_long"]) return {"bt": bt, "bz": bz, "by": by, "goes": goes} def clip_solar_to_range( solar: dict[str, pd.DataFrame], start_ts: pd.Timestamp, end_ts: pd.Timestamp, ) -> dict[str, pd.DataFrame]: out = {} for key, df in solar.items(): if df.empty: out[key] = df.copy() continue tmp = df[(df["time_utc"] >= start_ts) & (df["time_utc"] <= end_ts)].copy() out[key] = tmp.reset_index(drop=True) return out # --------------------------------------------------------------------- # Analysis logic # --------------------------------------------------------------------- def classify_distance_band(distance_km: float | int | None) -> str: if pd.isna(distance_km): return "Unknown" if distance_km < 200: return "<200 km" if distance_km < 400: return "200–400 km" if distance_km < 800: return "400–800 km" return ">800 km" def build_windowed( df: pd.DataFrame, window_minutes: int, lowess_frac: float, ) -> tuple[pd.DataFrame, float, pd.DataFrame]: main_freq = df["freqHz"].median() filtered_df = df[ (df["freqHz"] > main_freq - MAIN_CLUSTER_HALF_WIDTH_HZ) & (df["freqHz"] < main_freq + MAIN_CLUSTER_HALF_WIDTH_HZ) ].copy() if filtered_df.empty: raise ValueError("No points remain after main tone cluster filtering.") reference_hz = float(filtered_df["freqHz"].median()) point_df = filtered_df.copy() point_df["offset_hz"] = point_df["freqHz"] - reference_hz if USE_PER_RX_NORMALISATION: point_df["offset_hz"] = ( point_df["offset_hz"] - point_df.groupby("rxCall")["offset_hz"].transform("median") ) point_df["distance_band"] = point_df["distance_km"].apply(classify_distance_band) work = point_df.set_index("time_utc") window = f"{window_minutes}min" grouped = work["offset_hz"].resample(window) windowed = pd.DataFrame( { "median_offset_hz": grouped.median(), "q25_hz": grouped.quantile(0.25), "q75_hz": grouped.quantile(0.75), "n": grouped.count(), "mean_offset_hz": grouped.mean(), "std_offset_hz": grouped.std(), } ) windowed["iqr_hz"] = windowed["q75_hz"] - windowed["q25_hz"] windowed = windowed[windowed["n"] > 0].copy() if len(windowed) >= 3: valid = windowed["median_offset_hz"].notna() x = windowed.index.view("int64")[valid] y = windowed.loc[valid, "median_offset_hz"].to_numpy() smoothed = lowess(y, x, frac=lowess_frac, return_sorted=False) windowed.loc[valid, "lowess_median_offset_hz"] = smoothed else: windowed["lowess_median_offset_hz"] = windowed["median_offset_hz"] windowed["rolling3_median_offset_hz"] = ( windowed["median_offset_hz"] .rolling(3, center=True, min_periods=1) .mean() ) return windowed.reset_index(), reference_hz, point_df.reset_index(drop=True) def summary_stats(windowed: pd.DataFrame) -> dict[str, float | int | None]: out: dict[str, float | int | None] = {} out["windows"] = int(len(windowed)) out["points"] = int(windowed["n"].sum()) out["median_n_per_window"] = float(windowed["n"].median()) out["mean_n_per_window"] = float(windowed["n"].mean()) out["median_iqr_hz"] = float(windowed["iqr_hz"].median()) out["max_iqr_hz"] = float(windowed["iqr_hz"].max()) out["overall_median_of_window_medians_hz"] = float( windowed["median_offset_hz"].median() ) morning = windowed[ (windowed["time_utc"].dt.hour >= 10) & (windowed["time_utc"].dt.hour < 13) ]["median_offset_hz"].dropna() afternoon = windowed[ (windowed["time_utc"].dt.hour >= 14) & (windowed["time_utc"].dt.hour < 18) ]["median_offset_hz"].dropna() out["morning_windows"] = int(len(morning)) out["afternoon_windows"] = int(len(afternoon)) out["morning_mean_hz"] = float(morning.mean()) if len(morning) else None out["afternoon_mean_hz"] = float(afternoon.mean()) if len(afternoon) else None if len(morning) >= 2 and len(afternoon) >= 2: stat = mannwhitneyu(morning, afternoon, alternative="two-sided") out["mannwhitney_u"] = float(stat.statistic) out["mannwhitney_p"] = float(stat.pvalue) else: out["mannwhitney_u"] = None out["mannwhitney_p"] = None if len(windowed) >= 2: out["lag1_autocorr"] = float(windowed["median_offset_hz"].autocorr(lag=1)) else: out["lag1_autocorr"] = None return out def reporters_table(df: pd.DataFrame) -> pd.DataFrame: rep = ( df.groupby("rxCall", dropna=False) .agg( spots=("rxCall", "size"), first_seen_utc=("time_utc", "min"), last_seen_utc=("time_utc", "max"), median_freq_hz=("freqHz", "median"), median_snr_db=("snr", "median"), median_distance_km=("distance_km", "median"), ) .sort_values(["spots", "rxCall"], ascending=[False, True]) .reset_index() ) return rep # --------------------------------------------------------------------- # Plot helpers / outputs # --------------------------------------------------------------------- def _prepare_component_for_plot( df: pd.DataFrame, value_col: str, smooth_minutes: int, smooth_col: str, ) -> pd.DataFrame: if df.empty: return df.copy() tmp = df.copy().dropna(subset=["time_utc", value_col]).copy() if tmp.empty: return tmp tmp[smooth_col] = ( tmp.set_index("time_utc")[value_col] .rolling(f"{smooth_minutes}min", center=True, min_periods=1) .mean() .reset_index(drop=True) ) return tmp def _prepare_goes_for_plot(goes_df: pd.DataFrame, goes_smooth_minutes: int) -> pd.DataFrame: if goes_df.empty: return goes_df.copy() tmp = goes_df.copy() tmp = tmp[tmp["goes_long"] > 0].copy() if tmp.empty: return tmp tmp["goes_log10"] = tmp["goes_long"].apply(lambda x: pd.NA if x <= 0 else np.log10(x)) tmp = tmp.dropna(subset=["goes_log10"]).copy() tmp["goes_log10_smooth"] = ( tmp.set_index("time_utc")["goes_log10"] .rolling(f"{goes_smooth_minutes}min", center=True, min_periods=1) .mean() .reset_index(drop=True) ) return tmp def plot_windowed( windowed: pd.DataFrame, stats: dict[str, float | int | None], out_png: Path, title_suffix: str, lowess_frac: float, ) -> None: fig, (ax1, ax2) = plt.subplots( 2, 1, figsize=(13, 8), sharex=True, constrained_layout=True ) ax1.scatter( windowed["time_utc"], windowed["median_offset_hz"], s=30, label="Window median offset", zorder=3, ) ax1.plot( windowed["time_utc"], windowed["rolling3_median_offset_hz"], linewidth=1.6, label="Rolling mean (3 windows)", ) ax1.plot( windowed["time_utc"], windowed["lowess_median_offset_hz"], linewidth=2.2, label=f"LOWESS (frac={lowess_frac:.2f})", ) ax1.axhline(0, linestyle="--", linewidth=1) ax1.set_ylabel("Median offset (Hz)") ax1.set_title(f"Median offset vs daily reference — {title_suffix}") ax1.legend(loc="best") ax1.grid(True, alpha=0.3) ax2.bar(windowed["time_utc"], windowed["iqr_hz"], width=0.008, label="IQR") ax2.plot(windowed["time_utc"], windowed["n"], linewidth=1.5, label="n per window") ax2.set_ylabel("IQR / n") ax2.set_title( "Window spread and sample size" f" | windows={stats['windows']} points={stats['points']} " f"median n={stats['median_n_per_window']:.1f}" ) ax2.legend(loc="best") ax2.grid(True, alpha=0.3) fig.savefig(out_png, dpi=150) plt.close(fig) def plot_distance_scatter( point_df: pd.DataFrame, out_png: Path, title_suffix: str, ) -> None: usable = point_df.dropna(subset=["distance_km", "offset_hz"]).copy() if usable.empty: return fig, ax = plt.subplots(figsize=(13, 5), constrained_layout=True) sc = ax.scatter( usable["time_utc"], usable["offset_hz"], c=usable["distance_km"], cmap="viridis", s=24, alpha=0.85, edgecolors="none", ) ax.axhline(0, linestyle="--", linewidth=1) ax.set_title(f"Point offsets coloured by distance — {title_suffix}") ax.set_ylabel("Point offset (Hz)") ax.grid(True, alpha=0.3) cbar = fig.colorbar(sc, ax=ax) cbar.set_label("Distance (km)") fig.savefig(out_png, dpi=150) plt.close(fig) def plot_distance_band_scatter( point_df: pd.DataFrame, out_png: Path, title_suffix: str, ) -> None: usable = point_df.dropna(subset=["offset_hz"]).copy() if usable.empty: return band_order = [ "<200 km", "200–400 km", "400–800 km", ">800 km", "Unknown", ] fig, ax = plt.subplots(figsize=(13, 5), constrained_layout=True) for band in band_order: sub = usable[usable["distance_band"] == band] if sub.empty: continue ax.scatter( sub["time_utc"], sub["offset_hz"], s=26, alpha=0.8, label=band, ) ax.axhline(0, linestyle="--", linewidth=1) ax.set_title(f"Point offsets by distance band — {title_suffix}") ax.set_ylabel("Point offset (Hz)") ax.legend(loc="best") ax.grid(True, alpha=0.3) fig.savefig(out_png, dpi=150) plt.close(fig) def plot_distance_band_window_medians( point_df: pd.DataFrame, out_png: Path, title_suffix: str, window_minutes: int, ) -> None: usable = point_df.dropna(subset=["time_utc", "offset_hz", "distance_band"]).copy() usable = usable[usable["distance_band"] != "Unknown"].copy() if usable.empty: return usable = usable.set_index("time_utc") window = f"{window_minutes}min" grouped = ( usable.groupby("distance_band")["offset_hz"] .resample(window) .median() .reset_index() ) if grouped.empty: return pivot = grouped.pivot( index="time_utc", columns="distance_band", values="offset_hz" ) band_order = ["<200 km", "200–400 km", "400–800 km", ">800 km"] ordered_cols = [col for col in band_order if col in pivot.columns] if not ordered_cols: return fig, ax = plt.subplots(figsize=(13, 5), constrained_layout=True) for band in ordered_cols: ax.plot( pivot.index, pivot[band], linewidth=1.6, marker="o", markersize=3, label=band, ) ax.axhline(0, linestyle="--", linewidth=1) ax.set_title(f"Window median offset by distance band — {title_suffix}") ax.set_ylabel("Median offset (Hz)") ax.set_xlabel("UTC time") ax.legend(loc="best") ax.grid(True, alpha=0.3) fig.savefig(out_png, dpi=150) plt.close(fig) def plot_solar_context( solar: dict[str, pd.DataFrame], out_png: Path, title_suffix: str, bz_smooth_minutes: int, goes_smooth_minutes: int, ) -> None: bt = solar.get("bt", pd.DataFrame()) bz = _prepare_component_for_plot(solar.get("bz", pd.DataFrame()), "bz", bz_smooth_minutes, "bz_smooth") by = solar.get("by", pd.DataFrame()) goes = _prepare_goes_for_plot(solar.get("goes", pd.DataFrame()), goes_smooth_minutes) panels = [] if not bt.empty or not bz.empty or not by.empty: panels.append("mag") if not goes.empty: panels.append("goes") if not panels: return fig, axes = plt.subplots(len(panels), 1, figsize=(13, 3.2 * len(panels)), sharex=True, constrained_layout=True) if len(panels) == 1: axes = [axes] ax_idx = 0 if "mag" in panels: ax = axes[ax_idx] if not bt.empty: ax.plot(bt["time_utc"], bt["bt"], linewidth=1.6, label="Bt") if not bz.empty: ax.plot(bz["time_utc"], bz["bz"], linewidth=0.9, alpha=0.30, label="Bz raw") ax.plot(bz["time_utc"], bz["bz_smooth"], linewidth=2.2, alpha=SOLAR_LINE_ALPHA, label=f"Bz smoothed ({bz_smooth_minutes} min)") if not by.empty: ax.plot(by["time_utc"], by["by"], linewidth=1.2, label="By") ax.axhline(0, linestyle="--", linewidth=1) ax.set_ylabel("nT") ax.set_title(f"Solar wind magnetic context — {title_suffix}") ax.legend(loc="best") ax.grid(True, alpha=SOLAR_GRID_ALPHA) ax_idx += 1 if "goes" in panels: ax = axes[ax_idx] ax.plot(goes["time_utc"], goes["goes_log10"], linewidth=1.0, alpha=0.35, label="GOES raw log10") ax.plot(goes["time_utc"], goes["goes_log10_smooth"], linewidth=2.0, alpha=SOLAR_LINE_ALPHA, label=f"GOES smoothed ({goes_smooth_minutes} min)") ax.set_ylabel("log10 flux") ax.set_title(f"GOES 0.1–0.8 nm context — {title_suffix}") ax.legend(loc="best") ax.grid(True, alpha=SOLAR_GRID_ALPHA) fig.savefig(out_png, dpi=150) plt.close(fig) def plot_distance_band_with_solar( point_df: pd.DataFrame, solar: dict[str, pd.DataFrame], out_png: Path, title_suffix: str, window_minutes: int, bz_smooth_minutes: int, goes_smooth_minutes: int, ) -> None: usable = point_df.dropna(subset=["time_utc", "offset_hz", "distance_band"]).copy() usable = usable[usable["distance_band"] != "Unknown"].copy() if usable.empty: return usable = usable.set_index("time_utc") grouped = ( usable.groupby("distance_band")["offset_hz"] .resample(f"{window_minutes}min") .median() .reset_index() ) if grouped.empty: return pivot = grouped.pivot(index="time_utc", columns="distance_band", values="offset_hz") band_order = ["<200 km", "200–400 km", "400–800 km", ">800 km"] ordered_cols = [col for col in band_order if col in pivot.columns] if not ordered_cols: return bt = solar.get("bt", pd.DataFrame()) bz = _prepare_component_for_plot(solar.get("bz", pd.DataFrame()), "bz", bz_smooth_minutes, "bz_smooth") goes = _prepare_goes_for_plot(solar.get("goes", pd.DataFrame()), goes_smooth_minutes) nrows = 1 + int((not bt.empty) or (not bz.empty)) + int(not goes.empty) fig, axes = plt.subplots(nrows, 1, figsize=(13, 3.2 * nrows), sharex=True, constrained_layout=True) if nrows == 1: axes = [axes] ax0 = axes[0] for band in ordered_cols: ax0.plot(pivot.index, pivot[band], linewidth=1.6, marker="o", markersize=3, label=band) ax0.axhline(0, linestyle="--", linewidth=1) ax0.set_ylabel("Median offset (Hz)") ax0.set_title(f"Window median offset by distance band + solar context — {title_suffix}") ax0.legend(loc="best") ax0.grid(True, alpha=0.3) row = 1 if (not bt.empty) or (not bz.empty): ax = axes[row] if not bt.empty: ax.plot(bt["time_utc"], bt["bt"], linewidth=1.5, label="Bt") if not bz.empty: ax.plot(bz["time_utc"], bz["bz"], linewidth=0.9, alpha=0.30, label="Bz raw") ax.plot(bz["time_utc"], bz["bz_smooth"], linewidth=2.2, alpha=SOLAR_LINE_ALPHA, label=f"Bz smoothed ({bz_smooth_minutes} min)") ax.axhline(0, linestyle="--", linewidth=1) ax.set_ylabel("nT") ax.legend(loc="best") ax.grid(True, alpha=SOLAR_GRID_ALPHA) row += 1 if not goes.empty: ax = axes[row] ax.plot(goes["time_utc"], goes["goes_log10"], linewidth=0.9, alpha=0.3, label="GOES raw log10") ax.plot(goes["time_utc"], goes["goes_log10_smooth"], linewidth=2.0, alpha=SOLAR_LINE_ALPHA, label=f"GOES smoothed ({goes_smooth_minutes} min)") ax.set_ylabel("log10 flux") ax.legend(loc="best") ax.grid(True, alpha=SOLAR_GRID_ALPHA) fig.savefig(out_png, dpi=150) plt.close(fig) def write_summary( out_txt: Path, input_path: Path, filtered: pd.DataFrame, windowed: pd.DataFrame, stats: dict[str, float | int | None], reporters: pd.DataFrame, reference_hz: float, args_like: argparse.Namespace, title: str, solar: dict[str, pd.DataFrame] | None = None, ) -> None: with out_txt.open("w", encoding="utf-8") as fh: fh.write(f"Analysis: {title}\n") fh.write(f"Input file: {input_path}\n") fh.write(f"UTC start: {args_like.start_utc}\n") fh.write(f"UTC end: {args_like.end_utc}\n") fh.write( f"Band: {args_like.band_label} " f"({args_like.band_min_hz} to {args_like.band_max_hz} Hz)\n" ) fh.write(f"Window: {args_like.window_minutes} minutes\n") fh.write(f"LOWESS frac: {args_like.lowess_frac}\n") fh.write( f"Main tone cluster filter: median ±{MAIN_CLUSTER_HALF_WIDTH_HZ} Hz\n" ) fh.write(f"Daily median reference frequency: {reference_hz:.3f} Hz\n") fh.write(f"Per-reporter normalisation: {USE_PER_RX_NORMALISATION}\n") fh.write(f"Solar HTML inputs: {getattr(args_like, 'solar_html', [])}\n") fh.write(f"Solar wind JSON tail: {getattr(args_like, 'solar_wind_json', '')}\n") fh.write(f"GOES JSON tail: {getattr(args_like, 'goes_json', '')}\n") fh.write(f"Bz smoothing (min): {getattr(args_like, 'bz_smooth_minutes', BZ_SMOOTH_MINUTES)}\n") fh.write(f"GOES smoothing (min): {getattr(args_like, 'goes_smooth_minutes', GOES_SMOOTH_MINUTES)}\n\n") if solar is not None: fh.write("Solar context availability:\n") for key in ["bt", "bz", "by", "goes"]: df = solar.get(key, pd.DataFrame()) fh.write(f"- {key}: {len(df)} points\n") fh.write("\n") fh.write("Top reporters by spot count:\n") fh.write(reporters.head(TOP_RX_PREVIEW).to_string(index=False)) fh.write("\n\n") fh.write("Headline statistics:\n") for key, value in stats.items(): fh.write(f"- {key}: {value}\n") fh.write("\n") fh.write("Interpretation notes:\n") fh.write( "- median_offset_hz is relative to the filtered median frequency " "of the selected WSPR records.\n" ) fh.write( "- iqr_hz measures the within-window spread; large IQR means more " "frequency scatter.\n" ) fh.write("- LOWESS is the smoothed line through the window medians.\n") fh.write( "- morning/afternoon comparison uses a Mann-Whitney U test because " "small samples are common.\n" ) fh.write( "- Solar traces are context only. GOES zero-value dropouts are ignored.\n\n" ) fh.write("Windowed data preview:\n") fh.write(windowed.head(20).to_string(index=False)) fh.write("\n") # --------------------------------------------------------------------- # Runner # --------------------------------------------------------------------- def run_one_analysis( df: pd.DataFrame, outdir: Path, base_name: str, args_like: argparse.Namespace, title_suffix: str, input_path: Path, solar: dict[str, pd.DataFrame] | None = None, ) -> None: filtered = df.copy() reporters = reporters_table(filtered) windowed, reference_hz, point_df = build_windowed( filtered, args_like.window_minutes, args_like.lowess_frac, ) stats = summary_stats(windowed) csv_path = outdir / f"{base_name}_windowed.csv" reporters_csv = outdir / f"{base_name}_reporters.csv" plot_path = outdir / f"{base_name}_trend.png" distance_plot_path = outdir / f"{base_name}_distance_scatter.png" distance_band_plot_path = outdir / f"{base_name}_distance_bands.png" distance_band_window_plot_path = outdir / f"{base_name}_distance_band_window_medians.png" solar_context_plot_path = outdir / f"{base_name}_solar_context.png" solar_combined_plot_path = outdir / f"{base_name}_distance_band_with_solar.png" solar_csv_path = outdir / f"{base_name}_solar_merged.csv" summary_path = outdir / f"{base_name}_summary.txt" windowed.to_csv(csv_path, index=False) reporters.to_csv(reporters_csv, index=False) plot_windowed(windowed, stats, plot_path, title_suffix, args_like.lowess_frac) plot_distance_scatter(point_df, distance_plot_path, title_suffix) plot_distance_band_scatter(point_df, distance_band_plot_path, title_suffix) plot_distance_band_window_medians( point_df, distance_band_window_plot_path, title_suffix, args_like.window_minutes, ) if solar is not None: plot_solar_context(solar, solar_context_plot_path, title_suffix, args_like.bz_smooth_minutes, args_like.goes_smooth_minutes) plot_distance_band_with_solar( point_df, solar, solar_combined_plot_path, title_suffix, args_like.window_minutes, args_like.bz_smooth_minutes, args_like.goes_smooth_minutes, ) solar_parts = [] for key, col in [("bt", "bt"), ("bz", "bz"), ("by", "by"), ("goes", "goes_long")]: sdf = solar.get(key, pd.DataFrame()) if sdf.empty: continue tmp = sdf[["time_utc", col]].copy() tmp["source_series"] = key solar_parts.append(tmp) if solar_parts: pd.concat(solar_parts, ignore_index=True).to_csv(solar_csv_path, index=False) write_summary( summary_path, input_path.resolve(), filtered, windowed, stats, reporters, reference_hz, args_like, title_suffix, solar=solar, ) print(f"\n[{title_suffix}]") print(f"Filtered spots: {len(filtered)}") print(f"Distinct reporters: {filtered['rxCall'].nunique()}") print(f"Filtered median reference frequency: {reference_hz:.3f} Hz") print(f"Bz smoothing: {args_like.bz_smooth_minutes} min") print(f"GOES smoothing: {args_like.goes_smooth_minutes} min") print(f"Window CSV: {csv_path}") print(f"Reporters CSV: {reporters_csv}") print(f"Trend plot: {plot_path}") print(f"Distance plot: {distance_plot_path}") print(f"Distance-band plot: {distance_band_plot_path}") print(f"Band-window plot: {distance_band_window_plot_path}") if solar is not None: print(f"Solar context: {solar_context_plot_path}") print(f"Combined solar: {solar_combined_plot_path}") print(f"Merged solar CSV: {solar_csv_path}") print(f"Summary text: {summary_path}") print("Top reporters:") print(reporters.head(min(TOP_RX_PREVIEW, len(reporters))).to_string(index=False)) # --------------------------------------------------------------------- # GUI # --------------------------------------------------------------------- class GuiArgs: def __init__( self, csv_file: str, start_utc: str, end_utc: str, band_label: str, band_min_hz: int, band_max_hz: int, window_minutes: int, lowess_frac: float, top_rx: int, outdir: str, solar_html: list[str], solar_wind_json: str, goes_json: str, bz_smooth_minutes: int, goes_smooth_minutes: int, ) -> None: self.csv_file = csv_file self.start_utc = start_utc self.end_utc = end_utc self.band_label = band_label self.band_min_hz = band_min_hz self.band_max_hz = band_max_hz self.window_minutes = window_minutes self.lowess_frac = lowess_frac self.top_rx = top_rx self.outdir = outdir self.solar_html = solar_html self.solar_wind_json = solar_wind_json self.goes_json = goes_json self.bz_smooth_minutes = bz_smooth_minutes self.goes_smooth_minutes = goes_smooth_minutes def launch_gui() -> None: root = tk.Tk() root.title("WSPR Mode Bias Explorer V6") csv_var = tk.StringVar(value="wspr_logs.csv") outdir_var = tk.StringVar(value="atmos_wobble_outputs") start_var = tk.StringVar(value=f"{DATE_DEFAULT} 00:00") end_var = tk.StringVar(value=f"{DATE_DEFAULT} 23:59") band_label_var = tk.StringVar(value=BAND_LABEL_DEFAULT) band_min_var = tk.StringVar(value=str(BAND_MIN_HZ_DEFAULT / 1e6)) band_max_var = tk.StringVar(value=str(BAND_MAX_HZ_DEFAULT / 1e6)) window_var = tk.StringVar(value=str(WINDOW_DEFAULT_MIN)) lowess_var = tk.StringVar(value=str(LOWESS_FRAC_DEFAULT)) top_rx_var = tk.StringVar(value="0") solar_html_var = tk.StringVar(value="") solar_wind_json_var = tk.StringVar(value="") goes_json_var = tk.StringVar(value="") bz_smooth_var = tk.StringVar(value=str(BZ_SMOOTH_MINUTES)) goes_smooth_var = tk.StringVar(value=str(GOES_SMOOTH_MINUTES)) row = 0 def add_label_entry(label: str, var: tk.StringVar, width: int = 60) -> None: nonlocal row tk.Label(root, text=label, anchor="w").grid( row=row, column=0, sticky="w", padx=6, pady=4 ) tk.Entry(root, textvariable=var, width=width).grid( row=row, column=1, sticky="we", padx=6, pady=4 ) row += 1 def browse_csv() -> None: path = filedialog.askopenfilename( title="Select input CSV", filetypes=[("CSV files", "*.csv"), ("All files", "*.*")], ) if path: csv_var.set(path) def browse_outdir() -> None: path = filedialog.askdirectory(title="Select output directory") if path: outdir_var.set(path) def browse_solar_html() -> None: paths = filedialog.askopenfilenames( title="Select one or more PSK solar HTML files", filetypes=[("HTML files", "*.html;*.htm"), ("All files", "*.*")], ) if paths: solar_html_var.set(" | ".join(paths)) def browse_solar_wind_json() -> None: path = filedialog.askopenfilename( title="Select solar_wind_mag_1day.json (optional)", filetypes=[("JSON files", "*.json"), ("All files", "*.*")], ) if path: solar_wind_json_var.set(path) def browse_goes_json() -> None: path = filedialog.askopenfilename( title="Select goes_xray_1day.json (optional)", filetypes=[("JSON files", "*.json"), ("All files", "*.*")], ) if path: goes_json_var.set(path) tk.Label(root, text="Input CSV", anchor="w").grid( row=row, column=0, sticky="w", padx=6, pady=4 ) tk.Entry(root, textvariable=csv_var, width=60).grid( row=row, column=1, sticky="we", padx=6, pady=4 ) tk.Button(root, text="Browse...", command=browse_csv).grid( row=row, column=2, padx=6, pady=4 ) row += 1 tk.Label(root, text="Output directory", anchor="w").grid( row=row, column=0, sticky="w", padx=6, pady=4 ) tk.Entry(root, textvariable=outdir_var, width=60).grid( row=row, column=1, sticky="we", padx=6, pady=4 ) tk.Button(root, text="Browse...", command=browse_outdir).grid( row=row, column=2, padx=6, pady=4 ) row += 1 add_label_entry("UTC start (YYYY-MM-DD HH:MM)", start_var) add_label_entry("UTC end (YYYY-MM-DD HH:MM)", end_var) add_label_entry("Band label", band_label_var) add_label_entry("Band min MHz", band_min_var) add_label_entry("Band max MHz", band_max_var) add_label_entry("Window minutes", window_var) add_label_entry("LOWESS frac", lowess_var) add_label_entry("Top RX (0 = off)", top_rx_var) add_label_entry("Bz smoothing (min)", bz_smooth_var) add_label_entry("GOES smoothing (min)", goes_smooth_var) tk.Label(root, text="Solar HTML files", anchor="w").grid( row=row, column=0, sticky="w", padx=6, pady=4 ) tk.Entry(root, textvariable=solar_html_var, width=60).grid( row=row, column=1, sticky="we", padx=6, pady=4 ) tk.Button(root, text="Browse...", command=browse_solar_html).grid( row=row, column=2, padx=6, pady=4 ) row += 1 tk.Label(root, text="Solar wind JSON tail", anchor="w").grid( row=row, column=0, sticky="w", padx=6, pady=4 ) tk.Entry(root, textvariable=solar_wind_json_var, width=60).grid( row=row, column=1, sticky="we", padx=6, pady=4 ) tk.Button(root, text="Browse...", command=browse_solar_wind_json).grid( row=row, column=2, padx=6, pady=4 ) row += 1 tk.Label(root, text="GOES JSON tail", anchor="w").grid( row=row, column=0, sticky="w", padx=6, pady=4 ) tk.Entry(root, textvariable=goes_json_var, width=60).grid( row=row, column=1, sticky="we", padx=6, pady=4 ) tk.Button(root, text="Browse...", command=browse_goes_json).grid( row=row, column=2, padx=6, pady=4 ) row += 1 status = tk.StringVar(value="Ready.") def run_analysis() -> None: try: solar_html_list = [p.strip() for p in solar_html_var.get().split("|") if p.strip()] args_like = GuiArgs( csv_file=csv_var.get().strip(), start_utc=start_var.get().strip(), end_utc=end_var.get().strip(), band_label=band_label_var.get().strip(), band_min_hz=int(float(band_min_var.get().strip()) * 1e6), band_max_hz=int(float(band_max_var.get().strip()) * 1e6), window_minutes=int(window_var.get().strip()), lowess_frac=float(lowess_var.get().strip()), top_rx=int(top_rx_var.get().strip()), outdir=outdir_var.get().strip(), solar_html=solar_html_list, solar_wind_json=solar_wind_json_var.get().strip(), goes_json=goes_json_var.get().strip(), bz_smooth_minutes=int(bz_smooth_var.get().strip()), goes_smooth_minutes=int(goes_smooth_var.get().strip()), ) input_path = Path(args_like.csv_file).expanduser().resolve() outdir = Path(args_like.outdir).expanduser().resolve() outdir.mkdir(parents=True, exist_ok=True) status.set("Loading CSV...") root.update_idletasks() df = load_csv(input_path) status.set("Filtering selected range...") root.update_idletasks() filtered = filter_wspr_band_timerange( df, args_like.start_utc, args_like.end_utc, args_like.band_min_hz, args_like.band_max_hz, ) start_ts = parse_utc_datetime(args_like.start_utc) end_ts = parse_utc_datetime(args_like.end_utc) solar = None if args_like.solar_html or args_like.solar_wind_json or args_like.goes_json: status.set("Parsing solar context files...") root.update_idletasks() solar = merge_solar_sources( [Path(p).expanduser().resolve() for p in args_like.solar_html], Path(args_like.solar_wind_json).expanduser().resolve() if args_like.solar_wind_json else None, Path(args_like.goes_json).expanduser().resolve() if args_like.goes_json else None, ) solar = clip_solar_to_range(solar, start_ts, end_ts) start_tag = args_like.start_utc.replace(":", "").replace(" ", "_") end_tag = args_like.end_utc.replace(":", "").replace(" ", "_") base_name = f"{args_like.band_label}_{start_tag}_to_{end_tag}" status.set("Running main analysis...") root.update_idletasks() run_one_analysis( filtered, outdir, base_name=base_name, args_like=args_like, title_suffix=( f"{args_like.band_label} WSPR reports | " f"{args_like.start_utc} to {args_like.end_utc} UTC" ), input_path=input_path, solar=solar, ) if args_like.top_rx > 0: reps = reporters_table(filtered) chosen = reps.head(args_like.top_rx)["rxCall"].tolist() top_df = filtered[filtered["rxCall"].isin(chosen)].copy() status.set("Running top-RX subset analysis...") root.update_idletasks() run_one_analysis( top_df, outdir, base_name=f"{base_name}_top_{args_like.top_rx}_rx", args_like=args_like, title_suffix=( f"Top {args_like.top_rx} reporters only | " f"{args_like.band_label} | " f"{args_like.start_utc} to {args_like.end_utc} UTC" ), input_path=input_path, solar=solar, ) status.set(f"Done. Outputs in: {outdir}") messagebox.showinfo( "Analysis complete", f"Outputs written to:\n{outdir}" ) except Exception as exc: status.set("Error.") messagebox.showerror("Error", str(exc)) tk.Button(root, text="Run analysis", command=run_analysis).grid( row=row, column=0, columnspan=3, pady=10 ) row += 1 tk.Label(root, textvariable=status, anchor="w", fg="blue").grid( row=row, column=0, columnspan=3, sticky="w", padx=6, pady=4 ) root.columnconfigure(1, weight=1) root.mainloop() # --------------------------------------------------------------------- # Main # --------------------------------------------------------------------- def main() -> int: args = parse_args() if args.gui: launch_gui() return 0 input_path = Path(args.csv_file).expanduser().resolve() outdir = Path(args.outdir).expanduser().resolve() outdir.mkdir(parents=True, exist_ok=True) try: df = load_csv(input_path) filtered = filter_wspr_band_timerange( df, args.start_utc, args.end_utc, args.band_min_hz, args.band_max_hz, ) except Exception as exc: print(f"Error: {exc}", file=sys.stderr) return 1 start_ts = parse_utc_datetime(args.start_utc) end_ts = parse_utc_datetime(args.end_utc) solar = None if args.solar_html or args.solar_wind_json or args.goes_json: solar = merge_solar_sources( [Path(p).expanduser().resolve() for p in args.solar_html], Path(args.solar_wind_json).expanduser().resolve() if args.solar_wind_json else None, Path(args.goes_json).expanduser().resolve() if args.goes_json else None, ) solar = clip_solar_to_range(solar, start_ts, end_ts) start_tag = args.start_utc.replace(":", "").replace(" ", "_") end_tag = args.end_utc.replace(":", "").replace(" ", "_") base_name = f"{args.band_label}_{start_tag}_to_{end_tag}" norm_tag = " | per-RX normalised" if USE_PER_RX_NORMALISATION else "" run_one_analysis( filtered, outdir, base_name=base_name, args_like=args, title_suffix=( f"{args.band_label} WSPR reports | " f"{args.start_utc} to {args.end_utc} UTC" ), input_path=input_path, solar=solar, ) if args.top_rx and args.top_rx > 0: norm_tag = " | per-RX normalised" if USE_PER_RX_NORMALISATION else "" reps = reporters_table(filtered) chosen = reps.head(args.top_rx)["rxCall"].tolist() top_df = filtered[filtered["rxCall"].isin(chosen)].copy() run_one_analysis( top_df, outdir, base_name=f"{base_name}_top_{args.top_rx}_rx", args_like=args, title_suffix=( f"Top {args.top_rx} reporters only | " f"{args.band_label} | " f"{args.start_utc} to {args.end_utc} UTC" f"{norm_tag}" ), input_path=input_path, solar=solar, ) return 0 if __name__ == "__main__": raise SystemExit(main())