Mirror/intercept
1
0
Fork 0
mirror of https://github.com/smittix/intercept.git synced 2026-04-24 06:40:00 -07:00
Code Issues Packages Projects Releases Wiki Activity

Fix setup.sh hanging on Python 3.14/macOS and add satellite enhancements

Browse Source 
- Add --no-cache-dir and --timeout 120 to all pip calls to prevent hanging
  on corrupt/stale pip HTTP cache (cachecontrol .pyc issue)
- Replace silent python -c import verification with pip show to avoid
  import-time side effects hanging the installer
- Switch optional packages to --only-binary :all: to skip source compilation
  on Python versions without pre-built wheels (prevents gevent/numpy hangs)
- Warn early when Python 3.13+ is detected that some packages may be skipped
- Add ground track caching with 30-minute TTL to satellite route
- Add live satellite position tracker background thread via SSE fanout
- Add satellite_predict, satellite_telemetry, and satnogs utilities

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
James Smith
2026-03-18 11:09:00 +00:00
parent 3140f54419
commit dc84e933c1
9 changed files with 1497 additions and 440 deletions
Download Patch File Download Diff File Expand all files Collapse all files

277
routes/satellite.py
View File

@@ -3,13 +3,15 @@
from __future__ import annotations
import math
import time
import urllib.request
from datetime import datetime, timedelta
import requests
from flask import Blueprint, jsonify, render_template, request
from flask import Blueprint, Response, jsonify, render_template, request
from config import SHARED_OBSERVER_LOCATION_ENABLED
from config import DEFAULT_LATITUDE, DEFAULT_LONGITUDE, SHARED_OBSERVER_LOCATION_ENABLED
from utils.sse import sse_stream_fanout
from data.satellites import TLE_SATELLITES
from utils.database import (
add_tracked_satellite,
@@ -44,6 +46,11 @@ ALLOWED_TLE_HOSTS = ['celestrak.org', 'celestrak.com', 'www.celestrak.org', 'www
# Local TLE cache (can be updated via API)
_tle_cache = dict(TLE_SATELLITES)
# Ground track cache: key=(sat_name, tle_line1[:20]) -> (track_data, computed_at_timestamp)
# TTL is 1800 seconds (30 minutes)
_track_cache: dict = {}
_TRACK_CACHE_TTL = 1800
def _load_db_satellites_into_cache():
"""Load user-tracked satellites from DB into the TLE cache."""
@@ -64,6 +71,112 @@ def _load_db_satellites_into_cache():
logger.warning(f"Failed to load DB satellites into TLE cache: {e}")
def _start_satellite_tracker():
"""Background thread: push live satellite positions to satellite_queue every second."""
import app as app_module
try:
from skyfield.api import EarthSatellite, wgs84
except ImportError:
logger.warning("skyfield not installed; satellite tracker thread will not run")
return
ts = _get_timescale()
logger.info("Satellite tracker thread started")
while True:
try:
now = ts.now()
now_dt = now.utc_datetime()
obs_lat = DEFAULT_LATITUDE
obs_lon = DEFAULT_LONGITUDE
has_observer = (obs_lat != 0.0 or obs_lon != 0.0)
observer = wgs84.latlon(obs_lat, obs_lon) if has_observer else None
tracked = get_tracked_satellites(enabled_only=True)
positions = []
for sat_rec in tracked:
sat_name = sat_rec['name']
norad_id = sat_rec.get('norad_id', 0)
tle1 = sat_rec.get('tle_line1')
tle2 = sat_rec.get('tle_line2')
if not tle1 or not tle2:
# Fall back to TLE cache
cache_key = sat_name.replace(' ', '-').upper()
if cache_key not in _tle_cache:
continue
tle_entry = _tle_cache[cache_key]
tle1 = tle_entry[1]
tle2 = tle_entry[2]
try:
satellite = EarthSatellite(tle1, tle2, sat_name, ts)
geocentric = satellite.at(now)
subpoint = wgs84.subpoint(geocentric)
pos = {
'satellite': sat_name,
'norad_id': norad_id,
'lat': float(subpoint.latitude.degrees),
'lon': float(subpoint.longitude.degrees),
'altitude': float(geocentric.distance().km - 6371),
'visible': False,
}
if has_observer and observer is not None:
diff = satellite - observer
topocentric = diff.at(now)
alt, az, dist = topocentric.altaz()
pos['elevation'] = float(alt.degrees)
pos['azimuth'] = float(az.degrees)
pos['distance'] = float(dist.km)
pos['visible'] = bool(alt.degrees > 0)
# Ground track with caching (90 points, TTL 1800s)
cache_key_track = (sat_name, tle1[:20])
cached = _track_cache.get(cache_key_track)
if cached and (time.time() - cached[1]) < _TRACK_CACHE_TTL:
pos['groundTrack'] = cached[0]
else:
track = []
for minutes_offset in range(-45, 46, 1):
t_point = ts.utc(now_dt + timedelta(minutes=minutes_offset))
try:
geo = satellite.at(t_point)
sp = wgs84.subpoint(geo)
track.append({
'lat': float(sp.latitude.degrees),
'lon': float(sp.longitude.degrees),
'past': minutes_offset < 0,
})
except Exception:
continue
_track_cache[cache_key_track] = (track, time.time())
pos['groundTrack'] = track
positions.append(pos)
except Exception:
continue
if positions:
msg = {
'type': 'positions',
'positions': positions,
'timestamp': datetime.utcnow().isoformat(),
}
try:
app_module.satellite_queue.put_nowait(msg)
except Exception:
pass
except Exception as e:
logger.debug(f"Satellite tracker error: {e}")
time.sleep(1)
def init_tle_auto_refresh():
"""Initialize TLE auto-refresh. Called by app.py after initialization."""
import threading
@@ -81,6 +194,15 @@ def init_tle_auto_refresh():
threading.Timer(2.0, _auto_refresh_tle).start()
logger.info("TLE auto-refresh scheduled")
# Start live position tracker thread
tracker_thread = threading.Thread(
target=_start_satellite_tracker,
daemon=True,
name='satellite-tracker',
)
tracker_thread.start()
logger.info("Satellite tracker thread launched")
def _fetch_iss_realtime(observer_lat: float | None = None, observer_lon: float | None = None) -> dict | None:
"""
@@ -185,7 +307,6 @@ def satellite_dashboard():
def predict_passes():
"""Calculate satellite passes using skyfield."""
try:
from skyfield.almanac import find_discrete
from skyfield.api import EarthSatellite, wgs84
except ImportError:
return jsonify({
@@ -193,6 +314,8 @@ def predict_passes():
'message': 'skyfield library not installed. Run: pip install skyfield'
}), 503
from utils.satellite_predict import predict_passes as _predict_passes
data = request.json or {}
# Validate inputs
@@ -228,7 +351,6 @@ def predict_passes():
ts = _get_timescale()
observer = wgs84.latlon(lat, lon)
t0 = ts.now()
t1 = ts.utc(t0.utc_datetime() + timedelta(hours=hours))
@@ -237,97 +359,30 @@ def predict_passes():
continue
tle_data = _tle_cache[sat_name]
# Current position for map marker (computed once per satellite)
current_pos = None
try:
satellite = EarthSatellite(tle_data[1], tle_data[2], tle_data[0], ts)
geo = satellite.at(ts.now())
sp = wgs84.subpoint(geo)
current_pos = {
'lat': float(sp.latitude.degrees),
'lon': float(sp.longitude.degrees),
}
except Exception:
continue
pass
def above_horizon(t):
diff = satellite - observer
topocentric = diff.at(t)
alt, _, _ = topocentric.altaz()
return alt.degrees > 0
sat_passes = _predict_passes(tle_data, observer, ts, t0, t1, min_el=min_el)
for p in sat_passes:
p['satellite'] = sat_name
p['norad'] = name_to_norad.get(sat_name, 0)
p['color'] = colors.get(sat_name, '#00ff00')
if current_pos:
p['currentPos'] = current_pos
passes.extend(sat_passes)
above_horizon.step_days = 1/720
try:
times, events = find_discrete(t0, t1, above_horizon)
except Exception:
continue
i = 0
while i < len(times):
if i < len(events) and events[i]:
rise_time = times[i]
set_time = None
for j in range(i + 1, len(times)):
if not events[j]:
set_time = times[j]
i = j
break
if set_time is None:
i += 1
continue
trajectory = []
max_elevation = 0
num_points = 30
duration_seconds = (set_time.utc_datetime() - rise_time.utc_datetime()).total_seconds()
for k in range(num_points):
frac = k / (num_points - 1)
t_point = ts.utc(rise_time.utc_datetime() + timedelta(seconds=duration_seconds * frac))
diff = satellite - observer
topocentric = diff.at(t_point)
alt, az, _ = topocentric.altaz()
el = alt.degrees
azimuth = az.degrees
if el > max_elevation:
max_elevation = el
trajectory.append({'el': float(max(0, el)), 'az': float(azimuth)})
if max_elevation >= min_el:
duration_minutes = int(duration_seconds / 60)
ground_track = []
for k in range(60):
frac = k / 59
t_point = ts.utc(rise_time.utc_datetime() + timedelta(seconds=duration_seconds * frac))
geocentric = satellite.at(t_point)
subpoint = wgs84.subpoint(geocentric)
ground_track.append({
'lat': float(subpoint.latitude.degrees),
'lon': float(subpoint.longitude.degrees)
})
current_geo = satellite.at(ts.now())
current_subpoint = wgs84.subpoint(current_geo)
passes.append({
'satellite': sat_name,
'norad': name_to_norad.get(sat_name, 0),
'startTime': rise_time.utc_datetime().strftime('%Y-%m-%d %H:%M UTC'),
'startTimeISO': rise_time.utc_datetime().isoformat(),
'maxEl': float(round(max_elevation, 1)),
'duration': int(duration_minutes),
'trajectory': trajectory,
'groundTrack': ground_track,
'currentPos': {
'lat': float(current_subpoint.latitude.degrees),
'lon': float(current_subpoint.longitude.degrees)
},
'color': colors.get(sat_name, '#00ff00')
})
i += 1
passes.sort(key=lambda p: p['startTime'])
passes.sort(key=lambda p: p['startTimeISO'])
return jsonify({
'status': 'success',
@@ -458,6 +513,48 @@ def get_satellite_position():
})
@satellite_bp.route('/transmitters/<int:norad_id>')
def get_transmitters_endpoint(norad_id: int):
"""Return SatNOGS transmitter data for a satellite by NORAD ID."""
from utils.satnogs import get_transmitters
transmitters = get_transmitters(norad_id)
return jsonify({'status': 'success', 'norad_id': norad_id, 'transmitters': transmitters})
@satellite_bp.route('/parse-packet', methods=['POST'])
def parse_packet():
"""Parse a raw satellite telemetry packet (base64-encoded)."""
import base64
from utils.satellite_telemetry import auto_parse
data = request.json or {}
try:
raw_bytes = base64.b64decode(data.get('data', ''))
except Exception:
return api_error('Invalid base64 data', 400)
result = auto_parse(raw_bytes)
return jsonify({'status': 'success', 'parsed': result})
@satellite_bp.route('/stream_satellite')
def stream_satellite() -> Response:
"""SSE endpoint streaming live satellite positions from the background tracker."""
import app as app_module
response = Response(
sse_stream_fanout(
source_queue=app_module.satellite_queue,
channel_key='satellite',
timeout=1.0,
keepalive_interval=30.0,
),
mimetype='text/event-stream',
)
response.headers['Cache-Control'] = 'no-cache'
response.headers['X-Accel-Buffering'] = 'no'
response.headers['Connection'] = 'keep-alive'
return response
def refresh_tle_data() -> list:
"""
Refresh TLE data from CelesTrak.

34
setup.sh
View File

@@ -487,6 +487,16 @@ import sys
raise SystemExit(0 if sys.version_info >= (3,9) else 1)
PY
ok "Python version OK (>= 3.9)"
# Python 3.13+ warning: some packages (gevent, numpy, scipy) may not have
# pre-built wheels yet and will be skipped to avoid hanging on compilation.
python3 - <<'PY'
import sys
raise SystemExit(0 if sys.version_info >= (3,13) else 1)
PY
if [[ $? -eq 0 ]]; then
warn "Python 3.13+ detected: optional packages without pre-built wheels will be skipped (--prefer-binary)."
fi
}
install_python_deps() {
@@ -520,8 +530,11 @@ install_python_deps() {
source venv/bin/activate
local PIP="venv/bin/python -m pip"
local PY="venv/bin/python"
# --no-cache-dir avoids pip hanging on a corrupt/stale HTTP cache (cachecontrol .pyc issue)
# --timeout prevents pip from hanging indefinitely on slow/unresponsive PyPI connections
local PIP_OPTS="--no-cache-dir --timeout 120"
if ! $PIP install --upgrade pip setuptools wheel; then
if ! $PIP install $PIP_OPTS --upgrade pip setuptools wheel; then
warn "pip/setuptools/wheel upgrade failed - continuing with existing versions"
else
ok "Upgraded pip tooling"
@@ -530,16 +543,18 @@ install_python_deps() {
progress "Installing Python dependencies"
info "Installing core packages..."
$PIP install "flask>=3.0.0" "flask-wtf>=1.2.0" "flask-compress>=1.15" \
$PIP install $PIP_OPTS "flask>=3.0.0" "flask-wtf>=1.2.0" "flask-compress>=1.15" \
"flask-limiter>=2.5.4" "requests>=2.28.0" \
"Werkzeug>=3.1.5" "pyserial>=3.5" || true
# Verify core packages are importable from the venv (not user site-packages)
$PY -s -c "import flask; import requests; from flask_limiter import Limiter; import flask_compress; import flask_wtf" 2>/dev/null || {
fail "Critical Python packages (flask, requests, flask-limiter, flask-compress, flask-wtf) not installed"
echo "Try: venv/bin/pip install flask requests flask-limiter flask-compress flask-wtf"
exit 1
}
# Verify core packages are installed by checking pip's reported list (avoids hanging imports)
for core_pkg in flask requests flask-limiter flask-compress flask-wtf; do
if ! $PIP show "$core_pkg" >/dev/null 2>&1; then
fail "Critical Python package not installed: ${core_pkg}"
echo "Try: venv/bin/pip install ${core_pkg}"
exit 1
fi
done
ok "Core Python packages installed"
info "Installing optional packages..."
@@ -549,7 +564,8 @@ install_python_deps() {
"gunicorn>=21.2.0" "gevent>=23.9.0" "psutil>=5.9.0"; do
pkg_name="${pkg%%>=*}"
info " Installing ${pkg_name}..."
if ! $PIP install "$pkg"; then
# --only-binary :all: skips packages with no pre-built wheel, preventing source compilation hangs
if ! $PIP install $PIP_OPTS --only-binary :all: "$pkg"; then
warn "${pkg_name} failed to install (optional - related features may be unavailable)"
fi
done

11
static/js/modes/weather-satellite.js
View File

@@ -302,6 +302,16 @@ const WeatherSat = (function() {
}
}
/**
* Pre-select a satellite without starting capture.
* Used by the satellite dashboard handoff so the user can review
* settings before hitting Start.
*/
function preSelect(satellite) {
const satSelect = document.getElementById('weatherSatSelect');
if (satSelect) satSelect.value = satellite;
}
/**
* Start capture for a specific pass
*/
@@ -1910,6 +1920,7 @@ const WeatherSat = (function() {
destroy,
start,
stop,
preSelect,
startPass,
selectPass,
testDecode,

69
templates/index.html
View File

@@ -4571,6 +4571,12 @@
if (satFrame && satFrame.contentWindow) {
satFrame.contentWindow.postMessage({type: 'satellite-visibility', visible: mode === 'satellite'}, '*');
}
// Weather-sat handoff: when switching away from satellite mode, clear any pending handoff banner
if (mode !== 'satellite' && mode !== 'weathersat') {
const existing = document.getElementById('weatherSatHandoffBanner');
if (existing) existing.remove();
}
if (aprsVisuals) aprsVisuals.style.display = mode === 'aprs' ? 'flex' : 'none';
if (tscmVisuals) tscmVisuals.style.display = mode === 'tscm' ? 'flex' : 'none';
if (spyStationsVisuals) spyStationsVisuals.style.display = mode === 'spystations' ? 'flex' : 'none';
@@ -16314,6 +16320,69 @@
if (typeof VoiceAlerts !== 'undefined') VoiceAlerts.init();
if (typeof KeyboardShortcuts !== 'undefined') KeyboardShortcuts.init();
});
// ── Weather-satellite handoff from the satellite dashboard iframe ─────────
window.addEventListener('message', (event) => {
if (!event.data || event.data.type !== 'weather-sat-handoff') return;
const { satellite, aosTime, tcaEl, duration } = event.data;
if (!satellite) return;
// Determine how far away the pass is
const aosMs = aosTime ? (new Date(aosTime) - Date.now()) : Infinity;
const minsAway = aosMs / 60000;
// Switch to weather-satellite mode and pre-select the satellite
switchMode('weathersat', { updateUrl: true }).then(() => {
if (typeof WeatherSat !== 'undefined') {
if (minsAway <= 2) {
// Pass is imminent — start immediately
WeatherSat.startPass(satellite);
showNotification('Weather Sat', `Auto-starting capture: ${satellite}`);
} else {
// Pre-select so the user can review settings and hit Start
WeatherSat.preSelect(satellite);
showHandoffBanner(satellite, minsAway, tcaEl, duration);
}
}
});
});
function showHandoffBanner(satellite, minsAway, tcaEl, duration) {
// Remove any existing banner
const existing = document.getElementById('weatherSatHandoffBanner');
if (existing) existing.remove();
const mins = Math.round(minsAway);
const elStr = tcaEl != null ? `${Number(tcaEl).toFixed(0)}°` : '?°';
const durStr = duration != null ? `${Math.round(duration)} min` : '';
const banner = document.createElement('div');
banner.id = 'weatherSatHandoffBanner';
banner.style.cssText = [
'position:fixed', 'top:60px', 'left:50%', 'transform:translateX(-50%)',
'background:rgba(0,20,30,0.95)', 'border:1px solid rgba(0,255,136,0.5)',
'color:#00ff88', 'font-family:var(--font-mono,monospace)', 'font-size:12px',
'padding:10px 18px', 'border-radius:6px', 'z-index:9999',
'display:flex', 'align-items:center', 'gap:12px',
'box-shadow:0 0 20px rgba(0,255,136,0.2)'
].join(';');
banner.innerHTML = `
<span>📡 <strong>${satellite}</strong> pass in <strong>${mins} min</strong> · max ${elStr}${durStr ? ' · ' + durStr : ''} — satellite pre-selected</span>
<button onclick="if(typeof WeatherSat!=='undefined')WeatherSat.start();this.closest('#weatherSatHandoffBanner').remove();"
style="background:rgba(0,255,136,0.2);border:1px solid rgba(0,255,136,0.5);color:#00ff88;padding:3px 10px;border-radius:4px;cursor:pointer;font-family:inherit;font-size:11px;">
Start Now
</button>
<button onclick="this.closest('#weatherSatHandoffBanner').remove();"
style="background:none;border:none;color:#666;cursor:pointer;font-size:14px;padding:0 4px;">✕</button>
`;
document.body.appendChild(banner);
// Auto-dismiss after 2 minutes
setTimeout(() => { if (banner.parentNode) banner.remove(); }, 120000);
}
</script>
</body>

411
templates/satellite_dashboard.html
View File

@@ -194,6 +194,32 @@
</div>
</div>
</div>
<!-- Transmitters -->
<div class="panel transmitters-panel">
<div class="panel-header">
<span>TRANSMITTERS <span id="txCount" style="color:var(--accent-cyan);"></span></span>
<div class="panel-indicator"></div>
</div>
<div class="panel-content" id="transmittersList">
<div style="text-align:center;color:var(--text-secondary);padding:15px;font-size:11px;">
Select a satellite to load transmitters
</div>
</div>
</div>
<!-- Decoded Packets -->
<div class="panel packets-panel">
<div class="panel-header">
<span>DECODED PACKETS <span id="packetCount" style="color:var(--accent-cyan);"></span></span>
<div class="panel-indicator"></div>
</div>
<div class="panel-content" id="packetList">
<div style="text-align:center;color:var(--text-secondary);padding:15px;font-size:11px;">
No packets received.<br>Packet decoding requires an AFSK/FSK decoder (coming soon).
</div>
</div>
</div>
</div>
<!-- Controls Bar -->
@@ -253,6 +279,75 @@
background: #ff4444;
box-shadow: 0 0 6px #ff4444;
}
/* Pass event row */
.pass-event-row {
display: flex;
justify-content: space-between;
align-items: center;
font-size: 10px;
color: var(--accent-cyan);
opacity: 0.75;
margin-top: 4px;
font-family: var(--font-mono);
}
.pass-capture-btn {
background: rgba(0, 255, 136, 0.12);
border: 1px solid rgba(0, 255, 136, 0.4);
color: var(--accent-green, #00ff88);
font-size: 10px;
font-family: var(--font-mono);
padding: 2px 7px;
border-radius: 3px;
cursor: pointer;
white-space: nowrap;
transition: background 0.15s;
}
.pass-capture-btn:hover {
background: rgba(0, 255, 136, 0.25);
}
/* Transmitters panel */
.transmitters-panel, .packets-panel {
margin-top: 10px;
}
.tx-item {
display: flex;
align-items: flex-start;
gap: 8px;
padding: 8px;
border-bottom: 1px solid rgba(0,212,255,0.08);
font-size: 11px;
}
.tx-item:last-child { border-bottom: none; }
.tx-inactive { opacity: 0.5; }
.tx-status-dot {
width: 8px;
height: 8px;
border-radius: 50%;
flex-shrink: 0;
margin-top: 3px;
}
.tx-body { flex: 1; min-width: 0; }
.tx-desc {
color: var(--text-primary);
font-weight: 500;
white-space: nowrap;
overflow: hidden;
text-overflow: ellipsis;
}
.tx-freq {
color: var(--accent-cyan);
font-family: var(--font-mono);
font-size: 10px;
margin-top: 2px;
}
.tx-uplink { color: var(--accent-green, #00ff88); }
.tx-service {
color: var(--text-muted, #556677);
font-size: 10px;
margin-top: 1px;
}
</style>
<script>
// Check if embedded mode
@@ -324,6 +419,7 @@
if (orbitTrack) { groundMap.removeLayer(orbitTrack); orbitTrack = null; }
}
loadTransmitters(selectedSatellite);
calculatePasses();
}
@@ -362,29 +458,111 @@
return false;
}
let positionPollingInterval = null;
let satelliteSSE = null;
function startPositionPolling() {
if (!positionPollingInterval) {
updateRealTimePositions();
positionPollingInterval = setInterval(updateRealTimePositions, 5000);
function startSSETracking() {
if (satelliteSSE) return;
satelliteSSE = new EventSource('/satellite/stream_satellite');
satelliteSSE.onmessage = (e) => {
try {
const msg = JSON.parse(e.data);
if (msg.type === 'positions') handleLivePositions(msg.positions);
} catch (_) {}
};
satelliteSSE.onerror = () => {
// Reconnect automatically after 5s
if (satelliteSSE) { satelliteSSE.close(); satelliteSSE = null; }
setTimeout(startSSETracking, 5000);
};
}
function stopSSETracking() {
if (satelliteSSE) { satelliteSSE.close(); satelliteSSE = null; }
}
function handleLivePositions(positions) {
// Find the selected satellite by name or norad_id
const satName = satellites[selectedSatellite]?.name;
const pos = positions.find(p =>
p.norad_id === selectedSatellite ||
p.satellite === satName ||
p.satellite === satellites[selectedSatellite]?.name
);
// Update visible count from all positions
const visibleCount = positions.filter(p => p.visible).length;
const visEl = document.getElementById('statVisible');
if (visEl) visEl.textContent = visibleCount;
if (!pos) return;
// Update telemetry panel
const telLat = document.getElementById('telLat');
const telLon = document.getElementById('telLon');
const telAlt = document.getElementById('telAlt');
const telEl = document.getElementById('telEl');
const telAz = document.getElementById('telAz');
const telDist = document.getElementById('telDist');
if (telLat) telLat.textContent = (pos.lat ?? 0).toFixed(4) + '°';
if (telLon) telLon.textContent = (pos.lon ?? 0).toFixed(4) + '°';
if (telAlt) telAlt.textContent = (pos.altitude ?? 0).toFixed(0) + ' km';
if (telEl) telEl.textContent = (pos.elevation ?? 0).toFixed(1) + '°';
if (telAz) telAz.textContent = (pos.azimuth ?? 0).toFixed(1) + '°';
if (telDist) telDist.textContent = (pos.distance ?? 0).toFixed(0) + ' km';
// Update live marker on map
if (groundMap && pos.lat != null && pos.lon != null) {
const satColor = satellites[selectedSatellite]?.color || '#00d4ff';
if (satMarker) groundMap.removeLayer(satMarker);
const satIcon = L.divIcon({
className: 'sat-marker-live',
html: `<div style="width:20px;height:20px;background:${satColor};border-radius:50%;border:3px solid #fff;box-shadow:0 0 20px ${satColor},0 0 40px ${satColor};"></div>`,
iconSize: [20, 20], iconAnchor: [10, 10]
});
satMarker = L.marker([pos.lat, pos.lon], { icon: satIcon }).addTo(groundMap);
}
// Update orbit track from groundTrack if available
if (groundMap && pos.groundTrack && pos.groundTrack.length > 1) {
if (orbitTrack) { groundMap.removeLayer(orbitTrack); orbitTrack = null; }
const satColor = satellites[selectedSatellite]?.color || '#00d4ff';
const segments = splitAtAntimeridian(pos.groundTrack);
orbitTrack = L.layerGroup();
segments.forEach(seg => {
const past = seg.filter(p => p.past);
const future = seg.filter(p => !p.past);
if (past.length > 1) L.polyline(past.map(p => [p.lat, p.lon]), { color: satColor, weight: 2, opacity: 0.4 }).addTo(orbitTrack);
if (future.length > 1) L.polyline(future.map(p => [p.lat, p.lon]), { color: satColor, weight: 2, opacity: 0.7, dashArray: '5, 5' }).addTo(orbitTrack);
});
orbitTrack.addTo(groundMap);
}
}
function stopPositionPolling() {
if (positionPollingInterval) {
clearInterval(positionPollingInterval);
positionPollingInterval = null;
function splitAtAntimeridian(track) {
const segments = [];
let current = [];
for (let i = 0; i < track.length; i++) {
const p = track[i];
if (current.length > 0) {
const prev = current[current.length - 1];
if ((prev.lon > 90 && p.lon < -90) || (prev.lon < -90 && p.lon > 90)) {
if (current.length >= 2) segments.push(current);
current = [];
}
}
current.push(p);
}
if (current.length >= 2) segments.push(current);
return segments;
}
// Listen for visibility messages from parent page (embedded mode)
window.addEventListener('message', (event) => {
if (event.data && event.data.type === 'satellite-visibility') {
if (event.data.visible) {
startPositionPolling();
startSSETracking();
} else {
stopPositionPolling();
stopSSETracking();
}
}
});
@@ -397,12 +575,13 @@
updateClock();
setInterval(updateClock, 1000);
setInterval(updateCountdown, 1000);
// In standalone mode, start polling immediately.
// In standalone mode, start SSE tracking immediately.
// In embedded mode, wait for parent to signal visibility.
if (!isEmbedded) {
startPositionPolling();
startSSETracking();
}
loadAgents();
loadTransmitters(selectedSatellite);
if (!usedShared) {
getLocation();
}
@@ -595,6 +774,12 @@
}
}
// Satellites that can be handed off to the weather-satellite capture mode
const WEATHER_SAT_KEYS = new Set([
'NOAA-15', 'NOAA-18', 'NOAA-19', 'NOAA-20', 'NOAA-21',
'METEOR-M2', 'METEOR-M2-3', 'METEOR-M2-4'
]);
function renderPassList() {
const container = document.getElementById('passList');
const countEl = document.getElementById('passCount');
@@ -610,7 +795,15 @@
container.innerHTML = passes.slice(0, 10).map((pass, idx) => {
const quality = pass.maxEl >= 60 ? 'excellent' : pass.maxEl >= 30 ? 'good' : 'fair';
const qualityText = pass.maxEl >= 60 ? 'EXCELLENT' : pass.maxEl >= 30 ? 'GOOD' : 'FAIR';
const time = pass.startTime.split(' ')[1] || pass.startTime;
const aosAz = pass.aosAz != null ? pass.aosAz.toFixed(0) + '°' : '--';
const tcaEl = pass.tcaEl != null ? pass.tcaEl.toFixed(0) + '°' : (pass.maxEl != null ? pass.maxEl.toFixed(0) + '°' : '--');
const tcaAz = pass.tcaAz != null ? pass.tcaAz.toFixed(0) + '°' : '--';
const losAz = pass.losAz != null ? pass.losAz.toFixed(0) + '°' : '--';
const timeStr = (pass.aosTime || pass.startTime || '').split('T')[1]?.substring(0, 5) || pass.startTime?.split(' ')[1] || '--:--';
const isWeatherSat = WEATHER_SAT_KEYS.has(pass.satellite);
const captureBtn = isWeatherSat
? `<button class="pass-capture-btn" onclick="event.stopPropagation(); handoffToWeatherSat(${idx})" title="Switch to Weather Satellite mode for this pass">→ Capture</button>`
: '';
return `
<div class="pass-item ${selectedPass === idx ? 'active' : ''}" onclick="selectPass(${idx})">
@@ -619,14 +812,39 @@
<span class="pass-quality ${quality}">${qualityText}</span>
</div>
<div class="pass-item-details">
<span class="pass-time">${time}</span>
<span>${pass.maxEl.toFixed(0)}° · ${pass.duration} min</span>
<span class="pass-time">${timeStr} UTC</span>
<span>${tcaEl} · ${pass.duration} min</span>
</div>
<div class="pass-event-row">
<span title="AOS azimuth">↑ ${aosAz}</span>
<span title="TCA azimuth">⊙ ${tcaAz}</span>
<span title="LOS azimuth">↓ ${losAz}</span>
${captureBtn}
</div>
</div>
`;
}).join('');
}
function handoffToWeatherSat(passIdx) {
const pass = passes[passIdx];
if (!pass) return;
const msg = {
type: 'weather-sat-handoff',
satellite: pass.satellite,
aosTime: pass.aosTime || pass.startTimeISO,
tcaEl: pass.tcaEl ?? pass.maxEl,
duration: pass.duration,
};
// Prefer parent (embedded iframe), fall back to opener (new window)
const target = window.parent !== window ? window.parent : window.opener;
if (target) {
target.postMessage(msg, '*');
}
}
function selectPass(idx) {
selectedPass = idx;
renderPassList();
@@ -637,7 +855,6 @@
drawPolarPlot(pass);
updateGroundTrack(pass);
updateTelemetry(pass);
updateRealTimePositions(true);
}
function drawPolarPlot(pass) {
@@ -914,112 +1131,6 @@
}
}
async function updateRealTimePositions(fitBoundsToOrbit = false) {
const lat = parseFloat(document.getElementById('obsLat').value);
const lon = parseFloat(document.getElementById('obsLon').value);
let targetSatellite = selectedSatellite;
let satColor = satellites[selectedSatellite]?.color || '#00d4ff';
if (selectedPass !== null && passes[selectedPass]) {
const pass = passes[selectedPass];
targetSatellite = pass.satellite;
satColor = pass.color || satColor;
}
try {
const response = await fetch('/satellite/position', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
latitude: lat,
longitude: lon,
satellites: [targetSatellite],
includeTrack: true
})
});
const data = await response.json();
if (data.status === 'success' && data.positions.length > 0) {
const pos = data.positions[0];
document.getElementById('telLat').textContent = pos.lat.toFixed(4) + '°';
document.getElementById('telLon').textContent = pos.lon.toFixed(4) + '°';
document.getElementById('telAlt').textContent = pos.altitude.toFixed(0) + ' km';
document.getElementById('telEl').textContent = pos.elevation.toFixed(1) + '°';
document.getElementById('telAz').textContent = pos.azimuth.toFixed(1) + '°';
document.getElementById('telDist').textContent = pos.distance.toFixed(0) + ' km';
document.getElementById('statVisible').textContent = pos.elevation > 0 ? '1' : '0';
if (groundMap) {
if (satMarker) groundMap.removeLayer(satMarker);
const satIcon = L.divIcon({
className: 'sat-marker-live',
html: `<div style="width: 20px; height: 20px; background: ${satColor}; border-radius: 50%; border: 3px solid #fff; box-shadow: 0 0 20px ${satColor}, 0 0 40px ${satColor};"></div>`,
iconSize: [20, 20],
iconAnchor: [10, 10]
});
satMarker = L.marker([pos.lat, pos.lon], { icon: satIcon }).addTo(groundMap);
}
if (pos.track && groundMap) {
if (orbitTrack) groundMap.removeLayer(orbitTrack);
const segments = [];
let currentSegment = [];
for (let i = 0; i < pos.track.length; i++) {
const p = pos.track[i];
if (currentSegment.length > 0) {
const prevLon = currentSegment[currentSegment.length - 1][1];
const crossesAntimeridian = (prevLon > 90 && p.lon < -90) || (prevLon < -90 && p.lon > 90);
if (crossesAntimeridian) {
if (currentSegment.length >= 1) segments.push(currentSegment);
currentSegment = [];
}
}
currentSegment.push([p.lat, p.lon]);
}
if (currentSegment.length >= 1) segments.push(currentSegment);
orbitTrack = L.layerGroup();
const allOrbitCoords = [];
segments.forEach(seg => {
L.polyline(seg, {
color: satColor,
weight: 2,
opacity: 0.6,
dashArray: '5, 5'
}).addTo(orbitTrack);
allOrbitCoords.push(...seg);
});
orbitTrack.addTo(groundMap);
if (fitBoundsToOrbit && allOrbitCoords.length > 0) {
allOrbitCoords.push([lat, lon]);
groundMap.fitBounds(L.latLngBounds(allOrbitCoords), { padding: [30, 30] });
}
}
if (selectedPass !== null && passes[selectedPass]) {
drawPolarPlot(passes[selectedPass]);
drawCurrentPositionOnPolar(pos.azimuth, pos.elevation, satColor);
} else {
drawPolarPlotWithPosition(pos.azimuth, pos.elevation, satColor);
}
}
} catch (err) {
const transient = (typeof window.isTransientOrOffline === 'function' && window.isTransientOrOffline(err)) ||
(typeof navigator !== 'undefined' && navigator.onLine === false) ||
/failed to fetch|network io suspended|networkerror|timeout/i.test(String((err && err.message) || err || ''));
if (!transient) {
console.error('Position update error:', err);
}
}
}
function drawPolarPlotWithPosition(az, el, color) {
const canvas = document.getElementById('polarPlot');
const ctx = canvas.getContext('2d');
@@ -1129,6 +1240,60 @@
}
}
async function loadTransmitters(noradId) {
const container = document.getElementById('transmittersList');
const countEl = document.getElementById('txCount');
if (!container) return;
if (!noradId) {
container.innerHTML = '<div style="text-align:center;color:var(--text-secondary);padding:15px;font-size:11px;">Select a satellite</div>';
if (countEl) countEl.textContent = '';
return;
}
container.innerHTML = '<div style="text-align:center;color:var(--text-secondary);padding:15px;font-size:11px;">Loading...</div>';
try {
const r = await fetch(`/satellite/transmitters/${noradId}`);
const data = await r.json();
renderTransmitters(data.transmitters || []);
} catch (e) {
container.innerHTML = '<div style="text-align:center;color:var(--text-secondary);padding:15px;font-size:11px;">Failed to load</div>';
if (countEl) countEl.textContent = '';
}
}
function renderTransmitters(txList) {
const container = document.getElementById('transmittersList');
const countEl = document.getElementById('txCount');
if (!container) return;
const active = txList.filter(t => t.status === 'active');
const all = txList;
if (countEl) countEl.textContent = all.length ? `(${active.length}/${all.length})` : '';
if (!all.length) {
container.innerHTML = '<div style="text-align:center;color:var(--text-secondary);padding:15px;font-size:11px;">No transmitter data available</div>';
return;
}
container.innerHTML = all.map(tx => {
const isActive = tx.status === 'active';
const dl = tx.downlink_low != null ? tx.downlink_low.toFixed(3) + ' MHz' : null;
const dlHigh = tx.downlink_high != null && tx.downlink_high !== tx.downlink_low ? '–' + tx.downlink_high.toFixed(3) : '';
const ul = tx.uplink_low != null ? tx.uplink_low.toFixed(3) + ' MHz' : null;
const baud = tx.baud ? ` · ${tx.baud} Bd` : '';
const mode = tx.mode || '';
return `<div class="tx-item ${isActive ? 'tx-active' : 'tx-inactive'}">
<div class="tx-status-dot" style="background:${isActive ? 'var(--accent-green)' : '#444'};"></div>
<div class="tx-body">
<div class="tx-desc">${tx.description || 'Unknown'}</div>
${dl ? `<div class="tx-freq">↓ ${dl}${dlHigh} ${mode}${baud}</div>` : ''}
${ul ? `<div class="tx-freq tx-uplink">↑ ${ul}</div>` : ''}
<div class="tx-service">${tx.service || ''} ${tx.type || ''}</div>
</div>
</div>`;
}).join('');
}
function drawCurrentPositionOnPolar(az, el, color) {
const canvas = document.getElementById('polarPlot');
const ctx = canvas.getContext('2d');

210
utils/satellite_predict.py Normal file
View File

@@ -0,0 +1,210 @@
"""Shared satellite pass prediction utility.
Used by both the satellite tracking dashboard and the weather satellite scheduler.
Uses Skyfield's find_events() for accurate AOS/TCA/LOS event detection.
"""
from __future__ import annotations
import datetime
import math
from typing import Any
from utils.logging import get_logger
logger = get_logger('intercept.satellite_predict')
def predict_passes(
tle_data: tuple,
observer, # skyfield wgs84.latlon object
ts, # skyfield timescale
t0, # skyfield Time start
t1, # skyfield Time end
min_el: float = 10.0,
include_trajectory: bool = True,
include_ground_track: bool = True,
) -> list[dict[str, Any]]:
"""Predict satellite passes over an observer location.
Args:
tle_data: (name, line1, line2) tuple
observer: Skyfield wgs84.latlon observer
ts: Skyfield timescale
t0: Start time (Skyfield Time)
t1: End time (Skyfield Time)
min_el: Minimum peak elevation in degrees to include pass
include_trajectory: Include 30-point az/el trajectory for polar plot
include_ground_track: Include 60-point lat/lon ground track for map
Returns:
List of pass dicts sorted by AOS time. Each dict contains:
aosTime, aosAz, aosEl,
tcaTime, tcaEl, tcaAz,
losTime, losAz, losEl,
duration (minutes, float),
startTime (human-readable UTC),
startTimeISO (ISO string),
endTimeISO (ISO string),
maxEl (float, same as tcaEl),
trajectory (list of {az, el} if include_trajectory),
groundTrack (list of {lat, lon} if include_ground_track)
"""
from skyfield.api import EarthSatellite, wgs84
# Filter decaying satellites by checking ndot from TLE line1 chars 33-43
try:
line1 = tle_data[1]
ndot_str = line1[33:43].strip()
ndot = float(ndot_str)
if abs(ndot) > 0.01:
logger.debug(
'Skipping decaying satellite %s (ndot=%s)', tle_data[0], ndot
)
return []
except (ValueError, IndexError):
# Don't skip on parse error
pass
# Create EarthSatellite object
try:
satellite = EarthSatellite(tle_data[1], tle_data[2], tle_data[0], ts)
except Exception as exc:
logger.debug('Failed to create EarthSatellite for %s: %s', tle_data[0], exc)
return []
# Find events using Skyfield's native find_events()
# Event types: 0=AOS, 1=TCA, 2=LOS
try:
times, events = satellite.find_events(
observer, t0, t1, altitude_degrees=min_el
)
except Exception as exc:
logger.debug('find_events failed for %s: %s', tle_data[0], exc)
return []
# Group events into AOS->TCA->LOS triplets
passes = []
i = 0
total = len(events)
# Skip any leading non-AOS events (satellite already above horizon at t0)
while i < total and events[i] != 0:
i += 1
while i < total:
# Expect AOS (0)
if events[i] != 0:
i += 1
continue
aos_time = times[i]
i += 1
# Collect TCA and LOS, watching for premature next AOS
tca_time = None
los_time = None
while i < total and events[i] != 0:
if events[i] == 1:
tca_time = times[i]
elif events[i] == 2:
los_time = times[i]
i += 1
# Must have both AOS and LOS to form a valid pass
if los_time is None:
# Incomplete pass — skip
continue
# If TCA is missing, derive from midpoint between AOS and LOS
if tca_time is None:
aos_tt = aos_time.tt
los_tt = los_time.tt
tca_time = ts.tt_jd((aos_tt + los_tt) / 2.0)
# Compute topocentric positions at AOS, TCA, LOS
try:
aos_topo = (satellite - observer).at(aos_time)
tca_topo = (satellite - observer).at(tca_time)
los_topo = (satellite - observer).at(los_time)
aos_alt, aos_az, _ = aos_topo.altaz()
tca_alt, tca_az, _ = tca_topo.altaz()
los_alt, los_az, _ = los_topo.altaz()
aos_dt = aos_time.utc_datetime()
tca_dt = tca_time.utc_datetime()
los_dt = los_time.utc_datetime()
duration = (los_dt - aos_dt).total_seconds() / 60.0
pass_dict: dict[str, Any] = {
'aosTime': aos_dt.isoformat(),
'aosAz': round(float(aos_az.degrees), 1),
'aosEl': round(float(aos_alt.degrees), 1),
'tcaTime': tca_dt.isoformat(),
'tcaAz': round(float(tca_az.degrees), 1),
'tcaEl': round(float(tca_alt.degrees), 1),
'losTime': los_dt.isoformat(),
'losAz': round(float(los_az.degrees), 1),
'losEl': round(float(los_alt.degrees), 1),
'duration': round(duration, 1),
# Backwards-compatible fields
'startTime': aos_dt.strftime('%Y-%m-%d %H:%M UTC'),
'startTimeISO': aos_dt.isoformat(),
'endTimeISO': los_dt.isoformat(),
'maxEl': round(float(tca_alt.degrees), 1),
}
# Build 30-point az/el trajectory for polar plot
if include_trajectory:
trajectory = []
for step in range(30):
frac = step / 29.0
t_pt = ts.tt_jd(
aos_time.tt + frac * (los_time.tt - aos_time.tt)
)
try:
pt_alt, pt_az, _ = (satellite - observer).at(t_pt).altaz()
trajectory.append({
'az': round(float(pt_az.degrees), 1),
'el': round(float(max(0.0, pt_alt.degrees)), 1),
})
except Exception as pt_exc:
logger.debug(
'Trajectory point error for %s: %s', tle_data[0], pt_exc
)
pass_dict['trajectory'] = trajectory
# Build 60-point lat/lon ground track for map
if include_ground_track:
ground_track = []
for step in range(60):
frac = step / 59.0
t_pt = ts.tt_jd(
aos_time.tt + frac * (los_time.tt - aos_time.tt)
)
try:
geocentric = satellite.at(t_pt)
subpoint = wgs84.subpoint(geocentric)
ground_track.append({
'lat': round(float(subpoint.latitude.degrees), 4),
'lon': round(float(subpoint.longitude.degrees), 4),
})
except Exception as gt_exc:
logger.debug(
'Ground track point error for %s: %s', tle_data[0], gt_exc
)
pass_dict['groundTrack'] = ground_track
passes.append(pass_dict)
except Exception as exc:
logger.debug(
'Failed to compute pass details for %s: %s', tle_data[0], exc
)
continue
passes.sort(key=lambda p: p['startTimeISO'])
return passes

436
utils/satellite_telemetry.py Normal file
View File

@@ -0,0 +1,436 @@
"""Satellite telemetry packet parsers.
Provides pure-Python decoders for common amateur/CubeSat protocols:
- AX.25 (callsign-addressed frames)
- CSP (CubeSat Space Protocol)
- CCSDS TM (space packet primary header)
Also provides a PayloadAnalyzer that generates multi-interpretation
views of raw binary data (hex dump, float32, uint16/32, strings).
"""
from __future__ import annotations
import math
import struct
import string
from datetime import datetime
# ---------------------------------------------------------------------------
# AX.25 parser
# ---------------------------------------------------------------------------
def _decode_ax25_callsign(addr_bytes: bytes) -> str:
"""Decode a 7-byte AX.25 address field into a 'CALL-SSID' string.
The first 6 bytes encode the callsign (each ASCII character left-shifted
by 1 bit). The 7th byte encodes the SSID in bits 4-1.
Args:
addr_bytes: Exactly 7 bytes of raw address data.
Returns:
A callsign string such as ``"N0CALL-3"`` or ``"N0CALL"`` (no suffix
when SSID is 0).
"""
callsign = "".join(chr(b >> 1) for b in addr_bytes[:6]).rstrip()
ssid = (addr_bytes[6] >> 1) & 0x0F
return f"{callsign}-{ssid}" if ssid else callsign
def parse_ax25(data: bytes) -> dict | None:
"""Parse an AX.25 frame from raw bytes.
Decodes destination and source callsigns, optional repeater addresses,
control byte, optional PID byte, and payload.
Args:
data: Raw bytes of the AX.25 frame (without HDLC flags or FCS).
Returns:
A dict with parsed fields or ``None`` if the frame is too short or
cannot be decoded.
"""
try:
# Minimum: 7 (dest) + 7 (src) + 1 (control) = 15 bytes
if len(data) < 15:
return None
destination = _decode_ax25_callsign(data[0:7])
source = _decode_ax25_callsign(data[7:14])
# Walk repeater addresses. The H-bit (LSB of byte 6 in each address)
# being set means this is the last address in the chain.
offset = 14 # byte index of the last byte in the source field
repeaters: list[str] = []
if not (data[offset] & 0x01):
# More addresses follow; read up to 8 repeaters.
for _ in range(8):
rep_start = offset + 1
rep_end = rep_start + 7
if rep_end > len(data):
break
repeaters.append(_decode_ax25_callsign(data[rep_start:rep_end]))
offset = rep_end - 1 # last byte of this repeater field
if data[offset] & 0x01:
# H-bit set — this was the final address
break
# Control byte follows the last address field
ctrl_offset = offset + 1
if ctrl_offset >= len(data):
return None
control = data[ctrl_offset]
payload_offset = ctrl_offset + 1
# PID byte is present for I-frames (bits 0-1 == 0b00) and
# UI-frames (bits 0-5 == 0b000011). More generally: absent only
# for pure unnumbered frames where (control & 0x03) == 0x03 AND
# control is not 0x03 itself (UI).
pid: int | None = None
is_unnumbered = (control & 0x03) == 0x03
is_ui = control == 0x03
if not is_unnumbered or is_ui:
if payload_offset < len(data):
pid = data[payload_offset]
payload_offset += 1
payload = data[payload_offset:]
return {
"protocol": "AX.25",
"destination": destination,
"source": source,
"repeaters": repeaters,
"control": control,
"pid": pid,
"payload": payload,
"payload_hex": payload.hex(),
"payload_length": len(payload),
}
except Exception: # noqa: BLE001
return None
# ---------------------------------------------------------------------------
# CSP parser
# ---------------------------------------------------------------------------
def parse_csp(data: bytes) -> dict | None:
"""Parse a CSP v1 (CubeSat Space Protocol) header.
The first 4 bytes form a big-endian 32-bit header word with the
following bit layout::
bits 31-27 priority (5 bits)
bits 26-22 source (5 bits)
bits 21-17 destination (5 bits)
bits 16-12 dest_port (5 bits)
bits 11-6 src_port (6 bits)
bits 5-0 flags (6 bits)
Args:
data: Raw bytes starting from the CSP header.
Returns:
A dict with parsed CSP fields and payload, or ``None`` on failure.
"""
try:
if len(data) < 4:
return None
header: int = struct.unpack(">I", data[:4])[0]
priority = (header >> 27) & 0x1F
source = (header >> 22) & 0x1F
destination = (header >> 17) & 0x1F
dest_port = (header >> 12) & 0x1F
src_port = (header >> 6) & 0x3F
raw_flags = header & 0x3F
flags = {
"frag": bool(raw_flags & 0x10),
"hmac": bool(raw_flags & 0x08),
"xtea": bool(raw_flags & 0x04),
"rdp": bool(raw_flags & 0x02),
"crc": bool(raw_flags & 0x01),
}
payload = data[4:]
return {
"protocol": "CSP",
"priority": priority,
"source": source,
"destination": destination,
"dest_port": dest_port,
"src_port": src_port,
"flags": flags,
"payload": payload,
"payload_hex": payload.hex(),
"payload_length": len(payload),
}
except Exception: # noqa: BLE001
return None
# ---------------------------------------------------------------------------
# CCSDS parser
# ---------------------------------------------------------------------------
def parse_ccsds(data: bytes) -> dict | None:
"""Parse a CCSDS Space Packet primary header (6 bytes).
Header layout::
bytes 0-1: version (3 bits) | packet_type (1 bit) |
secondary_header_flag (1 bit) | APID (11 bits)
bytes 2-3: sequence_flags (2 bits) | sequence_count (14 bits)
bytes 4-5: data_length field (16 bits, = actual_payload_length - 1)
Args:
data: Raw bytes starting from the CCSDS primary header.
Returns:
A dict with parsed CCSDS fields and payload, or ``None`` on failure.
"""
try:
if len(data) < 6:
return None
word0: int = struct.unpack(">H", data[0:2])[0]
word1: int = struct.unpack(">H", data[2:4])[0]
word2: int = struct.unpack(">H", data[4:6])[0]
version = (word0 >> 13) & 0x07
packet_type = (word0 >> 12) & 0x01
secondary_header_flag = bool((word0 >> 11) & 0x01)
apid = word0 & 0x07FF
sequence_flags = (word1 >> 14) & 0x03
sequence_count = word1 & 0x3FFF
data_length = word2 # raw field; actual user data bytes = data_length + 1
payload = data[6:]
return {
"protocol": "CCSDS_TM",
"version": version,
"packet_type": packet_type,
"secondary_header": secondary_header_flag,
"apid": apid,
"sequence_flags": sequence_flags,
"sequence_count": sequence_count,
"data_length": data_length,
"payload": payload,
"payload_hex": payload.hex(),
"payload_length": len(payload),
}
except Exception: # noqa: BLE001
return None
# ---------------------------------------------------------------------------
# Payload analyzer
# ---------------------------------------------------------------------------
_PRINTABLE = set(string.printable) - set("\t\n\r\x0b\x0c")
def _hex_dump(data: bytes) -> str:
"""Format bytes as an annotated hex dump, 16 bytes per line.
Each line is formatted as::
OOOO: XX XX XX XX XX XX XX XX XX XX XX XX XX XX XX XX ASCII
where ``OOOO`` is the hex offset and ``ASCII`` shows printable characters
(non-printable replaced with ``'.'``).
Args:
data: Bytes to format.
Returns:
Multi-line hex dump string (trailing newline on each line).
"""
lines: list[str] = []
for row in range(0, len(data), 16):
chunk = data[row : row + 16]
# Build groups of 4 bytes separated by two spaces
groups: list[str] = []
for g in range(0, 16, 4):
group_bytes = chunk[g : g + 4]
groups.append(" ".join(f"{b:02X}" for b in group_bytes))
hex_part = " ".join(groups)
# Pad to fixed width: 16 bytes × 3 chars - 1 space + 3 group separators
# Maximum width: 11+2+11+2+11+2+11 = 50 chars; pad to 50
hex_part = hex_part.ljust(50)
ascii_part = "".join(chr(b) if chr(b) in _PRINTABLE else "." for b in chunk)
lines.append(f"{row:04X}: {hex_part} {ascii_part}\n")
return "".join(lines)
def _extract_strings(data: bytes, min_len: int = 3) -> list[str]:
"""Extract runs of printable ASCII characters of at least ``min_len``."""
results: list[str] = []
current: list[str] = []
for b in data:
ch = chr(b)
if ch in _PRINTABLE:
current.append(ch)
else:
if len(current) >= min_len:
results.append("".join(current))
current = []
if len(current) >= min_len:
results.append("".join(current))
return results
def analyze_payload(data: bytes) -> dict:
"""Generate a multi-interpretation analysis of raw bytes.
Produces a hex dump, several numeric/string interpretations, and a
list of heuristic observations about plausible sensor values.
Args:
data: Raw bytes to analyze.
Returns:
A dict containing ``hex_dump``, ``length``, ``interpretations``,
and ``heuristics`` keys. Never raises an exception.
"""
try:
hex_dump = _hex_dump(data)
length = len(data)
# --- float32 (little-endian) ---
float32_values: list[float] = []
for i in range(0, length - 3, 4):
(val,) = struct.unpack_from("<f", data, i)
if not math.isnan(val) and abs(val) <= 1e9:
float32_values.append(val)
# --- uint16 little-endian ---
uint16_values: list[int] = []
for i in range(0, length - 1, 2):
(val,) = struct.unpack_from("<H", data, i)
uint16_values.append(val)
# --- uint32 little-endian ---
uint32_values: list[int] = []
for i in range(0, length - 3, 4):
(val,) = struct.unpack_from("<I", data, i)
uint32_values.append(val)
# --- printable string runs ---
strings = _extract_strings(data, min_len=3)
interpretations = {
"float32": float32_values,
"uint16_le": uint16_values,
"uint32_le": uint32_values,
"strings": strings,
}
# --- heuristics ---
heuristics: list[str] = []
used_as_voltage: set[int] = set()
for idx, v in enumerate(float32_values):
# Voltage: small positive float
if 0.0 < v < 10.0:
heuristics.append(f"Possible voltage: {v:.3f} V (index {idx})")
used_as_voltage.add(idx)
for idx, v in enumerate(float32_values):
# Temperature: plausible range, not already flagged as voltage, not zero
if -50.0 < v < 120.0 and idx not in used_as_voltage and v != 0.0:
heuristics.append(f"Possible temperature: {v:.1f}°C (index {idx})")
for idx, v in enumerate(float32_values):
# Current: small positive float not already flagged as voltage
if 0.0 < v < 5.0 and idx not in used_as_voltage:
heuristics.append(f"Possible current: {v:.3f} A (index {idx})")
for idx, v in enumerate(float32_values):
# Unix timestamp: plausible range (roughly 2001–2033)
if 1_000_000_000.0 < v < 2_000_000_000.0:
ts = datetime.utcfromtimestamp(v)
heuristics.append(f"Possible Unix timestamp: {ts} (index {idx})")
return {
"hex_dump": hex_dump,
"length": length,
"interpretations": interpretations,
"heuristics": heuristics,
}
except Exception: # noqa: BLE001
# Guarantee a safe return even on completely malformed input
return {
"hex_dump": "",
"length": len(data) if isinstance(data, (bytes, bytearray)) else 0,
"interpretations": {"float32": [], "uint16_le": [], "uint32_le": [], "strings": []},
"heuristics": [],
}
# ---------------------------------------------------------------------------
# Auto-parser
# ---------------------------------------------------------------------------
def auto_parse(data: bytes) -> dict:
"""Attempt to decode a packet using each supported protocol in turn.
Tries parsers in priority order: CSP → CCSDS → AX.25. Returns the
first successful parse merged with a ``payload_analysis`` key produced
by :func:`analyze_payload`.
Args:
data: Raw bytes of the packet.
Returns:
A dict with parsed protocol fields plus ``payload_analysis``, or a
fallback dict with ``protocol: 'unknown'`` and a top-level
``analysis`` key if no parser succeeds.
"""
# CSP: 4-byte header minimum
if len(data) >= 4:
result = parse_csp(data)
if result is not None:
result["payload_analysis"] = analyze_payload(result["payload"])
return result
# CCSDS: 6-byte header minimum
if len(data) >= 6:
result = parse_ccsds(data)
if result is not None:
result["payload_analysis"] = analyze_payload(result["payload"])
return result
# AX.25: 15-byte frame minimum
if len(data) >= 15:
result = parse_ax25(data)
if result is not None:
result["payload_analysis"] = analyze_payload(result["payload"])
return result
# Nothing matched — return a raw analysis
return {
"protocol": "unknown",
"raw_hex": data.hex(),
"analysis": analyze_payload(data),
}

145
utils/satnogs.py Normal file
View File

@@ -0,0 +1,145 @@
"""SatNOGS transmitter data.
Fetches downlink/uplink frequency data from the SatNOGS database,
keyed by NORAD ID. Cached for 24 hours to avoid hammering the API.
"""
from __future__ import annotations
import json
import threading
import time
import urllib.request
from utils.logging import get_logger
logger = get_logger("intercept.satnogs")
# ---------------------------------------------------------------------------
# Module-level cache
# ---------------------------------------------------------------------------
_transmitters: dict[int, list[dict]] = {}
_fetched_at: float = 0.0
_CACHE_TTL = 86400 # 24 hours in seconds
_fetch_lock = threading.Lock()
_SATNOGS_URL = "https://db.satnogs.org/api/transmitters/?format=json"
_REQUEST_TIMEOUT = 15 # seconds
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _hz_to_mhz(value: float | int | None) -> float | None:
"""Convert a frequency in Hz to MHz, returning None if value is None."""
if value is None:
return None
return float(value) / 1_000_000.0
def _safe_float(value: object) -> float | None:
"""Return a float or None, silently swallowing conversion errors."""
if value is None:
return None
try:
return float(value) # type: ignore[arg-type]
except (TypeError, ValueError):
return None
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def fetch_transmitters() -> dict[int, list[dict]]:
"""Fetch transmitter records from the SatNOGS database API.
Makes a single HTTP GET to the SatNOGS transmitters endpoint, groups
results by NORAD catalogue ID, and converts all frequency fields from
Hz to MHz.
Returns:
A dict mapping NORAD ID (int) to a list of transmitter dicts.
Returns an empty dict on any network or parse error.
"""
try:
logger.info("Fetching SatNOGS transmitter data from %s", _SATNOGS_URL)
with urllib.request.urlopen(_SATNOGS_URL, timeout=_REQUEST_TIMEOUT) as resp:
raw = resp.read()
records: list[dict] = json.loads(raw)
grouped: dict[int, list[dict]] = {}
for item in records:
norad_id = item.get("norad_cat_id")
if norad_id is None:
continue
norad_id = int(norad_id)
entry: dict = {
"description": str(item.get("description") or ""),
"downlink_low": _hz_to_mhz(_safe_float(item.get("downlink_low"))),
"downlink_high": _hz_to_mhz(_safe_float(item.get("downlink_high"))),
"uplink_low": _hz_to_mhz(_safe_float(item.get("uplink_low"))),
"uplink_high": _hz_to_mhz(_safe_float(item.get("uplink_high"))),
"mode": str(item.get("mode") or ""),
"baud": _safe_float(item.get("baud")),
"status": str(item.get("status") or ""),
"type": str(item.get("type") or ""),
"service": str(item.get("service") or ""),
}
grouped.setdefault(norad_id, []).append(entry)
logger.info(
"SatNOGS fetch complete: %d satellites with transmitter data",
len(grouped),
)
return grouped
except Exception as exc: # noqa: BLE001
logger.warning("Failed to fetch SatNOGS transmitter data: %s", exc)
return {}
def get_transmitters(norad_id: int) -> list[dict]:
"""Return cached transmitter records for a given NORAD catalogue ID.
Refreshes the in-memory cache from the SatNOGS API when the cache is
empty or older than ``_CACHE_TTL`` seconds (24 hours).
Args:
norad_id: The NORAD catalogue ID of the satellite.
Returns:
A (possibly empty) list of transmitter dicts for that satellite.
"""
global _transmitters, _fetched_at # noqa: PLW0603
with _fetch_lock:
age = time.time() - _fetched_at
if not _transmitters or age > _CACHE_TTL:
_transmitters = fetch_transmitters()
_fetched_at = time.time()
return _transmitters.get(int(norad_id), [])
def refresh_transmitters() -> int:
"""Force-refresh the transmitter cache regardless of TTL.
Returns:
The number of satellites (unique NORAD IDs) with transmitter data
after the refresh.
"""
global _transmitters, _fetched_at # noqa: PLW0603
with _fetch_lock:
_transmitters = fetch_transmitters()
_fetched_at = time.time()
return len(_transmitters)

344
utils/weather_sat_predict.py
View File

@@ -1,218 +1,126 @@
"""Weather satellite pass prediction utility.
Shared prediction logic used by both the API endpoint and the auto-scheduler.
"""
from __future__ import annotations
import datetime
import time
from typing import Any
from utils.logging import get_logger
from utils.weather_sat import WEATHER_SATELLITES
logger = get_logger('intercept.weather_sat_predict')
# Cache skyfield timescale to avoid re-downloading/re-parsing per request
_cached_timescale = None
def _get_timescale():
global _cached_timescale
if _cached_timescale is None:
from skyfield.api import load
_cached_timescale = load.timescale()
return _cached_timescale
def _format_utc_iso(dt: datetime.datetime) -> str:
"""Return an ISO8601 UTC timestamp with a single timezone designator."""
if dt.tzinfo is None:
dt = dt.replace(tzinfo=datetime.timezone.utc)
else:
dt = dt.astimezone(datetime.timezone.utc)
return dt.isoformat().replace('+00:00', 'Z')
def predict_passes(
lat: float,
lon: float,
hours: int = 24,
min_elevation: float = 15.0,
include_trajectory: bool = False,
include_ground_track: bool = False,
) -> list[dict[str, Any]]:
"""Predict upcoming weather satellite passes for an observer location.
Args:
lat: Observer latitude (-90 to 90)
lon: Observer longitude (-180 to 180)
hours: Hours ahead to predict (1-72)
min_elevation: Minimum max elevation in degrees (0-90)
include_trajectory: Include az/el trajectory points (30 points)
include_ground_track: Include lat/lon ground track points (60 points)
Returns:
List of pass dicts sorted by start time.
Raises:
ImportError: If skyfield is not installed.
"""
from skyfield.almanac import find_discrete
from skyfield.api import EarthSatellite, wgs84
from data.satellites import TLE_SATELLITES
# Use live TLE cache from satellite module if available (refreshed from CelesTrak).
# Cache the reference locally so repeated calls don't re-import each time.
tle_source = TLE_SATELLITES
if not hasattr(predict_passes, '_tle_ref') or \
(time.time() - getattr(predict_passes, '_tle_ref_ts', 0)) > 3600:
try:
from routes.satellite import _tle_cache
if _tle_cache:
predict_passes._tle_ref = _tle_cache
predict_passes._tle_ref_ts = time.time()
except ImportError:
pass
if hasattr(predict_passes, '_tle_ref') and predict_passes._tle_ref:
tle_source = predict_passes._tle_ref
ts = _get_timescale()
observer = wgs84.latlon(lat, lon)
t0 = ts.now()
t1 = ts.utc(t0.utc_datetime() + datetime.timedelta(hours=hours))
all_passes: list[dict[str, Any]] = []
for sat_key, sat_info in WEATHER_SATELLITES.items():
if not sat_info['active']:
continue
tle_data = tle_source.get(sat_info['tle_key'])
if not tle_data:
continue
satellite = EarthSatellite(tle_data[1], tle_data[2], tle_data[0], ts)
def above_horizon(t, _sat=satellite):
diff = _sat - observer
topocentric = diff.at(t)
alt, _, _ = topocentric.altaz()
return alt.degrees > 0
above_horizon.step_days = 1 / 720
try:
times, events = find_discrete(t0, t1, above_horizon)
except Exception:
continue
i = 0
while i < len(times):
if i < len(events) and events[i]: # Rising
rise_time = times[i]
set_time = None
for j in range(i + 1, len(times)):
if not events[j]: # Setting
set_time = times[j]
i = j
break
else:
i += 1
continue
if set_time is None:
i += 1
continue
rise_dt = rise_time.utc_datetime()
set_dt = set_time.utc_datetime()
duration_seconds = (
set_dt - rise_dt
).total_seconds()
duration_minutes = round(duration_seconds / 60, 1)
# Calculate max elevation (always) and trajectory points (only if requested)
max_el = 0.0
max_el_az = 0.0
trajectory: list[dict[str, float]] = []
num_traj_points = 30
for k in range(num_traj_points):
frac = k / (num_traj_points - 1)
t_point = ts.utc(
rise_time.utc_datetime()
+ datetime.timedelta(seconds=duration_seconds * frac)
)
diff = satellite - observer
topocentric = diff.at(t_point)
alt, az, _ = topocentric.altaz()
if alt.degrees > max_el:
max_el = alt.degrees
max_el_az = az.degrees
if include_trajectory:
trajectory.append({
'el': float(max(0, alt.degrees)),
'az': float(az.degrees),
})
if max_el < min_elevation:
i += 1
continue
# Rise/set azimuths
rise_topo = (satellite - observer).at(rise_time)
_, rise_az, _ = rise_topo.altaz()
set_topo = (satellite - observer).at(set_time)
_, set_az, _ = set_topo.altaz()
pass_data: dict[str, Any] = {
'id': f"{sat_key}_{rise_dt.strftime('%Y%m%d%H%M%S')}",
'satellite': sat_key,
'name': sat_info['name'],
'frequency': sat_info['frequency'],
'mode': sat_info['mode'],
'startTime': rise_dt.strftime('%Y-%m-%d %H:%M UTC'),
'startTimeISO': _format_utc_iso(rise_dt),
'endTimeISO': _format_utc_iso(set_dt),
'maxEl': round(max_el, 1),
'maxElAz': round(max_el_az, 1),
'riseAz': round(rise_az.degrees, 1),
'setAz': round(set_az.degrees, 1),
'duration': duration_minutes,
'quality': (
'excellent' if max_el >= 60
else 'good' if max_el >= 30
else 'fair'
),
}
if include_trajectory:
pass_data['trajectory'] = trajectory
if include_ground_track:
ground_track: list[dict[str, float]] = []
for k in range(60):
frac = k / 59
t_point = ts.utc(
rise_time.utc_datetime()
+ datetime.timedelta(seconds=duration_seconds * frac)
)
geocentric = satellite.at(t_point)
subpoint = wgs84.subpoint(geocentric)
ground_track.append({
'lat': float(subpoint.latitude.degrees),
'lon': float(subpoint.longitude.degrees),
})
pass_data['groundTrack'] = ground_track
all_passes.append(pass_data)
i += 1
all_passes.sort(key=lambda p: p['startTimeISO'])
return all_passes
"""Weather satellite pass prediction utility.
Shared prediction logic used by both the API endpoint and the auto-scheduler.
Delegates to utils.satellite_predict for core pass detection, then enriches
results with weather-satellite-specific metadata.
"""
from __future__ import annotations
import datetime
import time
from typing import Any
from utils.logging import get_logger
from utils.weather_sat import WEATHER_SATELLITES
logger = get_logger('intercept.weather_sat_predict')
# Cache skyfield timescale to avoid re-downloading/re-parsing per request
_cached_timescale = None
def _get_timescale():
global _cached_timescale
if _cached_timescale is None:
from skyfield.api import load
_cached_timescale = load.timescale()
return _cached_timescale
def _get_tle_source() -> dict:
"""Return the best available TLE source (live cache preferred over static data)."""
from data.satellites import TLE_SATELLITES
if not hasattr(_get_tle_source, '_ref') or \
(time.time() - getattr(_get_tle_source, '_ref_ts', 0)) > 3600:
try:
from routes.satellite import _tle_cache
if _tle_cache:
_get_tle_source._ref = _tle_cache
_get_tle_source._ref_ts = time.time()
except ImportError:
pass
return getattr(_get_tle_source, '_ref', None) or TLE_SATELLITES
def predict_passes(
lat: float,
lon: float,
hours: int = 24,
min_elevation: float = 15.0,
include_trajectory: bool = False,
include_ground_track: bool = False,
) -> list[dict[str, Any]]:
"""Predict upcoming weather satellite passes for an observer location.
Args:
lat: Observer latitude (-90 to 90)
lon: Observer longitude (-180 to 180)
hours: Hours ahead to predict (1-72)
min_elevation: Minimum peak elevation in degrees (0-90)
include_trajectory: Include az/el trajectory points for polar plot
include_ground_track: Include lat/lon ground track points for map
Returns:
List of pass dicts sorted by start time, enriched with weather-satellite
fields: id, satellite, name, frequency, mode, quality, riseAz, setAz,
maxElAz, and all standard fields from utils.satellite_predict.
"""
from skyfield.api import wgs84
from utils.satellite_predict import predict_passes as _predict_passes
tle_source = _get_tle_source()
ts = _get_timescale()
observer = wgs84.latlon(lat, lon)
t0 = ts.now()
t1 = ts.utc(t0.utc_datetime() + datetime.timedelta(hours=hours))
all_passes: list[dict[str, Any]] = []
for sat_key, sat_info in WEATHER_SATELLITES.items():
if not sat_info['active']:
continue
tle_data = tle_source.get(sat_info['tle_key'])
if not tle_data:
continue
sat_passes = _predict_passes(
tle_data,
observer,
ts,
t0,
t1,
min_el=min_elevation,
include_trajectory=include_trajectory,
include_ground_track=include_ground_track,
)
for p in sat_passes:
aos_iso = p['startTimeISO']
try:
aos_dt = datetime.datetime.fromisoformat(aos_iso)
pass_id = f"{sat_key}_{aos_dt.strftime('%Y%m%d%H%M%S')}"
except Exception:
pass_id = f"{sat_key}_{aos_iso}"
# Enrich with weather-satellite-specific fields
p['id'] = pass_id
p['satellite'] = sat_key
p['name'] = sat_info['name']
p['frequency'] = sat_info['frequency']
p['mode'] = sat_info['mode']
# Backwards-compatible aliases
p['riseAz'] = p['aosAz']
p['setAz'] = p['losAz']
p['maxElAz'] = p['tcaAz']
p['quality'] = (
'excellent' if p['maxEl'] >= 60
else 'good' if p['maxEl'] >= 30
else 'fair'
)
all_passes.extend(sat_passes)
all_passes.sort(key=lambda p: p['startTimeISO'])
return all_passes