From 2faed68af45fa3e5cc2c303696fa9de81445b227 Mon Sep 17 00:00:00 2001 From: Smittix Date: Thu, 19 Feb 2026 12:29:27 +0000 Subject: [PATCH] Align ISS SSTV start flow with HF decoder contract --- routes/sstv.py | 36 ++++++++++- static/js/modes/sstv.js | 9 ++- tests/test_sstv_routes.py | 129 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 168 insertions(+), 6 deletions(-) create mode 100644 tests/test_sstv_routes.py diff --git a/routes/sstv.py b/routes/sstv.py index 1029dec..fbfe0d0 100644 --- a/routes/sstv.py +++ b/routes/sstv.py @@ -27,6 +27,12 @@ logger = get_logger('intercept.sstv') sstv_bp = Blueprint('sstv', __name__, url_prefix='/sstv') +# ISS SSTV runs on a fixed downlink; allow a small entry tolerance so users +# can type nearby values and still land on the canonical center frequency. +ISS_SSTV_MODULATION = 'fm' +ISS_SSTV_FREQUENCIES = (ISS_SSTV_FREQ,) +ISS_SSTV_FREQ_TOLERANCE_MHZ = 0.05 + # Queue for SSE progress streaming _sstv_queue: queue.Queue = queue.Queue(maxsize=100) @@ -46,6 +52,14 @@ def _progress_callback(data: dict) -> None: pass +def _normalize_iss_frequency(frequency_mhz: float) -> float | None: + """Snap near-match user input to a supported ISS SSTV center frequency.""" + for supported in ISS_SSTV_FREQUENCIES: + if abs(frequency_mhz - supported) <= ISS_SSTV_FREQ_TOLERANCE_MHZ: + return supported + return None + + @sstv_bp.route('/status') def get_status(): """ @@ -62,6 +76,7 @@ def get_status(): 'decoder': decoder.decoder_available, 'running': decoder.is_running, 'iss_frequency': ISS_SSTV_FREQ, + 'modulation': ISS_SSTV_MODULATION, 'image_count': len(decoder.get_images()), 'doppler_enabled': decoder.doppler_enabled, } @@ -82,6 +97,7 @@ def start_decoder(): JSON body (optional): { "frequency": 145.800, // Frequency in MHz (default: ISS 145.800) + "modulation": "fm", // ISS mode is FM-only "device": 0, // RTL-SDR device index "latitude": 40.7128, // Observer latitude for Doppler correction "longitude": -74.0060 // Observer longitude for Doppler correction @@ -106,6 +122,7 @@ def start_decoder(): return jsonify({ 'status': 'already_running', 'frequency': ISS_SSTV_FREQ, + 'modulation': ISS_SSTV_MODULATION, 'doppler_enabled': decoder.doppler_enabled }) @@ -119,18 +136,29 @@ def start_decoder(): # Get parameters data = request.get_json(silent=True) or {} frequency = data.get('frequency', ISS_SSTV_FREQ) + modulation = str(data.get('modulation', ISS_SSTV_MODULATION)).strip().lower() device_index = data.get('device', 0) latitude = data.get('latitude') longitude = data.get('longitude') + # Validate modulation (ISS mode is FM-only) + if modulation != ISS_SSTV_MODULATION: + return jsonify({ + 'status': 'error', + 'message': f'Modulation must be {ISS_SSTV_MODULATION} for ISS SSTV mode' + }), 400 + # Validate frequency try: frequency = float(frequency) - if not (100 <= frequency <= 500): # VHF range + normalized_frequency = _normalize_iss_frequency(frequency) + if normalized_frequency is None: + supported = ', '.join(f'{freq:.3f}' for freq in ISS_SSTV_FREQUENCIES) return jsonify({ 'status': 'error', - 'message': 'Frequency must be between 100-500 MHz' + 'message': f'Supported ISS SSTV frequency: {supported} MHz FM' }), 400 + frequency = normalized_frequency except (TypeError, ValueError): return jsonify({ 'status': 'error', @@ -178,7 +206,8 @@ def start_decoder(): frequency=frequency, device_index=device_index, latitude=latitude, - longitude=longitude + longitude=longitude, + modulation=ISS_SSTV_MODULATION, ) if success: @@ -187,6 +216,7 @@ def start_decoder(): result = { 'status': 'started', 'frequency': frequency, + 'modulation': ISS_SSTV_MODULATION, 'device': device_index, 'doppler_enabled': decoder.doppler_enabled } diff --git a/static/js/modes/sstv.js b/static/js/modes/sstv.js index ed6d13e..f8dbbcf 100644 --- a/static/js/modes/sstv.js +++ b/static/js/modes/sstv.js @@ -20,6 +20,7 @@ const SSTV = (function() { // ISS frequency const ISS_FREQ = 145.800; + const ISS_MODULATION = 'fm'; // Signal scope state let sstvScopeCtx = null; @@ -544,7 +545,7 @@ const SSTV = (function() { const response = await fetch('/sstv/start', { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ frequency, device }) + body: JSON.stringify({ frequency, modulation: ISS_MODULATION, device }) }); const data = await response.json(); @@ -554,9 +555,11 @@ const SSTV = (function() { if (typeof reserveDevice === 'function') { reserveDevice(device, 'sstv'); } - updateStatusUI('listening', `${frequency} MHz`); + const tunedFrequency = Number(data.frequency || frequency); + const modulationText = String(data.modulation || ISS_MODULATION).toUpperCase(); + updateStatusUI('listening', `${tunedFrequency.toFixed(3)} MHz ${modulationText}`); startStream(); - showNotification('SSTV', `Listening on ${frequency} MHz`); + showNotification('SSTV', `Listening on ${tunedFrequency.toFixed(3)} MHz ${modulationText}`); } else { updateStatusUI('idle', 'Start failed'); showStatusMessage(data.message || 'Failed to start decoder', 'error'); diff --git a/tests/test_sstv_routes.py b/tests/test_sstv_routes.py new file mode 100644 index 0000000..8eb9049 --- /dev/null +++ b/tests/test_sstv_routes.py @@ -0,0 +1,129 @@ +"""Tests for ISS SSTV route behavior.""" + +from __future__ import annotations + +import json +from unittest.mock import MagicMock, patch + +import pytest + +from utils.sstv import ISS_SSTV_FREQ + + +def _login_session(client) -> None: + """Mark the Flask test session as authenticated.""" + with client.session_transaction() as sess: + sess['logged_in'] = True + sess['username'] = 'test' + sess['role'] = 'admin' + + +class TestSSTVRoutes: + """ISS SSTV route tests.""" + + def test_status_reports_fm_modulation(self, client): + """GET /sstv/status should report the fixed ISS modulation.""" + _login_session(client) + mock_decoder = MagicMock() + mock_decoder.decoder_available = 'python-sstv' + mock_decoder.is_running = False + mock_decoder.get_images.return_value = [] + mock_decoder.doppler_enabled = False + mock_decoder.last_doppler_info = None + + with patch('routes.sstv.is_sstv_available', return_value=True), \ + patch('routes.sstv.get_sstv_decoder', return_value=mock_decoder): + response = client.get('/sstv/status') + + assert response.status_code == 200 + data = response.get_json() + assert data['available'] is True + assert data['modulation'] == 'fm' + assert data['iss_frequency'] == ISS_SSTV_FREQ + + def test_start_uses_fm_and_normalizes_supported_iss_frequency(self, client): + """POST /sstv/start should enforce FM and snap near ISS values.""" + _login_session(client) + mock_decoder = MagicMock() + mock_decoder.is_running = False + mock_decoder.start.return_value = True + mock_decoder.doppler_enabled = False + mock_decoder.last_doppler_info = None + + payload = { + 'frequency': ISS_SSTV_FREQ + 0.02, # Within tolerance; should normalize. + 'modulation': 'FM', + 'device': 0, + } + + with patch('routes.sstv.is_sstv_available', return_value=True), \ + patch('routes.sstv.get_sstv_decoder', return_value=mock_decoder), \ + patch('routes.sstv.app_module.claim_sdr_device', return_value=None): + response = client.post( + '/sstv/start', + data=json.dumps(payload), + content_type='application/json', + ) + + assert response.status_code == 200 + data = response.get_json() + assert data['status'] == 'started' + assert data['modulation'] == 'fm' + assert data['frequency'] == pytest.approx(ISS_SSTV_FREQ) + + mock_decoder.start.assert_called_once() + call_kwargs = mock_decoder.start.call_args.kwargs + assert call_kwargs['modulation'] == 'fm' + assert call_kwargs['frequency'] == pytest.approx(ISS_SSTV_FREQ) + + def test_start_rejects_non_fm_modulation(self, client): + """POST /sstv/start should reject non-FM modulation requests.""" + _login_session(client) + mock_decoder = MagicMock() + mock_decoder.is_running = False + + payload = { + 'frequency': ISS_SSTV_FREQ, + 'modulation': 'usb', + 'device': 0, + } + + with patch('routes.sstv.is_sstv_available', return_value=True), \ + patch('routes.sstv.get_sstv_decoder', return_value=mock_decoder): + response = client.post( + '/sstv/start', + data=json.dumps(payload), + content_type='application/json', + ) + + assert response.status_code == 400 + data = response.get_json() + assert data['status'] == 'error' + assert 'Modulation must be fm' in data['message'] + mock_decoder.start.assert_not_called() + + def test_start_rejects_non_iss_frequency(self, client): + """POST /sstv/start should reject unsupported non-ISS frequencies.""" + _login_session(client) + mock_decoder = MagicMock() + mock_decoder.is_running = False + + payload = { + 'frequency': 14.230, + 'modulation': 'fm', + 'device': 0, + } + + with patch('routes.sstv.is_sstv_available', return_value=True), \ + patch('routes.sstv.get_sstv_decoder', return_value=mock_decoder): + response = client.post( + '/sstv/start', + data=json.dumps(payload), + content_type='application/json', + ) + + assert response.status_code == 400 + data = response.get_json() + assert data['status'] == 'error' + assert 'Supported ISS SSTV frequency' in data['message'] + mock_decoder.start.assert_not_called()