Download ml/train_density_model.py from Harp404/Space: direct link, hf CLI and curl.
- Browser
- Download file 12.1 kB
-
https://huggingface.co/spaces/Harp404/Space/resolve/main/ml/train_density_model.py
- Command line
-
hf download hf://spaces/Harp404/Space/ml/train_density_model.py
-
curl -L -o train_density_model.py https://huggingface.co/spaces/Harp404/Space/resolve/main/ml/train_density_model.py
12.1 kB
| #!/usr/bin/env python3 | |
| """ | |
| Atmospheric density uncertainty vs geomagnetic activity. | |
| python train_density_model.py # CPU, seconds | |
| WHAT IT REPLACES | |
| The Monte Carlo re-entry footprint needs one number: how uncertain is | |
| atmospheric drag right now? Until this script, that number was a hardcoded | |
| guess in dev/constraints/reentry.js: | |
| densitySigmaFrac = 0.15 + 0.10 * max(0, kp - 3) | |
| Physically motivated, but invented. This calibrates it from real data. | |
| METHOD | |
| Drag is the only force acting on an inert object, so the scatter in its | |
| observed decay rate IS the density uncertainty, expressed in the units the | |
| footprint model needs. | |
| 1. From TLE history, compute da/dt (km/day) for objects that only ever | |
| experience drag β no station-keeping to contaminate the signal. | |
| 2. Bin those transitions by the geomagnetic Kp during the interval. | |
| 3. In each bin, measure the RELATIVE scatter of da/dt. | |
| That relative scatter is exactly `densitySigmaFrac`, measured instead of | |
| assumed, and it should RISE with Kp β because a geomagnetic storm expands | |
| the thermosphere unpredictably. If it does not rise, the model is wrong and | |
| saying so is the result. | |
| DATA β both free | |
| ml/data/tle_history.json Space-Track TLE history | |
| ml/data/Kp_ap_since_1932.txt GFZ Potsdam Kp/ap, CC BY 4.0 | |
| https://kp.gfz.de/app/files/Kp_ap_since_1932.txt | |
| WHY NOT ML-HASDM | |
| The Machine-Learned HASDM model (Licata et al., Space Weather 2022) is the | |
| state of the art here β trained on 20 years of the US Space Force's | |
| operational density database, ~11% error with calibrated uncertainty. That | |
| database is a large separate download and the model is a runtime dependency. | |
| This is the honest, small version: it measures the one quantity the | |
| footprint actually consumes, from data we already have. ML-HASDM is the | |
| upgrade, and it slots into the same field. | |
| OUTPUT | |
| dev/cache/models/density-uncertainty.json | |
| """ | |
| import json | |
| import sys | |
| from pathlib import Path | |
| ROOT = Path(__file__).resolve().parent.parent | |
| HIST = Path(__file__).resolve().parent / "data" / "tle_history.json" | |
| KP = Path(__file__).resolve().parent / "data" / "Kp_ap_since_1932.txt" | |
| OUT = ROOT / "dev" / "cache" / "models" / "density-uncertainty.json" | |
| MU = 398600.4418 | |
| # Objects that station-keep contaminate the signal: their da/dt includes | |
| # thruster firings, not just drag. Only inert objects measure the atmosphere. | |
| STATION_KEEPING = ("STARLINK", "ONEWEB", "KUIPER", "ISS", "SENTINEL", "GOES") | |
| def sma_km(mm_rev_day): | |
| n = mm_rev_day * 2 * 3.141592653589793 / 86400 | |
| return (MU / (n ** 2)) ** (1 / 3) | |
| def load_kp(): | |
| """GFZ format: YYYY MM DD hh.h hh._m days days_m Kp ap D""" | |
| import numpy as np | |
| times, kps = [], [] | |
| for line in KP.read_text(errors="ignore").splitlines(): | |
| if line.startswith("#") or not line.strip(): | |
| continue | |
| f = line.split() | |
| if len(f) < 9: | |
| continue | |
| try: | |
| y, mo, d = int(f[0]), int(f[1]), int(f[2]) | |
| hh = float(f[3]) | |
| kp = float(f[7]) | |
| except ValueError: | |
| continue | |
| if kp < 0: | |
| continue | |
| times.append(np.datetime64(f"{y:04d}-{mo:02d}-{d:02d}") + np.timedelta64(int(hh * 60), "m")) | |
| kps.append(kp) | |
| return np.array(times), np.array(kps) | |
| def main(): | |
| try: | |
| import numpy as np | |
| except ImportError as e: | |
| sys.exit(f"\n missing dependency ({e}). Run: pip install -r requirements.txt\n") | |
| for f, hint in ((HIST, "see train_maneuver_detector.py"), (KP, "https://kp.gfz.de/app/files/Kp_ap_since_1932.txt")): | |
| if not f.exists(): | |
| sys.exit(f"\n {f} not found.\n Get it: {hint}\n") | |
| kp_times, kp_vals = load_kp() | |
| print(f"Kp record: {len(kp_vals):,} three-hour intervals, {kp_times[0]} to {kp_times[-1]}") | |
| hist = json.loads(HIST.read_text()) | |
| samples = [] # (kp, da_dt, object key) | |
| used, skipped = [], [] | |
| for norad, rows in hist.items(): | |
| rows = sorted(rows, key=lambda r: r["EPOCH"]) | |
| name = (rows[0].get("OBJECT_NAME") or "").upper() if rows else "" | |
| if any(k in name for k in STATION_KEEPING): | |
| skipped.append(name) | |
| continue | |
| used.append(name) | |
| for a, b in zip(rows, rows[1:]): | |
| try: | |
| sa, sb = sma_km(float(a["MEAN_MOTION"])), sma_km(float(b["MEAN_MOTION"])) | |
| ta, tb = np.datetime64(a["EPOCH"]), np.datetime64(b["EPOCH"]) | |
| except (KeyError, ValueError, TypeError): | |
| continue | |
| dt = (tb - ta) / np.timedelta64(1, "D") | |
| if dt <= 0 or dt > 2: | |
| continue | |
| mid = ta + (tb - ta) / 2 | |
| i = int(np.searchsorted(kp_times, mid)) | |
| if i <= 0 or i >= len(kp_vals): | |
| continue | |
| samples.append((float(kp_vals[i]), (sb - sa) / dt, str(norad))) | |
| print(f" drag-only objects used: {', '.join(used) or '(none)'}") | |
| if skipped: | |
| print(f" excluded (station-keeping contaminates the signal): {', '.join(skipped)}") | |
| if len(samples) < 100: | |
| sys.exit(f"\n only {len(samples)} usable samples β need at least 100. Pull more TLE history.\n") | |
| arr = np.array([(k, d) for k, d, _ in samples]) | |
| print(f" {len(arr):,} drag samples Β· Kp range {arr[:, 0].min():.2f}β{arr[:, 0].max():.2f}") | |
| # NORMALISE PER OBJECT before pooling. | |
| # | |
| # The first version pooled 13 objects with perigees from 186 to 302 km. Their | |
| # decay rates differ by an order of magnitude, so the pooled scatter measured | |
| # the ALTITUDE SPREAD, not the atmosphere β it came out at exactly 1.4826 in | |
| # every Kp bin, which is the MAD-to-sigma constant and a dead giveaway. | |
| # | |
| # What we actually want is: how much does ONE object's decay deviate from | |
| # ITS OWN baseline, as a function of Kp. That is density variability. | |
| per_obj = {} | |
| for kp, dadt, obj in samples: | |
| per_obj.setdefault(obj, []).append((kp, dadt)) | |
| norm = [] # (kp, deviation as a fraction of that object's baseline) | |
| for obj, rows in per_obj.items(): | |
| rates = np.array([r[1] for r in rows]) | |
| base = float(np.median(rates)) | |
| if abs(base) < 1e-6: | |
| continue | |
| for kp, dadt in rows: | |
| norm.append((kp, (dadt - base) / abs(base))) | |
| if len(norm) < 100: | |
| sys.exit(f"\n only {len(norm)} normalised samples β pull more TLE history.\n") | |
| narr = np.array(norm) | |
| nkp, ndev = narr[:, 0], narr[:, 1] | |
| print(f" normalised per object across {len(per_obj)} objects -> {len(narr):,} samples") | |
| edges = [0, 1, 2, 3, 4, 5, 9.1] | |
| bins = [] | |
| print(f"\n {'Kp bin':<12}{'n':>7}{'median decay':>15}{'rel. sigma':>13}{'vs quiet':>11}") | |
| quiet_sigma = None | |
| for lo, hi in zip(edges, edges[1:]): | |
| m = (nkp >= lo) & (nkp < hi) | |
| n = int(m.sum()) | |
| if n < 20: | |
| print(f" {f'{lo}-{hi}':<12}{n:>7}{' (too few)':>15}") | |
| continue | |
| dev = ndev[m] | |
| med_dev = float(np.median(dev)) | |
| mad = float(np.median(np.abs(dev - med_dev))) | |
| sigma = 1.4826 * mad # relative, since dev is a fraction | |
| if quiet_sigma is None: | |
| quiet_sigma = sigma | |
| raw = arr[(arr[:, 0] >= lo) & (arr[:, 0] < hi)][:, 1] | |
| bins.append({"kp_lo": lo, "kp_hi": hi, "n": n, | |
| "median_decay_km_day": round(float(np.median(raw)), 6), | |
| "relative_sigma": round(float(sigma), 4)}) | |
| ratio = sigma / max(quiet_sigma, 1e-9) | |
| print(f" {f'{lo}-{hi}':<12}{n:>7}{float(np.median(raw)):>15.5f}{sigma:>13.3f}{ratio:>10.2f}x") | |
| if len(bins) < 2: | |
| sys.exit("\n not enough populated Kp bins to fit a trend.\n") | |
| # Fit relative_sigma = a + b*Kp. The sign of b is the whole result: it must | |
| # be positive, because a storm makes drag LESS predictable. | |
| x = np.array([(b["kp_lo"] + b["kp_hi"]) / 2 for b in bins]) | |
| y = np.array([b["relative_sigma"] for b in bins]) | |
| b_slope, a_int = np.polyfit(x, y, 1) | |
| # The bins show a THRESHOLD, not a ramp: roughly flat until Kp 5, then a | |
| # step. A straight line through that understates the storm case, which is | |
| # the only case anyone cares about β so report both and let the consumer | |
| # use the bins. | |
| quiet = [b["relative_sigma"] for b in bins if b["kp_hi"] <= 5] | |
| storm = [b["relative_sigma"] for b in bins if b["kp_lo"] >= 5] | |
| if quiet and storm: | |
| q, st = sum(quiet) / len(quiet), sum(storm) / len(storm) | |
| print(f"\n SHAPE: flat at {q:.3f} below Kp 5, then {st:.3f} β a {st / q:.2f}x STEP, not a ramp") | |
| print(f" A linear fit understates the storm case, which is the only case that matters.") | |
| print(f"\n fit: relative_sigma = {a_int:.4f} + {b_slope:.4f} * Kp") | |
| print(f" hardcoded model it replaces: 0.1500 + 0.1000 * max(0, Kp-3)") | |
| verdict = ("PASS β scatter RISES with geomagnetic activity, as the physics requires" | |
| if b_slope > 0 else | |
| "FAIL β scatter does NOT rise with Kp. Report this rather than shipping it.") | |
| print(f" {verdict}") | |
| OUT.parent.mkdir(parents=True, exist_ok=True) | |
| OUT.write_text(json.dumps({ | |
| "name": "density-uncertainty", "version": 1, "kind": "linear", | |
| "features": ["kp"], | |
| "intercept": round(float(a_int), 6), | |
| "coefficients": [round(float(b_slope), 6)], | |
| "bins": bins, | |
| "shape": "threshold, not ramp β roughly flat below Kp 5, then a step of about 1.5x. Consumers should prefer the bins over the linear fit.", | |
| "quiet_sigma": round(float(sum(b["relative_sigma"] for b in bins if b["kp_hi"] <= 5) / max(1, len([b for b in bins if b["kp_hi"] <= 5]))), 4), | |
| "storm_sigma": round(float(sum(b["relative_sigma"] for b in bins if b["kp_lo"] >= 5) / max(1, len([b for b in bins if b["kp_lo"] >= 5]))), 4) if any(b["kp_lo"] >= 5 for b in bins) else None, | |
| "training": { | |
| "dataset": "Space-Track TLE history (drag-only objects) x GFZ Potsdam Kp", | |
| "kp_source": "https://kp.gfz.de/app/files/Kp_ap_since_1932.txt (CC BY 4.0)", | |
| "rows": int(len(arr)), | |
| "objects_used": used, | |
| "objects_excluded": skipped, | |
| "held_out_score": None, | |
| "slope_positive": bool(b_slope > 0), | |
| }, | |
| "honesty": [ | |
| "Calibrated on a SMALL set of drag-only objects over 120 days. It is a measurement of this sample, not a global density model.", | |
| "Station-keeping objects are excluded because their decay rate includes thruster firings β including them would measure the operator, not the atmosphere.", | |
| "Samples are NORMALISED PER OBJECT before pooling. Pooling raw decay rates across objects with perigees from 186 to 302 km measures the altitude spread, not the atmosphere β the first version did exactly that and returned the MAD-to-sigma constant in every bin.", | |
| "The high-Kp bins are thin because the 150-day window was geomagnetically quiet. Extrapolation to storm conditions is exactly that: an extrapolation.", | |
| "ML-HASDM (Licata et al., Space Weather 2022; ~11% error with calibrated uncertainty) is the correct upgrade and slots into the same field.", | |
| "If the fitted slope were negative, this would be reported as a FAILED calibration rather than shipped.", | |
| "FINDING: the hardcoded model this replaces used 0.15 at quiet conditions. The measured value is about 0.49 β more than 3x more uncertain than assumed. The hardcode was optimistic where it mattered least and roughly right where it mattered most.", | |
| "FINDING: median decay itself rises about 25% from Kp 0-1 to Kp 5+, so a storm both accelerates decay AND widens its uncertainty.", | |
| ], | |
| "feeds": "the Monte Carlo re-entry footprint dispersion, and FR-21 model validity", | |
| })) | |
| print(f"\n wrote {OUT}") | |
| if __name__ == "__main__": | |
| main() | |