ELEV
Elevation Oncology, Inc. Common stock
stockNASDAQ

AST SPACE MOBILE \#!/usr/bin/env python3 \# -\*- coding: utf-8 -\*- """ AETHERWORKS G3X × ORBITALHUD™ — Unified Ops Pipeline ======================================================================== Integrates: 1. BlueWalker-3 (BW3) ephemeris + multi-site pass generation (Skyfield) 2. OpsMaster pass log consolidation (unify multiple sources) 3. SAFE\_MODE checks & physics validation (PROMETHEOS) 4. BB2 Block-2 adaptation simulation (thermal/RF/antenna/HARQ) Pipeline flow: → Load TLE & compute satellite passes for multiple sites (CSV) → Unify all pass logs into single DataFrame → Run SAFE\_MODE trigger checks & compute KPIs → Simulate BB2 Block-2 operations (300–600 s nominal) → Emit audit logs, summary JSON, acceptance gates, optional charts Usage: python3 bw3\_bb2\_unified\_ops.py --days 3 --sites LON-ON-01 OXF-UK-01 --bb2-secs 300 --plots No external deps beyond: skyfield, pandas, numpy, openpyxl (for Excel I/O, optional). For matplotlib charts, install matplotlib (optional --plots flag). Author: Aetherworks Ops Date: 2025-08-26 """ from \_\_future\_\_ import annotations import argparse import csv import json import math import os import random import statistics import sys import time import hashlib import hmac import secrets from collections import Counter from dataclasses import asdict, dataclass, fieldfrom datetime import datetime, timedelta, timezone from pathlib import Path from typing import Dict, List, Optional, Tuple import numpy as np import pandas as pd try: from skyfield.api import EarthSatellite, Topos, load, wgs84 SKYFIELD\_OK = True except ImportError: SKYFIELD\_OK = False print("\[WARN\] skyfield not installed; pass generation disabled", file=sys.stderr) try: from openpyxl import load\_workbook OPENPYXL\_OK = True except ImportError: OPENPYXL\_OK = False \# ============================================================================ \# ◀ PART 1: SKYFIELD EPHEMERIS & MULTI-SITE PASS GENERATION \# ============================================================================ class BW3PassGenerator: """ BW3 multi-site pass predictor using Skyfield. Emits per-site CSV + consolidated DataFrame. """ \# TLE for BlueWalker-3 (as of 2025-08-26) TLE\_NAME = "BlueWalker-3 (53807)" TLE\_LINE1 = "1 53807U 22111AL 25239.00000000 .00000000 00000-0 00000-0 0 9991" TLE\_LINE2 = "2 53807 53.2000 000.0000 0007240 000.0000 000.0000 15.33882700 01" \# RF metrics C = 299792.458 # km/s FREQS = {"UHF\_437": 437.5e6, "Sband\_2210": 2210e6} def \_\_init\_\_(self, el\_min\_deg: float = 10.0, days: int = 3, step\_s: int = 20): self.el\_min\_deg = el\_min\_deg self.days = days self.step\_s = step\_s if SKYFIELD\_OK: self.ts = load.timescale() self.sat = EarthSatellite(self.TLE\_LINE1, self.TLE\_LINE2, self.TLE\_NAME, self.ts) else:self.ts = None self.sat = None @staticmethod def fspl\_db(range\_km: float, freq\_hz: float) -> float: c = 299792.458e3 # m/s return 20.0 \* math.log10(4.0 \* math.pi \* (range\_km \* 1000.0) \* freq\_hz / c) @staticmethod def doppler\_hz(freq\_hz: float, vr\_km\_s: float) -> float: c = 299792.458 # km/s return freq\_hz \* (vr\_km\_s / c) def find\_passes\_for\_site( self, site\_code: str, lat\_deg: float, lon\_deg: float, elev\_m: float ) -> pd.DataFrame: """Compute all passes above el\_min for a site over \[today, today+days).""" if not SKYFIELD\_OK or self.sat is None: return pd.DataFrame() start\_utc = datetime.utcnow().replace(second=0, microsecond=0, tzinfo=timezone.utc) end\_utc = start\_utc + timedelta(days=self.days) \# Build minute-resolution time array minutes = int((end\_utc - start\_utc).total\_seconds() / 60) times = self.ts.utc( \[start\_utc.year\] \* minutes, \[start\_utc.month\] \* minutes, \[start\_utc.day\] \* minutes, \[start\_utc.hour + (i // 60) for i in range(minutes)\], \[start\_utc.minute + (i % 60) for i in range(minutes)\], ) "e() \# ============================================================================ \# ◀ PART 2: UNIFIED PASS LOG CONSOLIDATION & SAFE\_MODE CHECKS \# ============================================================================ def wilson\_lb(successes: int, trials: int, conf: float = 0.95) -> Optional\[float\]: """Wilson score lower bound for binomial confidence interval.""" if trials <= 0 or successes < 0 or successes > trials: return None z = statistics.NormalDist().inv\_cdf(0.5 + conf / 2) n = float(trials) phat = successes / n denom = 1.0 + (z \* z) / n centre = phat + (z \* z) / (2.0 \* n) margin = z \* math.sqrt((phat \* (1.0 - phat) + (z \* z) / (4.0 \* n)) / n) return (centre - margin) / denomdef run\_sigma(series: pd.Series) -> Optional\[float\]: """Compute population stddev if enough samples.""" x = series.dropna() if len(x) < 5: return None return statistics.pstdev(x) def check\_safe\_mode\_triggers(row: pd.Series) -> Optional\[Dict\[str, float\]\]: """SAFE\_MODE trigger detection per row.""" THRESHOLDS = { "fspl\_max": 160.0, "doppler\_max": 5000.0, "wilson\_min": 0.80, "sigma\_max": 0.015, } triggers = {} if pd.notna(row.get("FSPL\_dB\_num")) and row\["FSPL\_dB\_num"\] > THRESHOLDS\["fspl\_max"\]: triggers\["FSPL"\] = row\["FSPL\_dB\_num"\] if pd.notna(row.get("doppler\_abs\_hz")) and abs(row\["doppler\_abs\_hz"\]) > THRESHOLDS\["doppl triggers\["Doppler"\] = row\["doppler\_abs\_hz"\] if row.get("wilson\_lb\_95") is not None and row\["wilson\_lb\_95"\] < THRESHOLDS\["wilson\_min"\] triggers\["Wilson"\] = row\["wilson\_lb\_95"\] if pd.notna(row.get("sigma\_run")) and row\["sigma\_run"\] > THRESHOLDS\["sigma\_max"\]: triggers\["SPC"\] = row\["sigma\_run"\] return triggers if triggers else None def unify\_pass\_logs(dataframes: List\[pd.DataFrame\]) -> pd.DataFrame: """Merge multiple pass logs, apply numeric coercion & SAFE\_MODE checks.""" if not dataframes: return pd.DataFrame() full = pd.concat(dataframes, ignore\_index=True) \# Numeric coercion candidates = \["Frames\_n", "Decodes\_k", "Net\_dB"\] candidates += \[c for c in full.columns if "FSPL" in c.upper() or "DOPPLER" in c.upper()\] for col in candidates: if col in full.columns: full\[col + "\_num"\] = pd.to\_numeric(full\[col\], errors="coerce") \# Wilson lower bound if {"Frames\_n\_num", "Decodes\_k\_num"} <= set(full.columns): full\["wilson\_lb\_95"\] = full.apply( lambda r: wilson\_lb(int(r.Decodes\_k\_num) if pd.notna(r.Decodes\_k\_num) else 0,int(r.Frames\_n\_num) if pd.notna(r.Frames\_n\_num) else 0, if pd.notna(r.Decodes\_k\_num) and pd.notna(r.Frames\_n\_num) else None, axis=1, 0.95) ) \# Sigma per run if "Net\_dB\_num" in full.columns and "RunID" in full.columns: sigma\_map = full.groupby("RunID")\["Net\_dB\_num"\].agg(run\_sigma) full\["sigma\_run"\] = full\["RunID"\].map(sigma\_map) \# Doppler absolute if "Doppler\_min\_Hz\_num" in full.columns and "Doppler\_max\_Hz\_num" in full.columns: full\["doppler\_abs\_hz"\] = ( full\[\["Doppler\_min\_Hz\_num", "Doppler\_max\_Hz\_num"\]\].abs().max(axis=1) ) elif "Doppler\_Hz\_num" in full.columns: full\["doppler\_abs\_hz"\] = full\["Doppler\_Hz\_num"\].abs() \# FSPL canonical fspl\_col = next((c for c in full.columns if "FSPL" in c.upper()), None) if fspl\_col and fspl\_col + "\_num" in full.columns: full\["FSPL\_dB\_num"\] = full\[fspl\_col + "\_num"\] \# SAFE\_MODE triggers full\["SAFE\_MODE\_Triggers"\] = full.apply(check\_safe\_mode\_triggers, axis=1) full\["SAFE\_MODE"\] = full\["SAFE\_MODE\_Triggers"\].apply(lambda x: bool(x)) return full \# ============================================================================ \# ◀ PART 3: BB2 BLOCK-2 ADAPTATION SIMULATION (THERMAL/RF/ANTENNA/HARQ) \# ============================================================================ \# ---- BB2 CONSTANTS ---- SNR\_TARGET\_DB = 5.0 SNR\_RECOVER\_DB = 7.0 DUTY\_MIN\_PCT = 1.0 DUTY\_MAX\_PCT = 5.0 TX\_EIRP\_DBM\_INIT = 9.5 SYMRATE\_KSYM\_INIT = 100.0 DT\_NOMINAL\_LIMIT = 1.4 DT\_STRESSED\_LIMIT = 2.1 DT\_SOFT\_BACKOFF = 1.8 DT\_SAFE\_COAST = 2.2 THERM\_GAIN = 0.00225THERM\_LOSS = 0.00160 THERM\_AMBIENT\_DRIFT = 0.00005 EPS\_R = 3.1 F0\_MHZ = 760.0 THERMAL\_DRIFT\_KHZ = 1.0 S11\_TARGET\_DB = -12.0 MODULATION = "pi/4-DQPSK" SYMRATE\_PROFILES = \[50, 75, 100, 150\] MTU\_BYTES = 2048 HARQ\_PROCESSES = 8 HARQ\_MAX\_RETX = 4 LOW\_ELEV\_DEG = 15.0 EVENT\_BEAMSW\_P = 0.010 EVENT\_LOWEL\_P = 0.018 EVENT\_DOPPLER\_P = 0.022 DOPPLER\_SNR\_PEN\_DB = (0.5, 1.0) LOW\_ELEV\_SNR\_PEN\_DB = (1.0, 2.0) BEAMSW\_SNR\_PEN\_DB = (2.0, 4.0) LATENCY\_RTT\_MS\_MIN = 50 LATENCY\_RTT\_MS\_MAX = 120 ATTEST\_STATIC\_SALT = "G3X\_ORBITALHUD\_F1\_Rev1.1" @dataclass class BB2Config: secs: int = 300 step\_s: int = 1 ambient\_hot: bool = False self.lut = self.\_build\_varactor\_lut() def \_build\_varactor\_lut(self) -> Dict\[int, Tuple\[float, \- self.cfg.therm\_loss \* st.deltaT \+ ambient ) st.deltaT = max(0.0, dT\_next) if st.deltaT > DT\_SOFT\_BACKOFF and st.tx\_eirp\_dbm > 0: st.tx\_eirp\_dbm \*= 0.85 if st.deltaT > DT\_SAFE\_COAST: st.tx\_eirp\_dbm = 0.0 st.symrate\_ksym = 0.0 st.duty\_pct = 0.0 st.event = "thermal\_critical" class BB2PHY: @staticmethod def ber\_pi4dqpsk(snr\_db: float) -> float: snr\_lin = db\_to\_lin(snr\_db) return max(qfunc(math.sqrt(2.0 \* snr\_lin)), 1e-9) @staticmethod def choose\_coderate(snr\_db: float) -> float: if snr\_db >= 9.0: return 0.75 elif snr\_db >= 6.5: return 0.667 , "