Mirror/intercept
1
0
Fork 0
mirror of https://github.com/smittix/intercept.git synced 2026-07-17 05:48:11 -07:00
Code Issues Packages Projects Releases Wiki Activity

refactor: weather sat prediction reads TLEs from unified store

Browse Source 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
James Smith
2026-06-11 17:23:20 +01:00
parent 1a60c52fca
commit c8f2bc5a88
2 changed files with 217 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_weather_sat_predict.py
+169 -179
View File
@@ -17,13 +17,13 @@ from utils.weather_sat_predict import _format_utc_iso, predict_passes
# NOAA-18 was decommissioned Jun 2025 and is inactive in the real WEATHER_SATELLITES, # NOAA-18 was decommissioned Jun 2025 and is inactive in the real WEATHER_SATELLITES,
# so tests that assert on satellite-specific fields patch the module-level name. # so tests that assert on satellite-specific fields patch the module-level name.
_MOCK_WEATHER_SATS = { _MOCK_WEATHER_SATS = {
'NOAA-18': { "NOAA-18": {
'name': 'NOAA 18', "name": "NOAA 18",
'frequency': 137.9125, "frequency": 137.9125,
'mode': 'APT', "mode": "APT",
'pipeline': 'noaa_apt', "pipeline": "noaa_apt",
'tle_key': 'NOAA-18', "tle_key": "NOAA-18",
'active': True, "active": True,
} }
} }
@@ -31,8 +31,8 @@ _MOCK_WEATHER_SATS = {
class TestPredictPasses: class TestPredictPasses:
"""Tests for predict_passes() function.""" """Tests for predict_passes() function."""
@patch('utils.weather_sat_predict.load') @patch("utils.weather_sat_predict.load")
@patch('utils.weather_sat_predict.TLE_SATELLITES') @patch("utils.weather_sat_predict.TLE_SATELLITES")
def test_predict_passes_no_tle_data(self, mock_tle, mock_load): def test_predict_passes_no_tle_data(self, mock_tle, mock_load):
"""predict_passes() should handle missing TLE data.""" """predict_passes() should handle missing TLE data."""
mock_tle.get.return_value = None mock_tle.get.return_value = None
@@ -45,12 +45,12 @@ class TestPredictPasses:
assert passes == [] assert passes == []
@patch('utils.weather_sat_predict.WEATHER_SATELLITES', _MOCK_WEATHER_SATS) @patch("utils.weather_sat_predict.WEATHER_SATELLITES", _MOCK_WEATHER_SATS)
@patch('utils.weather_sat_predict.load') @patch("utils.weather_sat_predict.load")
@patch('utils.weather_sat_predict.TLE_SATELLITES') @patch("utils.weather_sat_predict.TLE_SATELLITES")
@patch('utils.weather_sat_predict.wgs84') @patch("utils.weather_sat_predict.wgs84")
@patch('utils.weather_sat_predict.EarthSatellite') @patch("utils.weather_sat_predict.EarthSatellite")
@patch('utils.weather_sat_predict.find_discrete') @patch("utils.weather_sat_predict.find_discrete")
def test_predict_passes_basic(self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load): def test_predict_passes_basic(self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load):
"""predict_passes() should predict basic passes.""" """predict_passes() should predict basic passes."""
# Mock timescale # Mock timescale
@@ -64,9 +64,9 @@ class TestPredictPasses:
# Mock TLE data # Mock TLE data
mock_tle.get.return_value = ( mock_tle.get.return_value = (
'NOAA-18', "NOAA-18",
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999', "1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999",
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000' "2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000",
) )
# Mock observer # Mock observer
@@ -103,23 +103,21 @@ class TestPredictPasses:
assert len(passes) == 1 assert len(passes) == 1
pass_data = passes[0] pass_data = passes[0]
assert pass_data['satellite'] == 'NOAA-18' assert pass_data["satellite"] == "NOAA-18"
assert pass_data['name'] == 'NOAA 18' assert pass_data["name"] == "NOAA 18"
assert pass_data['frequency'] == 137.9125 assert pass_data["frequency"] == 137.9125
assert pass_data['mode'] == 'APT' assert pass_data["mode"] == "APT"
assert 'maxEl' in pass_data assert "maxEl" in pass_data
assert 'duration' in pass_data assert "duration" in pass_data
assert 'quality' in pass_data assert "quality" in pass_data
@patch('utils.weather_sat_predict.WEATHER_SATELLITES', _MOCK_WEATHER_SATS) @patch("utils.weather_sat_predict.WEATHER_SATELLITES", _MOCK_WEATHER_SATS)
@patch('utils.weather_sat_predict.load') @patch("utils.weather_sat_predict.load")
@patch('utils.weather_sat_predict.TLE_SATELLITES') @patch("utils.weather_sat_predict.TLE_SATELLITES")
@patch('utils.weather_sat_predict.wgs84') @patch("utils.weather_sat_predict.wgs84")
@patch('utils.weather_sat_predict.EarthSatellite') @patch("utils.weather_sat_predict.EarthSatellite")
@patch('utils.weather_sat_predict.find_discrete') @patch("utils.weather_sat_predict.find_discrete")
def test_predict_passes_below_min_elevation( def test_predict_passes_below_min_elevation(self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load):
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should filter passes below min elevation.""" """predict_passes() should filter passes below min elevation."""
mock_ts = MagicMock() mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
@@ -130,9 +128,9 @@ class TestPredictPasses:
mock_load.timescale.return_value = mock_ts mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = ( mock_tle.get.return_value = (
'NOAA-18', "NOAA-18",
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999', "1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999",
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000' "2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000",
) )
mock_observer = MagicMock() mock_observer = MagicMock()
@@ -166,15 +164,13 @@ class TestPredictPasses:
assert len(passes) == 0 assert len(passes) == 0
@patch('utils.weather_sat_predict.WEATHER_SATELLITES', _MOCK_WEATHER_SATS) @patch("utils.weather_sat_predict.WEATHER_SATELLITES", _MOCK_WEATHER_SATS)
@patch('utils.weather_sat_predict.load') @patch("utils.weather_sat_predict.load")
@patch('utils.weather_sat_predict.TLE_SATELLITES') @patch("utils.weather_sat_predict.TLE_SATELLITES")
@patch('utils.weather_sat_predict.wgs84') @patch("utils.weather_sat_predict.wgs84")
@patch('utils.weather_sat_predict.EarthSatellite') @patch("utils.weather_sat_predict.EarthSatellite")
@patch('utils.weather_sat_predict.find_discrete') @patch("utils.weather_sat_predict.find_discrete")
def test_predict_passes_with_trajectory( def test_predict_passes_with_trajectory(self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load):
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should include trajectory when requested.""" """predict_passes() should include trajectory when requested."""
mock_ts = MagicMock() mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
@@ -185,9 +181,9 @@ class TestPredictPasses:
mock_load.timescale.return_value = mock_ts mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = ( mock_tle.get.return_value = (
'NOAA-18', "NOAA-18",
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999', "1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999",
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000' "2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000",
) )
mock_observer = MagicMock() mock_observer = MagicMock()
@@ -216,23 +212,19 @@ class TestPredictPasses:
mock_diff.at.side_effect = mock_topocentric mock_diff.at.side_effect = mock_topocentric
mock_satellite_obj.__sub__.return_value = mock_diff mock_satellite_obj.__sub__.return_value = mock_diff
passes = predict_passes( passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15, include_trajectory=True)
lat=51.5, lon=-0.1, hours=24, min_elevation=15, include_trajectory=True
)
assert len(passes) == 1 assert len(passes) == 1
assert 'trajectory' in passes[0] assert "trajectory" in passes[0]
assert len(passes[0]['trajectory']) == 30 assert len(passes[0]["trajectory"]) == 30
@patch('utils.weather_sat_predict.WEATHER_SATELLITES', _MOCK_WEATHER_SATS) @patch("utils.weather_sat_predict.WEATHER_SATELLITES", _MOCK_WEATHER_SATS)
@patch('utils.weather_sat_predict.load') @patch("utils.weather_sat_predict.load")
@patch('utils.weather_sat_predict.TLE_SATELLITES') @patch("utils.weather_sat_predict.TLE_SATELLITES")
@patch('utils.weather_sat_predict.wgs84') @patch("utils.weather_sat_predict.wgs84")
@patch('utils.weather_sat_predict.EarthSatellite') @patch("utils.weather_sat_predict.EarthSatellite")
@patch('utils.weather_sat_predict.find_discrete') @patch("utils.weather_sat_predict.find_discrete")
def test_predict_passes_with_ground_track( def test_predict_passes_with_ground_track(self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load):
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should include ground track when requested.""" """predict_passes() should include ground track when requested."""
mock_ts = MagicMock() mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
@@ -243,9 +235,9 @@ class TestPredictPasses:
mock_load.timescale.return_value = mock_ts mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = ( mock_tle.get.return_value = (
'NOAA-18', "NOAA-18",
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999', "1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999",
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000' "2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000",
) )
mock_observer = MagicMock() mock_observer = MagicMock()
@@ -291,23 +283,19 @@ class TestPredictPasses:
mock_subpoint.longitude = mock_lon mock_subpoint.longitude = mock_lon
mock_wgs84.subpoint.return_value = mock_subpoint mock_wgs84.subpoint.return_value = mock_subpoint
passes = predict_passes( passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15, include_ground_track=True)
lat=51.5, lon=-0.1, hours=24, min_elevation=15, include_ground_track=True
)
assert len(passes) == 1 assert len(passes) == 1
assert 'groundTrack' in passes[0] assert "groundTrack" in passes[0]
assert len(passes[0]['groundTrack']) == 60 assert len(passes[0]["groundTrack"]) == 60
@patch('utils.weather_sat_predict.WEATHER_SATELLITES', _MOCK_WEATHER_SATS) @patch("utils.weather_sat_predict.WEATHER_SATELLITES", _MOCK_WEATHER_SATS)
@patch('utils.weather_sat_predict.load') @patch("utils.weather_sat_predict.load")
@patch('utils.weather_sat_predict.TLE_SATELLITES') @patch("utils.weather_sat_predict.TLE_SATELLITES")
@patch('utils.weather_sat_predict.wgs84') @patch("utils.weather_sat_predict.wgs84")
@patch('utils.weather_sat_predict.EarthSatellite') @patch("utils.weather_sat_predict.EarthSatellite")
@patch('utils.weather_sat_predict.find_discrete') @patch("utils.weather_sat_predict.find_discrete")
def test_predict_passes_quality_excellent( def test_predict_passes_quality_excellent(self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load):
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should mark high elevation passes as excellent.""" """predict_passes() should mark high elevation passes as excellent."""
mock_ts = MagicMock() mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
@@ -318,9 +306,9 @@ class TestPredictPasses:
mock_load.timescale.return_value = mock_ts mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = ( mock_tle.get.return_value = (
'NOAA-18', "NOAA-18",
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999', "1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999",
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000' "2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000",
) )
mock_observer = MagicMock() mock_observer = MagicMock()
@@ -352,18 +340,16 @@ class TestPredictPasses:
passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15) passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
assert len(passes) == 1 assert len(passes) == 1
assert passes[0]['quality'] == 'excellent' assert passes[0]["quality"] == "excellent"
assert passes[0]['maxEl'] >= 60 assert passes[0]["maxEl"] >= 60
@patch('utils.weather_sat_predict.WEATHER_SATELLITES', _MOCK_WEATHER_SATS) @patch("utils.weather_sat_predict.WEATHER_SATELLITES", _MOCK_WEATHER_SATS)
@patch('utils.weather_sat_predict.load') @patch("utils.weather_sat_predict.load")
@patch('utils.weather_sat_predict.TLE_SATELLITES') @patch("utils.weather_sat_predict.TLE_SATELLITES")
@patch('utils.weather_sat_predict.wgs84') @patch("utils.weather_sat_predict.wgs84")
@patch('utils.weather_sat_predict.EarthSatellite') @patch("utils.weather_sat_predict.EarthSatellite")
@patch('utils.weather_sat_predict.find_discrete') @patch("utils.weather_sat_predict.find_discrete")
def test_predict_passes_quality_good( def test_predict_passes_quality_good(self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load):
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should mark medium elevation passes as good.""" """predict_passes() should mark medium elevation passes as good."""
mock_ts = MagicMock() mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
@@ -374,9 +360,9 @@ class TestPredictPasses:
mock_load.timescale.return_value = mock_ts mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = ( mock_tle.get.return_value = (
'NOAA-18', "NOAA-18",
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999', "1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999",
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000' "2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000",
) )
mock_observer = MagicMock() mock_observer = MagicMock()
@@ -408,18 +394,16 @@ class TestPredictPasses:
passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15) passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
assert len(passes) == 1 assert len(passes) == 1
assert passes[0]['quality'] == 'good' assert passes[0]["quality"] == "good"
assert 30 <= passes[0]['maxEl'] < 60 assert 30 <= passes[0]["maxEl"] < 60
@patch('utils.weather_sat_predict.WEATHER_SATELLITES', _MOCK_WEATHER_SATS) @patch("utils.weather_sat_predict.WEATHER_SATELLITES", _MOCK_WEATHER_SATS)
@patch('utils.weather_sat_predict.load') @patch("utils.weather_sat_predict.load")
@patch('utils.weather_sat_predict.TLE_SATELLITES') @patch("utils.weather_sat_predict.TLE_SATELLITES")
@patch('utils.weather_sat_predict.wgs84') @patch("utils.weather_sat_predict.wgs84")
@patch('utils.weather_sat_predict.EarthSatellite') @patch("utils.weather_sat_predict.EarthSatellite")
@patch('utils.weather_sat_predict.find_discrete') @patch("utils.weather_sat_predict.find_discrete")
def test_predict_passes_quality_fair( def test_predict_passes_quality_fair(self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load):
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should mark low elevation passes as fair.""" """predict_passes() should mark low elevation passes as fair."""
mock_ts = MagicMock() mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
@@ -430,9 +414,9 @@ class TestPredictPasses:
mock_load.timescale.return_value = mock_ts mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = ( mock_tle.get.return_value = (
'NOAA-18', "NOAA-18",
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999', "1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999",
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000' "2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000",
) )
mock_observer = MagicMock() mock_observer = MagicMock()
@@ -464,17 +448,15 @@ class TestPredictPasses:
passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15) passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
assert len(passes) == 1 assert len(passes) == 1
assert passes[0]['quality'] == 'fair' assert passes[0]["quality"] == "fair"
assert passes[0]['maxEl'] < 30 assert passes[0]["maxEl"] < 30
@patch('utils.weather_sat_predict.load') @patch("utils.weather_sat_predict.load")
@patch('utils.weather_sat_predict.TLE_SATELLITES') @patch("utils.weather_sat_predict.TLE_SATELLITES")
@patch('utils.weather_sat_predict.wgs84') @patch("utils.weather_sat_predict.wgs84")
@patch('utils.weather_sat_predict.EarthSatellite') @patch("utils.weather_sat_predict.EarthSatellite")
@patch('utils.weather_sat_predict.find_discrete') @patch("utils.weather_sat_predict.find_discrete")
def test_predict_passes_inactive_satellite( def test_predict_passes_inactive_satellite(self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load):
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should skip inactive satellites.""" """predict_passes() should skip inactive satellites."""
mock_ts = MagicMock() mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
@@ -485,25 +467,24 @@ class TestPredictPasses:
# Temporarily mark satellite as inactive # Temporarily mark satellite as inactive
from utils.weather_sat import WEATHER_SATELLITES from utils.weather_sat import WEATHER_SATELLITES
original_active = WEATHER_SATELLITES['NOAA-18']['active']
WEATHER_SATELLITES['NOAA-18']['active'] = False original_active = WEATHER_SATELLITES["NOAA-18"]["active"]
WEATHER_SATELLITES["NOAA-18"]["active"] = False
try: try:
passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15) passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
# Should not include NOAA-18 # Should not include NOAA-18
noaa_18_passes = [p for p in passes if p['satellite'] == 'NOAA-18'] noaa_18_passes = [p for p in passes if p["satellite"] == "NOAA-18"]
assert len(noaa_18_passes) == 0 assert len(noaa_18_passes) == 0
finally: finally:
WEATHER_SATELLITES['NOAA-18']['active'] = original_active WEATHER_SATELLITES["NOAA-18"]["active"] = original_active
@patch('utils.weather_sat_predict.load') @patch("utils.weather_sat_predict.load")
@patch('utils.weather_sat_predict.TLE_SATELLITES') @patch("utils.weather_sat_predict.TLE_SATELLITES")
@patch('utils.weather_sat_predict.wgs84') @patch("utils.weather_sat_predict.wgs84")
@patch('utils.weather_sat_predict.EarthSatellite') @patch("utils.weather_sat_predict.EarthSatellite")
@patch('utils.weather_sat_predict.find_discrete') @patch("utils.weather_sat_predict.find_discrete")
def test_predict_passes_exception_handling( def test_predict_passes_exception_handling(self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load):
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should handle exceptions gracefully.""" """predict_passes() should handle exceptions gracefully."""
mock_ts = MagicMock() mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
@@ -514,9 +495,9 @@ class TestPredictPasses:
mock_load.timescale.return_value = mock_ts mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = ( mock_tle.get.return_value = (
'NOAA-18', "NOAA-18",
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999', "1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999",
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000' "2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000",
) )
mock_observer = MagicMock() mock_observer = MagicMock()
@@ -526,40 +507,41 @@ class TestPredictPasses:
mock_sat.return_value = mock_satellite_obj mock_sat.return_value = mock_satellite_obj
# Make find_discrete raise exception # Make find_discrete raise exception
mock_find.side_effect = Exception('Computation error') mock_find.side_effect = Exception("Computation error")
# Should not raise, just skip this satellite # Should not raise, just skip this satellite
passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15) passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
# May include passes from other satellites or be empty # May include passes from other satellites or be empty
assert isinstance(passes, list) assert isinstance(passes, list)
@patch('utils.weather_sat_predict.load') @patch("utils.weather_sat_predict.load")
@patch('utils.weather_sat_predict.TLE_SATELLITES') @patch("utils.weather_sat_predict.TLE_SATELLITES")
def test_predict_passes_uses_tle_cache(self, mock_tle, mock_load): def test_predict_passes_uses_tle_cache(self, mock_tle, mock_load):
"""predict_passes() should use live TLE cache if available.""" """predict_passes() should use live TLE store if available."""
with patch('utils.weather_sat_predict._tle_cache', {'NOAA-18': ('NOAA-18', 'line1', 'line2')}): with patch(
"utils.weather_sat_predict._get_tle_source", return_value={"NOAA-18": ("NOAA-18", "line1", "line2")}
):
mock_ts = MagicMock() mock_ts = MagicMock()
mock_ts.now.return_value = MagicMock() mock_ts.now.return_value = MagicMock()
mock_ts.utc.return_value = MagicMock() mock_ts.utc.return_value = MagicMock()
mock_load.timescale.return_value = mock_ts mock_load.timescale.return_value = mock_ts
# Even though TLE_SATELLITES is mocked, should use _tle_cache # Even though TLE_SATELLITES is mocked, should use the unified store
with patch('utils.weather_sat_predict.wgs84'), \ with (
patch('utils.weather_sat_predict.EarthSatellite'), \ patch("utils.weather_sat_predict.wgs84"),
patch('utils.weather_sat_predict.find_discrete', return_value=([], [])): patch("utils.weather_sat_predict.EarthSatellite"),
patch("utils.weather_sat_predict.find_discrete", return_value=([], [])),
):
predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15) predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
# Should not raise # Should not raise
@patch('utils.weather_sat_predict.WEATHER_SATELLITES', _MOCK_WEATHER_SATS) @patch("utils.weather_sat_predict.WEATHER_SATELLITES", _MOCK_WEATHER_SATS)
@patch('utils.weather_sat_predict.load') @patch("utils.weather_sat_predict.load")
@patch('utils.weather_sat_predict.TLE_SATELLITES') @patch("utils.weather_sat_predict.TLE_SATELLITES")
@patch('utils.weather_sat_predict.wgs84') @patch("utils.weather_sat_predict.wgs84")
@patch('utils.weather_sat_predict.EarthSatellite') @patch("utils.weather_sat_predict.EarthSatellite")
@patch('utils.weather_sat_predict.find_discrete') @patch("utils.weather_sat_predict.find_discrete")
def test_predict_passes_sorted_by_time( def test_predict_passes_sorted_by_time(self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load):
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should return passes sorted by start time.""" """predict_passes() should return passes sorted by start time."""
mock_ts = MagicMock() mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
@@ -570,9 +552,9 @@ class TestPredictPasses:
mock_load.timescale.return_value = mock_ts mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = ( mock_tle.get.return_value = (
'NOAA-18', "NOAA-18",
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999', "1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999",
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000' "2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000",
) )
mock_observer = MagicMock() mock_observer = MagicMock()
@@ -611,7 +593,7 @@ class TestPredictPasses:
# Should be sorted with earliest pass first # Should be sorted with earliest pass first
if len(passes) >= 2: if len(passes) >= 2:
assert passes[0]['startTimeISO'] < passes[1]['startTimeISO'] assert passes[0]["startTimeISO"] < passes[1]["startTimeISO"]
@staticmethod @staticmethod
def _mock_time(dt): def _mock_time(dt):
@@ -627,15 +609,13 @@ class TestPredictPasses:
class TestPassDataStructure: class TestPassDataStructure:
"""Tests for pass data structure.""" """Tests for pass data structure."""
@patch('utils.weather_sat_predict.WEATHER_SATELLITES', _MOCK_WEATHER_SATS) @patch("utils.weather_sat_predict.WEATHER_SATELLITES", _MOCK_WEATHER_SATS)
@patch('utils.weather_sat_predict.load') @patch("utils.weather_sat_predict.load")
@patch('utils.weather_sat_predict.TLE_SATELLITES') @patch("utils.weather_sat_predict.TLE_SATELLITES")
@patch('utils.weather_sat_predict.wgs84') @patch("utils.weather_sat_predict.wgs84")
@patch('utils.weather_sat_predict.EarthSatellite') @patch("utils.weather_sat_predict.EarthSatellite")
@patch('utils.weather_sat_predict.find_discrete') @patch("utils.weather_sat_predict.find_discrete")
def test_pass_data_fields( def test_pass_data_fields(self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load):
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""Pass data should contain all required fields.""" """Pass data should contain all required fields."""
mock_ts = MagicMock() mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
@@ -646,9 +626,9 @@ class TestPassDataStructure:
mock_load.timescale.return_value = mock_ts mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = ( mock_tle.get.return_value = (
'NOAA-18', "NOAA-18",
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999', "1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999",
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000' "2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000",
) )
mock_observer = MagicMock() mock_observer = MagicMock()
@@ -684,17 +664,27 @@ class TestPassDataStructure:
# Check all required fields # Check all required fields
required_fields = [ required_fields = [
'id', 'satellite', 'name', 'frequency', 'mode', "id",
'startTime', 'startTimeISO', 'endTimeISO', "satellite",
'maxEl', 'maxElAz', 'riseAz', 'setAz', "name",
'duration', 'quality' "frequency",
"mode",
"startTime",
"startTimeISO",
"endTimeISO",
"maxEl",
"maxElAz",
"riseAz",
"setAz",
"duration",
"quality",
] ]
for field in required_fields: for field in required_fields:
assert field in pass_data, f"Missing required field: {field}" assert field in pass_data, f"Missing required field: {field}"
def test_import_error_propagates(self): def test_import_error_propagates(self):
"""predict_passes() should raise ImportError if skyfield unavailable.""" """predict_passes() should raise ImportError if skyfield unavailable."""
with patch.dict('sys.modules', {'skyfield': None, 'skyfield.api': None}): with patch.dict("sys.modules", {"skyfield": None, "skyfield.api": None}):
with pytest.raises((ImportError, AttributeError)): with pytest.raises((ImportError, AttributeError)):
predict_passes(lat=51.5, lon=-0.1) predict_passes(lat=51.5, lon=-0.1)
@@ -706,11 +696,11 @@ class TestTimestampFormatting:
"""Aware UTC datetimes should not get a duplicate UTC suffix.""" """Aware UTC datetimes should not get a duplicate UTC suffix."""
dt = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) dt = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
value = _format_utc_iso(dt) value = _format_utc_iso(dt)
assert value == '2024-01-01T12:00:00Z' assert value == "2024-01-01T12:00:00Z"
assert '+00:00Z' not in value assert "+00:00Z" not in value
def test_format_utc_iso_from_naive_datetime(self): def test_format_utc_iso_from_naive_datetime(self):
"""Naive datetimes should be treated as UTC and serialized consistently.""" """Naive datetimes should be treated as UTC and serialized consistently."""
dt = datetime(2024, 1, 1, 12, 0, 0) dt = datetime(2024, 1, 1, 12, 0, 0)
value = _format_utc_iso(dt) value = _format_utc_iso(dt)
assert value == '2024-01-01T12:00:00Z' assert value == "2024-01-01T12:00:00Z"
utils/weather_sat_predict.py
+48 -44
View File
@@ -17,11 +17,7 @@ from data.satellites import TLE_SATELLITES
from utils.logging import get_logger from utils.logging import get_logger
from utils.weather_sat import WEATHER_SATELLITES from utils.weather_sat import WEATHER_SATELLITES
logger = get_logger('intercept.weather_sat_predict') logger = get_logger("intercept.weather_sat_predict")
# Live TLE cache — populated by routes/satellite.py at startup.
# Module-level so tests can patch it with patch('utils.weather_sat_predict._tle_cache', ...).
_tle_cache: dict = {}
def _format_utc_iso(dt: datetime.datetime) -> str: def _format_utc_iso(dt: datetime.datetime) -> str:
@@ -32,13 +28,16 @@ def _format_utc_iso(dt: datetime.datetime) -> str:
""" """
if dt.tzinfo is not None: if dt.tzinfo is not None:
dt = dt.astimezone(datetime.timezone.utc).replace(tzinfo=None) dt = dt.astimezone(datetime.timezone.utc).replace(tzinfo=None)
return dt.strftime('%Y-%m-%dT%H:%M:%SZ') return dt.strftime("%Y-%m-%dT%H:%M:%SZ")
def _get_tle_source() -> dict: def _get_tle_source() -> dict:
"""Return the best available TLE source (live cache preferred over static data).""" """Return the best available TLE source (unified store, static fallback)."""
if _tle_cache: from utils import tle_store
return _tle_cache
tles = tle_store.all_tles()
if tles:
return tles
return TLE_SATELLITES return TLE_SATELLITES
@@ -79,11 +78,11 @@ def predict_passes(
all_passes: list[dict[str, Any]] = [] all_passes: list[dict[str, Any]] = []
for sat_key, sat_info in WEATHER_SATELLITES.items(): for sat_key, sat_info in WEATHER_SATELLITES.items():
if not sat_info['active']: if not sat_info["active"]:
continue continue
try: try:
tle_data = tle_source.get(sat_info['tle_key']) tle_data = tle_source.get(sat_info["tle_key"])
if not tle_data: if not tle_data:
continue continue
@@ -104,18 +103,25 @@ def predict_passes(
rise_t = t rise_t = t
elif rise_t is not None: elif rise_t is not None:
_process_pass( _process_pass(
sat_key, sat_info, satellite, diff, ts, sat_key,
rise_t, t, min_elevation, sat_info,
include_trajectory, include_ground_track, satellite,
diff,
ts,
rise_t,
t,
min_elevation,
include_trajectory,
include_ground_track,
all_passes, all_passes,
) )
rise_t = None rise_t = None
except Exception as exc: except Exception as exc:
logger.debug('Error predicting passes for %s: %s', sat_key, exc) logger.debug("Error predicting passes for %s: %s", sat_key, exc)
continue continue
all_passes.sort(key=lambda p: p['startTimeISO']) all_passes.sort(key=lambda p: p["startTimeISO"])
return all_passes return all_passes
@@ -155,7 +161,7 @@ def _process_pass(
max_el = el max_el = el
max_el_az = az_deg max_el_az = az_deg
if include_trajectory: if include_trajectory:
traj_points.append({'az': round(az_deg, 1), 'el': round(max(0.0, el), 1)}) traj_points.append({"az": round(az_deg, 1), "el": round(max(0.0, el), 1)})
except Exception: except Exception:
pass pass
@@ -181,32 +187,28 @@ def _process_pass(
pass_id = f"{sat_key}_{aos_iso}" pass_id = f"{sat_key}_{aos_iso}"
pass_dict: dict[str, Any] = { pass_dict: dict[str, Any] = {
'id': pass_id, "id": pass_id,
'satellite': sat_key, "satellite": sat_key,
'name': sat_info['name'], "name": sat_info["name"],
'frequency': sat_info['frequency'], "frequency": sat_info["frequency"],
'mode': sat_info['mode'], "mode": sat_info["mode"],
'startTime': rise_dt.strftime('%Y-%m-%d %H:%M UTC'), "startTime": rise_dt.strftime("%Y-%m-%d %H:%M UTC"),
'startTimeISO': aos_iso, "startTimeISO": aos_iso,
'endTimeISO': _format_utc_iso(set_dt), "endTimeISO": _format_utc_iso(set_dt),
'maxEl': round(max_el, 1), "maxEl": round(max_el, 1),
'maxElAz': round(max_el_az, 1), "maxElAz": round(max_el_az, 1),
'riseAz': round(rise_az, 1), "riseAz": round(rise_az, 1),
'setAz': round(set_az, 1), "setAz": round(set_az, 1),
'duration': round(duration_secs, 1), "duration": round(duration_secs, 1),
'quality': ( "quality": ("excellent" if max_el >= 60 else "good" if max_el >= 30 else "fair"),
'excellent' if max_el >= 60
else 'good' if max_el >= 30
else 'fair'
),
# Backwards-compatible aliases used by weather_sat_scheduler and the frontend # Backwards-compatible aliases used by weather_sat_scheduler and the frontend
'aosAz': round(rise_az, 1), "aosAz": round(rise_az, 1),
'losAz': round(set_az, 1), "losAz": round(set_az, 1),
'tcaAz': round(max_el_az, 1), "tcaAz": round(max_el_az, 1),
} }
if include_trajectory: if include_trajectory:
pass_dict['trajectory'] = traj_points pass_dict["trajectory"] = traj_points
if include_ground_track: if include_ground_track:
ground_track = [] ground_track = []
@@ -217,12 +219,14 @@ def _process_pass(
try: try:
geocentric = satellite.at(t_pt) geocentric = satellite.at(t_pt)
subpoint = wgs84.subpoint(geocentric) subpoint = wgs84.subpoint(geocentric)
ground_track.append({ ground_track.append(
'lat': round(float(subpoint.latitude.degrees), 4), {
'lon': round(float(subpoint.longitude.degrees), 4), "lat": round(float(subpoint.latitude.degrees), 4),
}) "lon": round(float(subpoint.longitude.degrees), 4),
}
)
except Exception: except Exception:
pass pass
pass_dict['groundTrack'] = ground_track pass_dict["groundTrack"] = ground_track
all_passes.append(pass_dict) all_passes.append(pass_dict)