#!/usr/bin/env python3 from __future__ import annotations import argparse import sys from pathlib import Path import tkinter as tk from tkinter import filedialog, messagebox import matplotlib.pyplot as plt import pandas as pd 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 DIST_NEAR_MAX_KM = 400 DIST_MID_MAX_KM = 1500 # --------------------------------------------------------------------- # 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. " "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)", ) 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 # --------------------------------------------------------------------- # 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 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 # --------------------------------------------------------------------- # Plotting / outputs # --------------------------------------------------------------------- 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 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, ) -> 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\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\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, ) -> 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" 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, ) write_summary( summary_path, input_path.resolve(), filtered, windowed, stats, reporters, reference_hz, args_like, title_suffix, ) 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"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}") 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, ) -> 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 def launch_gui() -> None: root = tk.Tk() root.title("WSPR Mode Bias Explorer") 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") row = 0 def add_label_entry(label: str, var: tk.StringVar, width: int = 40) -> 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) 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=40).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=40).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) status = tk.StringVar(value="Ready.") def run_analysis() -> None: try: 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(), ) 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_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, ) 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, ) 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_tag = args.start_utc.replace(":", "").replace(" ", "_") end_tag = args.end_utc.replace(":", "").replace(" ", "_") base_name = f"{args.band_label}_{start_tag}_to_{end_tag}" 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, ) if args.top_rx and args.top_rx > 0: 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" ), input_path=input_path, ) return 0 if __name__ == "__main__": raise SystemExit(main()) reference