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(satellite): strip observer-relative fields from SSE tracker

Browse Source 
SSE runs server-wide with DEFAULT_LAT/LON defaults of 0,0. Emitting
elevation/azimuth/distance/visible from the tracker produced wrong
values (always visible:False) that overwrote correct data from the
per-client HTTP poll every second.

The HTTP poll (/satellite/position) owns all observer-relative data.
SSE now only emits lat/lon/altitude/groundTrack. Also removes the
unused DEFAULT_LATITUDE/DEFAULT_LONGITUDE import.
This commit is contained in:
James Smith
2026-03-19 21:45:48 +00:00
parent 51b332f4cf
commit d20808fb35
2 changed files with 402 additions and 362 deletions
Download Patch File Download Diff File Expand all files Collapse all files

713
routes/satellite.py
View File

@@ -8,9 +8,9 @@ import urllib.request
from datetime import datetime, timedelta
import requests
from flask import Blueprint, Response, jsonify, make_response, render_template, request
from flask import Blueprint, Response, jsonify, make_response, render_template, request
from config import DEFAULT_LATITUDE, DEFAULT_LONGITUDE, SHARED_OBSERVER_LOCATION_ENABLED
from config import SHARED_OBSERVER_LOCATION_ENABLED
from utils.sse import sse_stream_fanout
from data.satellites import TLE_SATELLITES
from utils.database import (
@@ -30,13 +30,13 @@ satellite_bp = Blueprint('satellite', __name__, url_prefix='/satellite')
_cached_timescale = None
def _get_timescale():
global _cached_timescale
if _cached_timescale is None:
from skyfield.api import load
# Use bundled timescale data so the first request does not block on network I/O.
_cached_timescale = load.timescale(builtin=True)
return _cached_timescale
def _get_timescale():
global _cached_timescale
if _cached_timescale is None:
from skyfield.api import load
# Use bundled timescale data so the first request does not block on network I/O.
_cached_timescale = load.timescale(builtin=True)
return _cached_timescale
# Maximum response size for external requests (1MB)
MAX_RESPONSE_SIZE = 1024 * 1024
@@ -49,20 +49,20 @@ _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
_pass_cache: dict = {}
_PASS_CACHE_TTL = 300
_BUILTIN_NORAD_TO_KEY = {
25544: 'ISS',
40069: 'METEOR-M2',
57166: 'METEOR-M2-3',
59051: 'METEOR-M2-4',
}
_track_cache: dict = {}
_TRACK_CACHE_TTL = 1800
_pass_cache: dict = {}
_PASS_CACHE_TTL = 300
_BUILTIN_NORAD_TO_KEY = {
25544: 'ISS',
40069: 'METEOR-M2',
57166: 'METEOR-M2-3',
59051: 'METEOR-M2-4',
}
def _load_db_satellites_into_cache():
def _load_db_satellites_into_cache():
"""Load user-tracked satellites from DB into the TLE cache."""
global _tle_cache
try:
@@ -77,127 +77,127 @@ def _load_db_satellites_into_cache():
loaded += 1
if loaded:
logger.info(f"Loaded {loaded} user-tracked satellites into TLE cache")
except Exception as e:
logger.warning(f"Failed to load DB satellites into TLE cache: {e}")
def _normalize_satellite_name(value: object) -> str:
"""Normalize satellite identifiers for loose name matching."""
return str(value or '').strip().replace(' ', '-').upper()
def _get_tracked_satellite_maps() -> tuple[dict[int, dict], dict[str, dict]]:
"""Return tracked satellites indexed by NORAD ID and normalized name."""
by_norad: dict[int, dict] = {}
by_name: dict[str, dict] = {}
try:
for sat in get_tracked_satellites():
try:
norad_id = int(sat['norad_id'])
except (TypeError, ValueError):
continue
by_norad[norad_id] = sat
by_name[_normalize_satellite_name(sat.get('name'))] = sat
except Exception as e:
logger.warning(f"Failed to read tracked satellites for lookup: {e}")
return by_norad, by_name
def _resolve_satellite_request(sat: object, tracked_by_norad: dict[int, dict], tracked_by_name: dict[str, dict]) -> tuple[str, int | None, tuple[str, str, str] | None]:
"""Resolve a satellite request to display name, NORAD ID, and TLE data."""
norad_id: int | None = None
sat_key: str | None = None
tracked: dict | None = None
if isinstance(sat, int):
norad_id = sat
elif isinstance(sat, str):
stripped = sat.strip()
if stripped.isdigit():
norad_id = int(stripped)
else:
sat_key = stripped
if norad_id is not None:
tracked = tracked_by_norad.get(norad_id)
sat_key = _BUILTIN_NORAD_TO_KEY.get(norad_id) or (tracked.get('name') if tracked else str(norad_id))
else:
normalized = _normalize_satellite_name(sat_key)
tracked = tracked_by_name.get(normalized)
if tracked:
try:
norad_id = int(tracked['norad_id'])
except (TypeError, ValueError):
norad_id = None
sat_key = tracked.get('name') or sat_key
tle_data = None
candidate_keys: list[str] = []
if sat_key:
candidate_keys.extend([
sat_key,
_normalize_satellite_name(sat_key),
])
if tracked and tracked.get('name'):
candidate_keys.extend([
tracked['name'],
_normalize_satellite_name(tracked['name']),
])
seen: set[str] = set()
for key in candidate_keys:
norm = _normalize_satellite_name(key)
if norm in seen:
continue
seen.add(norm)
if key in _tle_cache:
tle_data = _tle_cache[key]
break
if norm in _tle_cache:
tle_data = _tle_cache[norm]
break
if tle_data is None and tracked and tracked.get('tle_line1') and tracked.get('tle_line2'):
display_name = tracked.get('name') or sat_key or str(norad_id or 'UNKNOWN')
tle_data = (display_name, tracked['tle_line1'], tracked['tle_line2'])
_tle_cache[_normalize_satellite_name(display_name)] = tle_data
if tle_data is None and sat_key:
normalized = _normalize_satellite_name(sat_key)
for key, value in _tle_cache.items():
if key == normalized or _normalize_satellite_name(value[0]) == normalized:
tle_data = value
break
display_name = _BUILTIN_NORAD_TO_KEY.get(norad_id or -1)
if not display_name:
display_name = (tracked.get('name') if tracked else None) or (tle_data[0] if tle_data else None) or (sat_key if sat_key else str(norad_id or 'UNKNOWN'))
return display_name, norad_id, tle_data
def _make_pass_cache_key(
lat: float,
lon: float,
hours: int,
min_el: float,
resolved_satellites: list[tuple[str, int, tuple[str, str, str]]],
) -> tuple:
"""Build a stable cache key for predicted passes."""
return (
round(lat, 4),
round(lon, 4),
int(hours),
round(float(min_el), 1),
tuple(
(
sat_name,
norad_id,
tle_data[1][:32],
tle_data[2][:32],
)
for sat_name, norad_id, tle_data in resolved_satellites
),
)
except Exception as e:
logger.warning(f"Failed to load DB satellites into TLE cache: {e}")
def _normalize_satellite_name(value: object) -> str:
"""Normalize satellite identifiers for loose name matching."""
return str(value or '').strip().replace(' ', '-').upper()
def _get_tracked_satellite_maps() -> tuple[dict[int, dict], dict[str, dict]]:
"""Return tracked satellites indexed by NORAD ID and normalized name."""
by_norad: dict[int, dict] = {}
by_name: dict[str, dict] = {}
try:
for sat in get_tracked_satellites():
try:
norad_id = int(sat['norad_id'])
except (TypeError, ValueError):
continue
by_norad[norad_id] = sat
by_name[_normalize_satellite_name(sat.get('name'))] = sat
except Exception as e:
logger.warning(f"Failed to read tracked satellites for lookup: {e}")
return by_norad, by_name
def _resolve_satellite_request(sat: object, tracked_by_norad: dict[int, dict], tracked_by_name: dict[str, dict]) -> tuple[str, int | None, tuple[str, str, str] | None]:
"""Resolve a satellite request to display name, NORAD ID, and TLE data."""
norad_id: int | None = None
sat_key: str | None = None
tracked: dict | None = None
if isinstance(sat, int):
norad_id = sat
elif isinstance(sat, str):
stripped = sat.strip()
if stripped.isdigit():
norad_id = int(stripped)
else:
sat_key = stripped
if norad_id is not None:
tracked = tracked_by_norad.get(norad_id)
sat_key = _BUILTIN_NORAD_TO_KEY.get(norad_id) or (tracked.get('name') if tracked else str(norad_id))
else:
normalized = _normalize_satellite_name(sat_key)
tracked = tracked_by_name.get(normalized)
if tracked:
try:
norad_id = int(tracked['norad_id'])
except (TypeError, ValueError):
norad_id = None
sat_key = tracked.get('name') or sat_key
tle_data = None
candidate_keys: list[str] = []
if sat_key:
candidate_keys.extend([
sat_key,
_normalize_satellite_name(sat_key),
])
if tracked and tracked.get('name'):
candidate_keys.extend([
tracked['name'],
_normalize_satellite_name(tracked['name']),
])
seen: set[str] = set()
for key in candidate_keys:
norm = _normalize_satellite_name(key)
if norm in seen:
continue
seen.add(norm)
if key in _tle_cache:
tle_data = _tle_cache[key]
break
if norm in _tle_cache:
tle_data = _tle_cache[norm]
break
if tle_data is None and tracked and tracked.get('tle_line1') and tracked.get('tle_line2'):
display_name = tracked.get('name') or sat_key or str(norad_id or 'UNKNOWN')
tle_data = (display_name, tracked['tle_line1'], tracked['tle_line2'])
_tle_cache[_normalize_satellite_name(display_name)] = tle_data
if tle_data is None and sat_key:
normalized = _normalize_satellite_name(sat_key)
for key, value in _tle_cache.items():
if key == normalized or _normalize_satellite_name(value[0]) == normalized:
tle_data = value
break
display_name = _BUILTIN_NORAD_TO_KEY.get(norad_id or -1)
if not display_name:
display_name = (tracked.get('name') if tracked else None) or (tle_data[0] if tle_data else None) or (sat_key if sat_key else str(norad_id or 'UNKNOWN'))
return display_name, norad_id, tle_data
def _make_pass_cache_key(
lat: float,
lon: float,
hours: int,
min_el: float,
resolved_satellites: list[tuple[str, int, tuple[str, str, str]]],
) -> tuple:
"""Build a stable cache key for predicted passes."""
return (
round(lat, 4),
round(lon, 4),
int(hours),
round(float(min_el), 1),
tuple(
(
sat_name,
norad_id,
tle_data[1][:32],
tle_data[2][:32],
)
for sat_name, norad_id, tle_data in resolved_satellites
),
)
def _start_satellite_tracker():
@@ -218,11 +218,6 @@ def _start_satellite_tracker():
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 = []
@@ -245,24 +240,18 @@ def _start_satellite_tracker():
geocentric = satellite.at(now)
subpoint = wgs84.subpoint(geocentric)
# SSE stream is server-wide and cannot know per-client observer
# location. Observer-relative fields (elevation, azimuth, distance,
# visible) are intentionally omitted here — the per-client HTTP poll
# at /satellite/position owns those using the client's actual location.
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,
'altitude': float(subpoint.elevation.km),
}
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)
@@ -372,13 +361,13 @@ def _fetch_iss_realtime(observer_lat: float | None = None, observer_lon: float |
if iss_lat is None:
return None
result = {
'satellite': 'ISS',
'norad_id': 25544,
'lat': iss_lat,
'lon': iss_lon,
'altitude': iss_alt,
'source': source
result = {
'satellite': 'ISS',
'norad_id': 25544,
'lat': iss_lat,
'lon': iss_lon,
'altitude': iss_alt,
'source': source
}
# Calculate observer-relative data if location provided
@@ -422,131 +411,131 @@ def _fetch_iss_realtime(observer_lat: float | None = None, observer_lon: float |
return result
@satellite_bp.route('/dashboard')
def satellite_dashboard():
"""Popout satellite tracking dashboard."""
embedded = request.args.get('embedded', 'false') == 'true'
response = make_response(render_template(
'satellite_dashboard.html',
shared_observer_location=SHARED_OBSERVER_LOCATION_ENABLED,
embedded=embedded,
))
response.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate, max-age=0'
response.headers['Pragma'] = 'no-cache'
response.headers['Expires'] = '0'
return response
@satellite_bp.route('/dashboard')
def satellite_dashboard():
"""Popout satellite tracking dashboard."""
embedded = request.args.get('embedded', 'false') == 'true'
response = make_response(render_template(
'satellite_dashboard.html',
shared_observer_location=SHARED_OBSERVER_LOCATION_ENABLED,
embedded=embedded,
))
response.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate, max-age=0'
response.headers['Pragma'] = 'no-cache'
response.headers['Expires'] = '0'
return response
@satellite_bp.route('/predict', methods=['POST'])
def predict_passes():
"""Calculate satellite passes using skyfield."""
try:
from skyfield.api import EarthSatellite, wgs84
except ImportError:
return jsonify({
'status': 'error',
'message': 'skyfield library not installed. Run: pip install skyfield'
}), 503
from utils.satellite_predict import predict_passes as _predict_passes
data = request.json or {}
try:
# Validate inputs
lat = validate_latitude(data.get('latitude', data.get('lat', 51.5074)))
lon = validate_longitude(data.get('longitude', data.get('lon', -0.1278)))
hours = validate_hours(data.get('hours', 24))
min_el = validate_elevation(data.get('minEl', 10))
except ValueError as e:
return api_error(str(e), 400)
try:
sat_input = data.get('satellites', ['ISS', 'METEOR-M2-3', 'METEOR-M2-4'])
passes = []
colors = {
'ISS': '#00ffff',
'METEOR-M2': '#9370DB',
'METEOR-M2-3': '#ff00ff',
'METEOR-M2-4': '#00ff88',
}
tracked_by_norad, tracked_by_name = _get_tracked_satellite_maps()
resolved_satellites: list[tuple[str, int, tuple[str, str, str]]] = []
for sat in sat_input:
sat_name, norad_id, tle_data = _resolve_satellite_request(
sat,
tracked_by_norad,
tracked_by_name,
)
if not tle_data:
continue
resolved_satellites.append((sat_name, norad_id or 0, tle_data))
if not resolved_satellites:
return jsonify({
'status': 'success',
'passes': [],
'cached': False,
})
cache_key = _make_pass_cache_key(lat, lon, hours, min_el, resolved_satellites)
cached = _pass_cache.get(cache_key)
now_ts = time.time()
if cached and (now_ts - cached[1]) < _PASS_CACHE_TTL:
return jsonify({
'status': 'success',
'passes': cached[0],
'cached': True,
})
ts = _get_timescale()
observer = wgs84.latlon(lat, lon)
t0 = ts.now()
t1 = ts.utc(t0.utc_datetime() + timedelta(hours=hours))
for sat_name, norad_id, tle_data in resolved_satellites:
current_pos = None
try:
satellite = EarthSatellite(tle_data[1], tle_data[2], tle_data[0], ts)
geo = satellite.at(t0)
sp = wgs84.subpoint(geo)
current_pos = {
'lat': float(sp.latitude.degrees),
'lon': float(sp.longitude.degrees),
}
except Exception:
pass
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'] = norad_id
p['color'] = colors.get(sat_name, '#00ff00')
if current_pos:
p['currentPos'] = current_pos
passes.extend(sat_passes)
passes.sort(key=lambda p: p['startTimeISO'])
_pass_cache[cache_key] = (passes, now_ts)
return jsonify({
'status': 'success',
'passes': passes,
'cached': False,
})
except Exception as exc:
logger.exception('Satellite pass calculation failed')
if 'cache_key' in locals():
stale_cached = _pass_cache.get(cache_key)
if stale_cached and stale_cached[0]:
return jsonify({
'status': 'success',
'passes': stale_cached[0],
'cached': True,
'stale': True,
})
return api_error(f'Failed to calculate passes: {exc}', 500)
@satellite_bp.route('/predict', methods=['POST'])
def predict_passes():
"""Calculate satellite passes using skyfield."""
try:
from skyfield.api import EarthSatellite, wgs84
except ImportError:
return jsonify({
'status': 'error',
'message': 'skyfield library not installed. Run: pip install skyfield'
}), 503
from utils.satellite_predict import predict_passes as _predict_passes
data = request.json or {}
try:
# Validate inputs
lat = validate_latitude(data.get('latitude', data.get('lat', 51.5074)))
lon = validate_longitude(data.get('longitude', data.get('lon', -0.1278)))
hours = validate_hours(data.get('hours', 24))
min_el = validate_elevation(data.get('minEl', 10))
except ValueError as e:
return api_error(str(e), 400)
try:
sat_input = data.get('satellites', ['ISS', 'METEOR-M2-3', 'METEOR-M2-4'])
passes = []
colors = {
'ISS': '#00ffff',
'METEOR-M2': '#9370DB',
'METEOR-M2-3': '#ff00ff',
'METEOR-M2-4': '#00ff88',
}
tracked_by_norad, tracked_by_name = _get_tracked_satellite_maps()
resolved_satellites: list[tuple[str, int, tuple[str, str, str]]] = []
for sat in sat_input:
sat_name, norad_id, tle_data = _resolve_satellite_request(
sat,
tracked_by_norad,
tracked_by_name,
)
if not tle_data:
continue
resolved_satellites.append((sat_name, norad_id or 0, tle_data))
if not resolved_satellites:
return jsonify({
'status': 'success',
'passes': [],
'cached': False,
})
cache_key = _make_pass_cache_key(lat, lon, hours, min_el, resolved_satellites)
cached = _pass_cache.get(cache_key)
now_ts = time.time()
if cached and (now_ts - cached[1]) < _PASS_CACHE_TTL:
return jsonify({
'status': 'success',
'passes': cached[0],
'cached': True,
})
ts = _get_timescale()
observer = wgs84.latlon(lat, lon)
t0 = ts.now()
t1 = ts.utc(t0.utc_datetime() + timedelta(hours=hours))
for sat_name, norad_id, tle_data in resolved_satellites:
current_pos = None
try:
satellite = EarthSatellite(tle_data[1], tle_data[2], tle_data[0], ts)
geo = satellite.at(t0)
sp = wgs84.subpoint(geo)
current_pos = {
'lat': float(sp.latitude.degrees),
'lon': float(sp.longitude.degrees),
}
except Exception:
pass
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'] = norad_id
p['color'] = colors.get(sat_name, '#00ff00')
if current_pos:
p['currentPos'] = current_pos
passes.extend(sat_passes)
passes.sort(key=lambda p: p['startTimeISO'])
_pass_cache[cache_key] = (passes, now_ts)
return jsonify({
'status': 'success',
'passes': passes,
'cached': False,
})
except Exception as exc:
logger.exception('Satellite pass calculation failed')
if 'cache_key' in locals():
stale_cached = _pass_cache.get(cache_key)
if stale_cached and stale_cached[0]:
return jsonify({
'status': 'success',
'passes': stale_cached[0],
'cached': True,
'stale': True,
})
return api_error(f'Failed to calculate passes: {exc}', 500)
@satellite_bp.route('/position', methods=['POST'])
@@ -566,63 +555,63 @@ def get_satellite_position():
except ValueError as e:
return api_error(str(e), 400)
sat_input = data.get('satellites', [])
include_track = bool(data.get('includeTrack', True))
prefer_realtime_api = bool(data.get('preferRealtimeApi', False))
sat_input = data.get('satellites', [])
include_track = bool(data.get('includeTrack', True))
prefer_realtime_api = bool(data.get('preferRealtimeApi', False))
observer = wgs84.latlon(lat, lon)
ts = None
now = None
now_dt = None
tracked_by_norad, tracked_by_name = _get_tracked_satellite_maps()
positions = []
for sat in sat_input:
sat_name, norad_id, tle_data = _resolve_satellite_request(sat, tracked_by_norad, tracked_by_name)
# Optional special handling for ISS. The dashboard does not enable this
# because external API latency can make live updates stall.
if prefer_realtime_api and (norad_id == 25544 or sat_name == 'ISS'):
iss_data = _fetch_iss_realtime(lat, lon)
if iss_data:
# Add orbit track if requested (using TLE for track prediction)
if include_track and 'ISS' in _tle_cache:
try:
if ts is None:
ts = _get_timescale()
now = ts.now()
now_dt = now.utc_datetime()
tle_data = _tle_cache['ISS']
satellite = EarthSatellite(tle_data[1], tle_data[2], tle_data[0], ts)
orbit_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)
orbit_track.append({
'lat': float(sp.latitude.degrees),
'lon': float(sp.longitude.degrees),
'past': minutes_offset < 0
})
except Exception:
continue
iss_data['track'] = orbit_track
except Exception:
pass
positions.append(iss_data)
continue
# Other satellites - use TLE data
if not tle_data:
continue
try:
if ts is None:
ts = _get_timescale()
now = ts.now()
now_dt = now.utc_datetime()
satellite = EarthSatellite(tle_data[1], tle_data[2], tle_data[0], ts)
observer = wgs84.latlon(lat, lon)
ts = None
now = None
now_dt = None
tracked_by_norad, tracked_by_name = _get_tracked_satellite_maps()
positions = []
for sat in sat_input:
sat_name, norad_id, tle_data = _resolve_satellite_request(sat, tracked_by_norad, tracked_by_name)
# Optional special handling for ISS. The dashboard does not enable this
# because external API latency can make live updates stall.
if prefer_realtime_api and (norad_id == 25544 or sat_name == 'ISS'):
iss_data = _fetch_iss_realtime(lat, lon)
if iss_data:
# Add orbit track if requested (using TLE for track prediction)
if include_track and 'ISS' in _tle_cache:
try:
if ts is None:
ts = _get_timescale()
now = ts.now()
now_dt = now.utc_datetime()
tle_data = _tle_cache['ISS']
satellite = EarthSatellite(tle_data[1], tle_data[2], tle_data[0], ts)
orbit_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)
orbit_track.append({
'lat': float(sp.latitude.degrees),
'lon': float(sp.longitude.degrees),
'past': minutes_offset < 0
})
except Exception:
continue
iss_data['track'] = orbit_track
except Exception:
pass
positions.append(iss_data)
continue
# Other satellites - use TLE data
if not tle_data:
continue
try:
if ts is None:
ts = _get_timescale()
now = ts.now()
now_dt = now.utc_datetime()
satellite = EarthSatellite(tle_data[1], tle_data[2], tle_data[0], ts)
geocentric = satellite.at(now)
subpoint = wgs84.subpoint(geocentric)
@@ -631,23 +620,23 @@ def get_satellite_position():
topocentric = diff.at(now)
alt, az, distance = topocentric.altaz()
pos_data = {
'satellite': sat_name,
'norad_id': norad_id,
'lat': float(subpoint.latitude.degrees),
'lon': float(subpoint.longitude.degrees),
'altitude': float(geocentric.distance().km - 6371),
pos_data = {
'satellite': sat_name,
'norad_id': norad_id,
'lat': float(subpoint.latitude.degrees),
'lon': float(subpoint.longitude.degrees),
'altitude': float(geocentric.distance().km - 6371),
'elevation': float(alt.degrees),
'azimuth': float(az.degrees),
'distance': float(distance.km),
'visible': bool(alt.degrees > 0)
}
if include_track:
orbit_track = []
for minutes_offset in range(-45, 46, 1):
t_point = ts.utc(now_dt + timedelta(minutes=minutes_offset))
try:
if include_track:
orbit_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)
orbit_track.append({
@@ -655,13 +644,13 @@ def get_satellite_position():
'lon': float(sp.longitude.degrees),
'past': minutes_offset < 0
})
except Exception:
continue
pos_data['track'] = orbit_track
pos_data['groundTrack'] = orbit_track
positions.append(pos_data)
except Exception:
continue
pos_data['track'] = orbit_track
pos_data['groundTrack'] = orbit_track
positions.append(pos_data)
except Exception:
continue

51
tests/test_satellite.py
View File

@@ -1,3 +1,5 @@
import queue
import threading
from unittest.mock import MagicMock, patch
import pytest
@@ -70,6 +72,55 @@ def test_get_satellite_position_skyfield_error(mock_load, client):
assert response.status_code == 200
assert response.json['positions'] == []
def test_tracker_position_has_no_observer_fields():
"""SSE tracker positions must NOT include observer-relative fields.
The tracker runs server-side with a fixed (potentially wrong) observer
location. Only the per-request /satellite/position endpoint, which
receives the client's actual location, should emit elevation/azimuth/
distance/visible.
"""
import sys
from routes.satellite import _start_satellite_tracker
ISS_TLE = (
'ISS (ZARYA)',
'1 25544U 98067A 24001.00000000 .00016717 00000-0 30171-3 0 9993',
'2 25544 51.6416 20.4567 0004561 45.3212 67.8912 15.49876543123457',
)
sat_q = queue.Queue(maxsize=5)
mock_app = MagicMock()
mock_app.satellite_queue = sat_q
from skyfield.api import load as _real_load
real_ts = _real_load.timescale(builtin=True)
# Pre-populate track cache so the tracker loop doesn't block computing 90 points
tle_key = (ISS_TLE[0], ISS_TLE[1][:20])
stub_track = [{'lat': 0.0, 'lon': float(i), 'past': i < 45} for i in range(91)]
with patch('routes.satellite._tle_cache', {'ISS': ISS_TLE}), \
patch('routes.satellite.get_tracked_satellites') as mock_tracked, \
patch('routes.satellite._track_cache', {tle_key: (stub_track, 1e18)}), \
patch('routes.satellite._get_timescale', return_value=real_ts), \
patch.dict('sys.modules', {'app': mock_app}):
mock_tracked.return_value = [{
'name': 'ISS (ZARYA)', 'norad_id': 25544,
'tle_line1': ISS_TLE[1], 'tle_line2': ISS_TLE[2],
}]
t = threading.Thread(target=_start_satellite_tracker, daemon=True)
t.start()
msg = sat_q.get(timeout=10)
assert msg['type'] == 'positions'
pos = msg['positions'][0]
for forbidden in ('elevation', 'azimuth', 'distance', 'visible'):
assert forbidden not in pos, f"SSE tracker must not emit '{forbidden}'"
for required in ('lat', 'lon', 'altitude', 'satellite', 'norad_id'):
assert required in pos, f"SSE tracker must emit '{required}'"
# Logic Integration Test (Simulating prediction)
def test_predict_passes_empty_cache(client):
"""Verify that if the satellite is not in cache, no passes are returned."""