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

Align ISS SSTV start flow with HF decoder contract

Browse Source
This commit is contained in:
Smittix
2026-02-19 12:29:27 +00:00
parent bec0881018
commit 2faed68af4
3 changed files with 168 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
routes/sstv.py
+33 -3
View File
@@ -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
}
static/js/modes/sstv.js
+6 -3
View File
@@ -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');
tests/test_sstv_routes.py
+129
View File
@@ -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()