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

Replace broken slowrx dependency with pure Python SSTV decoder

Browse Source 
slowrx is a GTK GUI app that doesn't support CLI usage, so the SSTV
decoder was silently failing. This replaces it with a pure Python
implementation using numpy and Pillow that supports Robot36/72,
Martin1/2, Scottie1/2, and PD120/180 modes via VIS header auto-detection.

Key implementation details:
- Generalized Goertzel (DTFT) for exact-frequency tone detection
- Vectorized batch Goertzel for real-time pixel decoding performance
- Overlapping analysis windows for short-window frequency estimation
- VIS header detection state machine with parity validation
- Per-line sync re-synchronization for drift tolerance

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Smittix
2026-02-06 19:47:02 +00:00
parent ae9fe5d063
commit ef7d8cca9f
16 changed files with 2978 additions and 877 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/sstv/__init__.py
+33
View File
@@ -0,0 +1,33 @@
"""SSTV (Slow-Scan Television) decoder package.
Pure Python SSTV decoder using Goertzel-based DSP for VIS header detection
and scanline-by-scanline image decoding. Supports Robot36/72, Martin1/2,
Scottie1/2, and PD120/180 modes.
Replaces the external slowrx dependency with numpy/scipy + Pillow.
"""
from .constants import ISS_SSTV_FREQ, SSTV_MODES
from .sstv_decoder import (
DecodeProgress,
DopplerInfo,
DopplerTracker,
SSTVDecoder,
SSTVImage,
get_general_sstv_decoder,
get_sstv_decoder,
is_sstv_available,
)
__all__ = [
'DecodeProgress',
'DopplerInfo',
'DopplerTracker',
'ISS_SSTV_FREQ',
'SSTV_MODES',
'SSTVDecoder',
'SSTVImage',
'get_general_sstv_decoder',
'get_sstv_decoder',
'is_sstv_available',
]
utils/sstv/constants.py
+92
View File
@@ -0,0 +1,92 @@
"""SSTV protocol constants.
VIS (Vertical Interval Signaling) codes, frequency assignments, and timing
constants for all supported SSTV modes per the SSTV protocol specification.
"""
from __future__ import annotations
# ---------------------------------------------------------------------------
# Audio / DSP
# ---------------------------------------------------------------------------
SAMPLE_RATE = 48000 # Hz - standard audio sample rate used by rtl_fm
# Window size for Goertzel tone detection (5 ms at 48 kHz = 240 samples)
GOERTZEL_WINDOW = 240
# Chunk size for reading from rtl_fm (100 ms = 4800 samples)
STREAM_CHUNK_SAMPLES = 4800
# ---------------------------------------------------------------------------
# SSTV tone frequencies (Hz)
# ---------------------------------------------------------------------------
FREQ_VIS_BIT_1 = 1100 # VIS logic 1
FREQ_SYNC = 1200 # Horizontal sync pulse
FREQ_VIS_BIT_0 = 1300 # VIS logic 0
FREQ_BREAK = 1200 # Break tone in VIS header (same as sync)
FREQ_LEADER = 1900 # Leader / calibration tone
FREQ_BLACK = 1500 # Black level
FREQ_WHITE = 2300 # White level
# Pixel luminance mapping range
FREQ_PIXEL_LOW = 1500 # 0 luminance
FREQ_PIXEL_HIGH = 2300 # 255 luminance
# Frequency tolerance for tone detection (Hz)
FREQ_TOLERANCE = 50
# ---------------------------------------------------------------------------
# VIS header timing (seconds)
# ---------------------------------------------------------------------------
VIS_LEADER_MIN = 0.200 # Minimum leader tone duration
VIS_LEADER_MAX = 0.500 # Maximum leader tone duration
VIS_LEADER_NOMINAL = 0.300 # Nominal leader tone duration
VIS_BREAK_DURATION = 0.010 # Break pulse duration (10 ms)
VIS_BIT_DURATION = 0.030 # Each VIS data bit (30 ms)
VIS_START_BIT_DURATION = 0.030 # Start bit (30 ms)
VIS_STOP_BIT_DURATION = 0.030 # Stop bit (30 ms)
# Timing tolerance for VIS detection
VIS_TIMING_TOLERANCE = 0.5 # 50% tolerance on durations
# ---------------------------------------------------------------------------
# VIS code → mode name mapping
# ---------------------------------------------------------------------------
VIS_CODES: dict[int, str] = {
8: 'Robot36',
12: 'Robot72',
44: 'Martin1',
40: 'Martin2',
60: 'Scottie1',
56: 'Scottie2',
93: 'PD120',
95: 'PD180',
# Less common but recognized
4: 'Robot24',
36: 'Martin3',
52: 'Scottie3',
55: 'ScottieDX',
113: 'PD240',
96: 'PD90',
98: 'PD160',
}
# Reverse mapping: mode name → VIS code
MODE_TO_VIS: dict[str, int] = {v: k for k, v in VIS_CODES.items()}
# ---------------------------------------------------------------------------
# Common SSTV modes list (for UI / status)
# ---------------------------------------------------------------------------
SSTV_MODES = [
'PD120', 'PD180', 'Martin1', 'Martin2',
'Scottie1', 'Scottie2', 'Robot36', 'Robot72',
]
# ISS SSTV frequency
ISS_SSTV_FREQ = 145.800 # MHz
# Speed of light in m/s
SPEED_OF_LIGHT = 299_792_458
# Minimum energy ratio for valid tone detection (vs noise floor)
MIN_ENERGY_RATIO = 5.0
utils/sstv/dsp.py
+232
View File
@@ -0,0 +1,232 @@
"""DSP utilities for SSTV decoding.
Goertzel algorithm for efficient single-frequency energy detection,
frequency estimation, and frequency-to-pixel luminance mapping.
"""
from __future__ import annotations
import math
import numpy as np
from .constants import (
FREQ_PIXEL_HIGH,
FREQ_PIXEL_LOW,
MIN_ENERGY_RATIO,
SAMPLE_RATE,
)
def goertzel(samples: np.ndarray, target_freq: float,
sample_rate: int = SAMPLE_RATE) -> float:
"""Compute Goertzel energy at a single target frequency.
O(N) per frequency - more efficient than FFT when only a few
frequencies are needed.
Args:
samples: Audio samples (float64, -1.0 to 1.0).
target_freq: Frequency to detect (Hz).
sample_rate: Sample rate (Hz).
Returns:
Magnitude squared (energy) at the target frequency.
"""
n = len(samples)
if n == 0:
return 0.0
# Generalized Goertzel (DTFT): use exact target frequency rather than
# rounding to the nearest DFT bin. This is critical for short windows
# (e.g. 13 samples/pixel) where integer-k Goertzel quantizes all SSTV
# pixel frequencies into 1-2 bins, making estimation impossible.
w = 2.0 * math.pi * target_freq / sample_rate
coeff = 2.0 * math.cos(w)
s0 = 0.0
s1 = 0.0
s2 = 0.0
for sample in samples:
s0 = sample + coeff * s1 - s2
s2 = s1
s1 = s0
return s1 * s1 + s2 * s2 - coeff * s1 * s2
def goertzel_mag(samples: np.ndarray, target_freq: float,
sample_rate: int = SAMPLE_RATE) -> float:
"""Compute Goertzel magnitude (square root of energy).
Args:
samples: Audio samples.
target_freq: Frequency to detect (Hz).
sample_rate: Sample rate (Hz).
Returns:
Magnitude at the target frequency.
"""
return math.sqrt(max(0.0, goertzel(samples, target_freq, sample_rate)))
def detect_tone(samples: np.ndarray, candidates: list[float],
sample_rate: int = SAMPLE_RATE) -> tuple[float | None, float]:
"""Detect which candidate frequency has the strongest energy.
Args:
samples: Audio samples.
candidates: List of candidate frequencies (Hz).
sample_rate: Sample rate (Hz).
Returns:
Tuple of (detected_frequency or None, energy_ratio).
Returns None if no tone significantly dominates.
"""
if len(samples) == 0 or not candidates:
return None, 0.0
energies = {f: goertzel(samples, f, sample_rate) for f in candidates}
max_freq = max(energies, key=energies.get) # type: ignore[arg-type]
max_energy = energies[max_freq]
if max_energy <= 0:
return None, 0.0
# Calculate ratio of strongest to average of others
others = [e for f, e in energies.items() if f != max_freq]
avg_others = sum(others) / len(others) if others else 0.0
ratio = max_energy / avg_others if avg_others > 0 else float('inf')
if ratio >= MIN_ENERGY_RATIO:
return max_freq, ratio
return None, ratio
def estimate_frequency(samples: np.ndarray, freq_low: float = 1000.0,
freq_high: float = 2500.0, step: float = 25.0,
sample_rate: int = SAMPLE_RATE) -> float:
"""Estimate the dominant frequency in a range using Goertzel sweep.
Sweeps through frequencies in the given range and returns the one
with maximum energy. Uses a coarse sweep followed by a fine sweep
for accuracy.
Args:
samples: Audio samples.
freq_low: Lower bound of frequency range (Hz).
freq_high: Upper bound of frequency range (Hz).
step: Coarse step size (Hz).
sample_rate: Sample rate (Hz).
Returns:
Estimated dominant frequency (Hz).
"""
if len(samples) == 0:
return 0.0
# Coarse sweep
best_freq = freq_low
best_energy = 0.0
freq = freq_low
while freq <= freq_high:
energy = goertzel(samples, freq, sample_rate)
if energy > best_energy:
best_energy = energy
best_freq = freq
freq += step
# Fine sweep around the coarse peak (+/- one step, 5 Hz resolution)
fine_low = max(freq_low, best_freq - step)
fine_high = min(freq_high, best_freq + step)
freq = fine_low
while freq <= fine_high:
energy = goertzel(samples, freq, sample_rate)
if energy > best_energy:
best_energy = energy
best_freq = freq
freq += 5.0
return best_freq
def freq_to_pixel(frequency: float) -> int:
"""Convert SSTV audio frequency to pixel luminance value (0-255).
Linear mapping: 1500 Hz = 0 (black), 2300 Hz = 255 (white).
Args:
frequency: Detected frequency (Hz).
Returns:
Pixel value clamped to 0-255.
"""
normalized = (frequency - FREQ_PIXEL_LOW) / (FREQ_PIXEL_HIGH - FREQ_PIXEL_LOW)
return max(0, min(255, int(normalized * 255 + 0.5)))
def samples_for_duration(duration_s: float,
sample_rate: int = SAMPLE_RATE) -> int:
"""Calculate number of samples for a given duration.
Args:
duration_s: Duration in seconds.
sample_rate: Sample rate (Hz).
Returns:
Number of samples.
"""
return int(duration_s * sample_rate + 0.5)
def goertzel_batch(audio_matrix: np.ndarray, frequencies: np.ndarray,
sample_rate: int = SAMPLE_RATE) -> np.ndarray:
"""Compute Goertzel energy for multiple audio segments at multiple frequencies.
Vectorized implementation using numpy broadcasting. Processes all
pixel windows and all candidate frequencies simultaneously, giving
roughly 50-100x speed-up over the scalar ``goertzel`` called in a
Python loop.
Args:
audio_matrix: Shape (M, N) – M audio segments of N samples each.
frequencies: 1-D array of F target frequencies in Hz.
sample_rate: Sample rate in Hz.
Returns:
Shape (M, F) array of energy values.
"""
if audio_matrix.size == 0 or len(frequencies) == 0:
return np.zeros((audio_matrix.shape[0], len(frequencies)))
_M, N = audio_matrix.shape
# Generalized Goertzel (DTFT): exact target frequencies, no bin rounding
w = 2.0 * np.pi * frequencies / sample_rate
coeff = 2.0 * np.cos(w) # (F,)
s1 = np.zeros((audio_matrix.shape[0], len(frequencies)))
s2 = np.zeros_like(s1)
for n in range(N):
samples_n = audio_matrix[:, n:n + 1] # (M, 1) — broadcasts with (M, F)
s0 = samples_n + coeff * s1 - s2
s2 = s1
s1 = s0
return s1 * s1 + s2 * s2 - coeff * s1 * s2
def normalize_audio(raw: np.ndarray) -> np.ndarray:
"""Normalize int16 PCM audio to float64 in range [-1.0, 1.0].
Args:
raw: Raw int16 samples from rtl_fm.
Returns:
Float64 normalized samples.
"""
return raw.astype(np.float64) / 32768.0
utils/sstv/image_decoder.py
+453
View File
@@ -0,0 +1,453 @@
"""SSTV scanline-by-scanline image decoder.
Decodes raw audio samples into a PIL Image for all supported SSTV modes.
Handles sync pulse re-synchronization on each line for robust decoding
under weak-signal or drifting conditions.
"""
from __future__ import annotations
from typing import Callable
import numpy as np
from .constants import (
FREQ_BLACK,
FREQ_PIXEL_HIGH,
FREQ_PIXEL_LOW,
FREQ_SYNC,
SAMPLE_RATE,
)
from .dsp import (
goertzel,
goertzel_batch,
samples_for_duration,
)
from .modes import (
ColorModel,
SSTVMode,
SyncPosition,
)
# Pillow is imported lazily to keep the module importable when Pillow
# is not installed (is_sstv_available() just returns True, but actual
# decoding would fail gracefully).
try:
from PIL import Image
except ImportError:
Image = None # type: ignore[assignment,misc]
# Type alias for progress callback: (current_line, total_lines)
ProgressCallback = Callable[[int, int], None]
class SSTVImageDecoder:
"""Decode an SSTV image from a stream of audio samples.
Usage::
decoder = SSTVImageDecoder(mode)
decoder.feed(samples)
...
if decoder.is_complete:
image = decoder.get_image()
"""
def __init__(self, mode: SSTVMode, sample_rate: int = SAMPLE_RATE,
progress_cb: ProgressCallback | None = None):
self._mode = mode
self._sample_rate = sample_rate
self._progress_cb = progress_cb
self._buffer = np.array([], dtype=np.float64)
self._current_line = 0
self._complete = False
# Pre-calculate sample counts
self._sync_samples = samples_for_duration(
mode.sync_duration_ms / 1000.0, sample_rate)
self._porch_samples = samples_for_duration(
mode.sync_porch_ms / 1000.0, sample_rate)
self._line_samples = samples_for_duration(
mode.line_duration_ms / 1000.0, sample_rate)
self._separator_samples = (
samples_for_duration(mode.channel_separator_ms / 1000.0, sample_rate)
if mode.channel_separator_ms > 0 else 0
)
self._channel_samples = [
samples_for_duration(ch.duration_ms / 1000.0, sample_rate)
for ch in mode.channels
]
# For PD modes, each "line" of audio produces 2 image lines
if mode.color_model == ColorModel.YCRCB_DUAL:
self._total_audio_lines = mode.height // 2
else:
self._total_audio_lines = mode.height
# Initialize pixel data arrays per channel
self._channel_data: list[np.ndarray] = []
for _i, _ch_spec in enumerate(mode.channels):
if mode.color_model == ColorModel.YCRCB_DUAL:
# Y1, Cr, Cb, Y2 - all are width-wide
self._channel_data.append(
np.zeros((self._total_audio_lines, mode.width), dtype=np.uint8))
else:
self._channel_data.append(
np.zeros((mode.height, mode.width), dtype=np.uint8))
# Pre-compute candidate frequencies for batch pixel decoding (5 Hz step)
self._freq_candidates = np.arange(
FREQ_PIXEL_LOW - 100, FREQ_PIXEL_HIGH + 105, 5.0)
# Track sync position for re-synchronization
self._expected_line_start = 0 # Sample offset within buffer
self._synced = False
@property
def is_complete(self) -> bool:
return self._complete
@property
def current_line(self) -> int:
return self._current_line
@property
def total_lines(self) -> int:
return self._total_audio_lines
@property
def progress_percent(self) -> int:
if self._total_audio_lines == 0:
return 0
return min(100, int(100 * self._current_line / self._total_audio_lines))
def feed(self, samples: np.ndarray) -> bool:
"""Feed audio samples into the decoder.
Args:
samples: Float64 audio samples.
Returns:
True when image is complete.
"""
if self._complete:
return True
self._buffer = np.concatenate([self._buffer, samples])
# Process complete lines
while not self._complete and len(self._buffer) >= self._line_samples:
self._decode_line()
# Prevent unbounded buffer growth - keep at most 2 lines worth
max_buffer = self._line_samples * 2
if len(self._buffer) > max_buffer and not self._complete:
self._buffer = self._buffer[-max_buffer:]
return self._complete
def _find_sync(self, search_region: np.ndarray) -> int | None:
"""Find the 1200 Hz sync pulse within a search region.
Scans through the region looking for a stretch of 1200 Hz
tone of approximately the right duration.
Args:
search_region: Audio samples to search within.
Returns:
Sample offset of the sync pulse start, or None if not found.
"""
window_size = min(self._sync_samples, 200)
if len(search_region) < window_size:
return None
best_pos = None
best_energy = 0.0
step = window_size // 2
for pos in range(0, len(search_region) - window_size, step):
chunk = search_region[pos:pos + window_size]
sync_energy = goertzel(chunk, FREQ_SYNC, self._sample_rate)
# Check it's actually sync, not data at 1200 Hz area
black_energy = goertzel(chunk, FREQ_BLACK, self._sample_rate)
if sync_energy > best_energy and sync_energy > black_energy * 2:
best_energy = sync_energy
best_pos = pos
return best_pos
def _decode_line(self) -> None:
"""Decode one scanline from the buffer."""
if self._current_line >= self._total_audio_lines:
self._complete = True
return
# Try to find sync pulse for re-synchronization
# Search within +/-10% of expected line start
search_margin = max(100, self._line_samples // 10)
line_start = 0
if self._mode.sync_position in (SyncPosition.FRONT, SyncPosition.FRONT_PD):
# Sync is at the beginning of each line
search_start = 0
search_end = min(len(self._buffer), self._sync_samples + search_margin)
search_region = self._buffer[search_start:search_end]
sync_pos = self._find_sync(search_region)
if sync_pos is not None:
line_start = sync_pos
# Skip sync + porch to get to pixel data
pixel_start = line_start + self._sync_samples + self._porch_samples
elif self._mode.sync_position == SyncPosition.MIDDLE:
# Scottie: sep(1.5ms) -> G -> sep(1.5ms) -> B -> sync(9ms) -> porch(1.5ms) -> R
# Skip initial separator (same duration as porch)
pixel_start = self._porch_samples
line_start = 0
else:
pixel_start = self._sync_samples + self._porch_samples
# Decode each channel
pos = pixel_start
for ch_idx, ch_samples in enumerate(self._channel_samples):
if pos + ch_samples > len(self._buffer):
# Not enough data yet - put the data back and wait
return
channel_audio = self._buffer[pos:pos + ch_samples]
pixels = self._decode_channel_pixels(channel_audio)
self._channel_data[ch_idx][self._current_line, :] = pixels
pos += ch_samples
# Add inter-channel gaps based on mode family
if ch_idx < len(self._channel_samples) - 1:
if self._mode.sync_position == SyncPosition.MIDDLE:
if ch_idx == 0:
# Scottie: separator between G and B
pos += self._porch_samples
else:
# Scottie: sync + porch between B and R
pos += self._sync_samples + self._porch_samples
elif self._separator_samples > 0:
# Robot: separator + porch between channels
pos += self._separator_samples
elif (self._mode.sync_position == SyncPosition.FRONT
and self._mode.color_model == ColorModel.RGB):
# Martin: porch between channels
pos += self._porch_samples
# Advance buffer past this line
consumed = max(pos, self._line_samples)
self._buffer = self._buffer[consumed:]
self._current_line += 1
if self._progress_cb:
self._progress_cb(self._current_line, self._total_audio_lines)
if self._current_line >= self._total_audio_lines:
self._complete = True
# Minimum analysis window for meaningful Goertzel frequency estimation.
# With 96 samples (2ms at 48kHz), frequency accuracy is within ~25 Hz,
# giving pixel-level accuracy of ~8/255 levels.
_MIN_ANALYSIS_WINDOW = 96
def _decode_channel_pixels(self, audio: np.ndarray) -> np.ndarray:
"""Decode pixel values from a channel's audio data.
Uses batch Goertzel to estimate frequencies for all pixels
simultaneously, then maps to luminance values. When pixels have
fewer samples than ``_MIN_ANALYSIS_WINDOW``, overlapping analysis
windows are used to maintain frequency estimation accuracy.
Args:
audio: Audio samples for one channel of one scanline.
Returns:
Array of pixel values (0-255), shape (width,).
"""
width = self._mode.width
samples_per_pixel = max(1, len(audio) // width)
if len(audio) < width or samples_per_pixel < 2:
return np.zeros(width, dtype=np.uint8)
window_size = max(samples_per_pixel, self._MIN_ANALYSIS_WINDOW)
if window_size > samples_per_pixel and len(audio) >= window_size:
# Use overlapping windows centered on each pixel position
windows = np.lib.stride_tricks.sliding_window_view(
audio, window_size)
# Pixel centers, clamped to valid window indices
centers = np.arange(width) * samples_per_pixel
indices = np.minimum(centers, len(windows) - 1)
audio_matrix = np.ascontiguousarray(windows[indices])
else:
# Non-overlapping: each pixel has enough samples
usable = width * samples_per_pixel
audio_matrix = audio[:usable].reshape(width, samples_per_pixel)
# Batch Goertzel at all candidate frequencies
energies = goertzel_batch(
audio_matrix, self._freq_candidates, self._sample_rate)
# Find peak frequency per pixel
best_idx = np.argmax(energies, axis=1)
best_freqs = self._freq_candidates[best_idx]
# Map frequencies to pixel values (1500 Hz = 0, 2300 Hz = 255)
normalized = (best_freqs - FREQ_PIXEL_LOW) / (FREQ_PIXEL_HIGH - FREQ_PIXEL_LOW)
return np.clip(normalized * 255 + 0.5, 0, 255).astype(np.uint8)
def get_image(self) -> Image.Image | None:
"""Convert decoded channel data to a PIL Image.
Returns:
PIL Image in RGB mode, or None if Pillow is not available
or decoding is incomplete.
"""
if Image is None:
return None
mode = self._mode
if mode.color_model == ColorModel.RGB:
return self._assemble_rgb()
elif mode.color_model == ColorModel.YCRCB:
return self._assemble_ycrcb()
elif mode.color_model == ColorModel.YCRCB_DUAL:
return self._assemble_ycrcb_dual()
return None
def _assemble_rgb(self) -> Image.Image:
"""Assemble RGB image from sequential R, G, B channel data.
Martin/Scottie channel order: G, B, R.
"""
height = self._mode.height
# Channel order for Martin/Scottie: [0]=G, [1]=B, [2]=R
g_data = self._channel_data[0][:height]
b_data = self._channel_data[1][:height]
r_data = self._channel_data[2][:height]
rgb = np.stack([r_data, g_data, b_data], axis=-1)
return Image.fromarray(rgb, 'RGB')
def _assemble_ycrcb(self) -> Image.Image:
"""Assemble image from YCrCb data (Robot modes).
Robot36: Y every line, Cr/Cb alternating (half-rate chroma).
Robot72: Y, Cr, Cb every line (full-rate chroma).
"""
height = self._mode.height
width = self._mode.width
if not self._mode.has_half_rate_chroma:
# Full-rate chroma (Robot72): Y, Cr, Cb as separate channels
y_data = self._channel_data[0][:height].astype(np.float64)
cr = self._channel_data[1][:height].astype(np.float64)
cb = self._channel_data[2][:height].astype(np.float64)
return self._ycrcb_to_rgb(y_data, cr, cb, height, width)
# Half-rate chroma (Robot36): Y + alternating Cr/Cb
y_data = self._channel_data[0][:height].astype(np.float64)
chroma_data = self._channel_data[1][:height].astype(np.float64)
# Separate Cr (even lines) and Cb (odd lines), then interpolate
cr = np.zeros((height, width), dtype=np.float64)
cb = np.zeros((height, width), dtype=np.float64)
for line in range(height):
if line % 2 == 0:
cr[line] = chroma_data[line]
else:
cb[line] = chroma_data[line]
# Interpolate missing chroma lines
for line in range(height):
if line % 2 == 1:
# Missing Cr - interpolate from neighbors
prev_cr = line - 1 if line > 0 else line + 1
next_cr = line + 1 if line + 1 < height else line - 1
cr[line] = (cr[prev_cr] + cr[next_cr]) / 2
else:
# Missing Cb - interpolate from neighbors
prev_cb = line - 1 if line > 0 else line + 1
next_cb = line + 1 if line + 1 < height else line - 1
if prev_cb >= 0 and next_cb < height:
cb[line] = (cb[prev_cb] + cb[next_cb]) / 2
elif prev_cb >= 0:
cb[line] = cb[prev_cb]
else:
cb[line] = cb[next_cb]
return self._ycrcb_to_rgb(y_data, cr, cb, height, width)
def _assemble_ycrcb_dual(self) -> Image.Image:
"""Assemble image from dual-luminance YCrCb data (PD modes).
PD modes send Y1, Cr, Cb, Y2 per audio line, producing 2 image lines.
"""
audio_lines = self._total_audio_lines
width = self._mode.width
height = self._mode.height
y1_data = self._channel_data[0][:audio_lines].astype(np.float64)
cr_data = self._channel_data[1][:audio_lines].astype(np.float64)
cb_data = self._channel_data[2][:audio_lines].astype(np.float64)
y2_data = self._channel_data[3][:audio_lines].astype(np.float64)
# Interleave Y1 and Y2 to produce full-height luminance
y_full = np.zeros((height, width), dtype=np.float64)
cr_full = np.zeros((height, width), dtype=np.float64)
cb_full = np.zeros((height, width), dtype=np.float64)
for i in range(audio_lines):
even_line = i * 2
odd_line = i * 2 + 1
if even_line < height:
y_full[even_line] = y1_data[i]
cr_full[even_line] = cr_data[i]
cb_full[even_line] = cb_data[i]
if odd_line < height:
y_full[odd_line] = y2_data[i]
cr_full[odd_line] = cr_data[i]
cb_full[odd_line] = cb_data[i]
return self._ycrcb_to_rgb(y_full, cr_full, cb_full, height, width)
@staticmethod
def _ycrcb_to_rgb(y: np.ndarray, cr: np.ndarray, cb: np.ndarray,
height: int, width: int) -> Image.Image:
"""Convert YCrCb pixel data to an RGB PIL Image.
Uses the SSTV convention where pixel values 0-255 map to the
standard Y'CbCr color space used by JPEG/SSTV.
"""
# Normalize from 0-255 pixel range to standard ranges
# Y: 0-255, Cr/Cb: 0-255 centered at 128
y_norm = y
cr_norm = cr - 128.0
cb_norm = cb - 128.0
# ITU-R BT.601 conversion
r = y_norm + 1.402 * cr_norm
g = y_norm - 0.344136 * cb_norm - 0.714136 * cr_norm
b = y_norm + 1.772 * cb_norm
# Clip and convert
r = np.clip(r, 0, 255).astype(np.uint8)
g = np.clip(g, 0, 255).astype(np.uint8)
b = np.clip(b, 0, 255).astype(np.uint8)
rgb = np.stack([r, g, b], axis=-1)
return Image.fromarray(rgb, 'RGB')
utils/sstv/modes.py
+250
View File
@@ -0,0 +1,250 @@
"""SSTV mode specifications.
Dataclass definitions for each supported SSTV mode, encoding resolution,
color model, line timing, and sync characteristics.
"""
from __future__ import annotations
import enum
from dataclasses import dataclass, field
class ColorModel(enum.Enum):
"""Color encoding models used by SSTV modes."""
RGB = 'rgb' # Sequential R, G, B channels per line
YCRCB = 'ycrcb' # Luminance + chrominance (Robot modes)
YCRCB_DUAL = 'ycrcb_dual' # Dual-luminance YCrCb (PD modes)
class SyncPosition(enum.Enum):
"""Where the horizontal sync pulse appears in each line."""
FRONT = 'front' # Sync at start of line (Robot, Martin)
MIDDLE = 'middle' # Sync between G and B channels (Scottie)
FRONT_PD = 'front_pd' # PD-style sync at start
@dataclass(frozen=True)
class ChannelTiming:
"""Timing for a single color channel within a scanline.
Attributes:
duration_ms: Duration of this channel's pixel data in milliseconds.
"""
duration_ms: float
@dataclass(frozen=True)
class SSTVMode:
"""Complete specification of an SSTV mode.
Attributes:
name: Human-readable mode name (e.g. 'Robot36').
vis_code: VIS code that identifies this mode.
width: Image width in pixels.
height: Image height in lines.
color_model: Color encoding model.
sync_position: Where the sync pulse falls in each line.
sync_duration_ms: Horizontal sync pulse duration (ms).
sync_porch_ms: Porch (gap) after sync pulse (ms).
channels: Timing for each color channel per line.
line_duration_ms: Total duration of one complete scanline (ms).
has_half_rate_chroma: Whether chroma is sent at half vertical rate
(Robot modes: Cr and Cb alternate every other line).
"""
name: str
vis_code: int
width: int
height: int
color_model: ColorModel
sync_position: SyncPosition
sync_duration_ms: float
sync_porch_ms: float
channels: list[ChannelTiming] = field(default_factory=list)
line_duration_ms: float = 0.0
has_half_rate_chroma: bool = False
channel_separator_ms: float = 0.0 # Time gap between color channels (ms)
# ---------------------------------------------------------------------------
# Robot family
# ---------------------------------------------------------------------------
ROBOT_36 = SSTVMode(
name='Robot36',
vis_code=8,
width=320,
height=240,
color_model=ColorModel.YCRCB,
sync_position=SyncPosition.FRONT,
sync_duration_ms=9.0,
sync_porch_ms=3.0,
channels=[
ChannelTiming(duration_ms=88.0), # Y (luminance)
ChannelTiming(duration_ms=44.0), # Cr or Cb (alternating)
],
line_duration_ms=150.0,
has_half_rate_chroma=True,
channel_separator_ms=6.0,
)
ROBOT_72 = SSTVMode(
name='Robot72',
vis_code=12,
width=320,
height=240,
color_model=ColorModel.YCRCB,
sync_position=SyncPosition.FRONT,
sync_duration_ms=9.0,
sync_porch_ms=3.0,
channels=[
ChannelTiming(duration_ms=138.0), # Y (luminance)
ChannelTiming(duration_ms=69.0), # Cr
ChannelTiming(duration_ms=69.0), # Cb
],
line_duration_ms=300.0,
has_half_rate_chroma=False,
channel_separator_ms=6.0,
)
# ---------------------------------------------------------------------------
# Martin family
# ---------------------------------------------------------------------------
MARTIN_1 = SSTVMode(
name='Martin1',
vis_code=44,
width=320,
height=256,
color_model=ColorModel.RGB,
sync_position=SyncPosition.FRONT,
sync_duration_ms=4.862,
sync_porch_ms=0.572,
channels=[
ChannelTiming(duration_ms=146.432), # Green
ChannelTiming(duration_ms=146.432), # Blue
ChannelTiming(duration_ms=146.432), # Red
],
line_duration_ms=446.446,
)
MARTIN_2 = SSTVMode(
name='Martin2',
vis_code=40,
width=320,
height=256,
color_model=ColorModel.RGB,
sync_position=SyncPosition.FRONT,
sync_duration_ms=4.862,
sync_porch_ms=0.572,
channels=[
ChannelTiming(duration_ms=73.216), # Green
ChannelTiming(duration_ms=73.216), # Blue
ChannelTiming(duration_ms=73.216), # Red
],
line_duration_ms=226.798,
)
# ---------------------------------------------------------------------------
# Scottie family
# ---------------------------------------------------------------------------
SCOTTIE_1 = SSTVMode(
name='Scottie1',
vis_code=60,
width=320,
height=256,
color_model=ColorModel.RGB,
sync_position=SyncPosition.MIDDLE,
sync_duration_ms=9.0,
sync_porch_ms=1.5,
channels=[
ChannelTiming(duration_ms=138.240), # Green
ChannelTiming(duration_ms=138.240), # Blue
ChannelTiming(duration_ms=138.240), # Red
],
line_duration_ms=428.220,
)
SCOTTIE_2 = SSTVMode(
name='Scottie2',
vis_code=56,
width=320,
height=256,
color_model=ColorModel.RGB,
sync_position=SyncPosition.MIDDLE,
sync_duration_ms=9.0,
sync_porch_ms=1.5,
channels=[
ChannelTiming(duration_ms=88.064), # Green
ChannelTiming(duration_ms=88.064), # Blue
ChannelTiming(duration_ms=88.064), # Red
],
line_duration_ms=277.692,
)
# ---------------------------------------------------------------------------
# PD (Pasokon) family
# ---------------------------------------------------------------------------
PD_120 = SSTVMode(
name='PD120',
vis_code=93,
width=640,
height=496,
color_model=ColorModel.YCRCB_DUAL,
sync_position=SyncPosition.FRONT_PD,
sync_duration_ms=20.0,
sync_porch_ms=2.080,
channels=[
ChannelTiming(duration_ms=121.600), # Y1 (even line luminance)
ChannelTiming(duration_ms=121.600), # Cr
ChannelTiming(duration_ms=121.600), # Cb
ChannelTiming(duration_ms=121.600), # Y2 (odd line luminance)
],
line_duration_ms=508.480,
)
PD_180 = SSTVMode(
name='PD180',
vis_code=95,
width=640,
height=496,
color_model=ColorModel.YCRCB_DUAL,
sync_position=SyncPosition.FRONT_PD,
sync_duration_ms=20.0,
sync_porch_ms=2.080,
channels=[
ChannelTiming(duration_ms=183.040), # Y1
ChannelTiming(duration_ms=183.040), # Cr
ChannelTiming(duration_ms=183.040), # Cb
ChannelTiming(duration_ms=183.040), # Y2
],
line_duration_ms=754.240,
)
# ---------------------------------------------------------------------------
# Mode registry
# ---------------------------------------------------------------------------
ALL_MODES: dict[int, SSTVMode] = {
m.vis_code: m for m in [
ROBOT_36, ROBOT_72,
MARTIN_1, MARTIN_2,
SCOTTIE_1, SCOTTIE_2,
PD_120, PD_180,
]
}
MODE_BY_NAME: dict[str, SSTVMode] = {m.name: m for m in ALL_MODES.values()}
def get_mode(vis_code: int) -> SSTVMode | None:
"""Look up an SSTV mode by its VIS code."""
return ALL_MODES.get(vis_code)
def get_mode_by_name(name: str) -> SSTVMode | None:
"""Look up an SSTV mode by name."""
return MODE_BY_NAME.get(name)
utils/sstv.py → utils/sstv/sstv_decoder.py
+782 -792
View File
File diff suppressed because it is too large Load Diff
utils/sstv/vis.py
+318
View File
@@ -0,0 +1,318 @@
"""VIS (Vertical Interval Signaling) header detection.
State machine that processes audio samples to detect the VIS header
that precedes every SSTV image transmission. The VIS header identifies
the SSTV mode (Robot36, Martin1, etc.) via an 8-bit code with even parity.
VIS header structure:
Leader tone (1900 Hz, ~300ms)
Break (1200 Hz, ~10ms)
Leader tone (1900 Hz, ~300ms)
Start bit (1200 Hz, 30ms)
8 data bits (1100 Hz = 1, 1300 Hz = 0, 30ms each)
Parity bit (even parity, 30ms)
Stop bit (1200 Hz, 30ms)
"""
from __future__ import annotations
import enum
import numpy as np
from .constants import (
FREQ_LEADER,
FREQ_SYNC,
FREQ_VIS_BIT_0,
FREQ_VIS_BIT_1,
SAMPLE_RATE,
VIS_BIT_DURATION,
VIS_CODES,
VIS_LEADER_MAX,
VIS_LEADER_MIN,
)
from .dsp import goertzel, samples_for_duration
# Use 10ms window (480 samples at 48kHz) for 100Hz frequency resolution.
# This cleanly separates 1100, 1200, 1300, 1500, 1900, 2300 Hz tones.
VIS_WINDOW = 480
class VISState(enum.Enum):
"""States of the VIS detection state machine."""
IDLE = 'idle'
LEADER_1 = 'leader_1'
BREAK = 'break'
LEADER_2 = 'leader_2'
START_BIT = 'start_bit'
DATA_BITS = 'data_bits'
PARITY = 'parity'
STOP_BIT = 'stop_bit'
DETECTED = 'detected'
# The four tone classes we need to distinguish in VIS detection.
_VIS_FREQS = [FREQ_VIS_BIT_1, FREQ_SYNC, FREQ_VIS_BIT_0, FREQ_LEADER]
# 1100, 1200, 1300, 1900 Hz
def _classify_tone(samples: np.ndarray,
sample_rate: int = SAMPLE_RATE) -> float | None:
"""Classify which VIS tone is present in the given samples.
Computes Goertzel energy at each of the four VIS frequencies and returns
the one with the highest energy, provided it dominates sufficiently.
Returns:
The detected frequency (1100, 1200, 1300, or 1900), or None.
"""
if len(samples) < 16:
return None
energies = {f: goertzel(samples, f, sample_rate) for f in _VIS_FREQS}
best_freq = max(energies, key=energies.get) # type: ignore[arg-type]
best_energy = energies[best_freq]
if best_energy <= 0:
return None
# Require the best frequency to be at least 2x stronger than the
# next-strongest tone.
others = sorted(
[e for f, e in energies.items() if f != best_freq], reverse=True)
second_best = others[0] if others else 0.0
if second_best > 0 and best_energy / second_best < 2.0:
return None
return best_freq
class VISDetector:
"""VIS header detection state machine.
Feed audio samples via ``feed()`` and it returns the detected VIS code
(and mode name) when a valid header is found.
The state machine uses a simple approach:
- **Leader detection**: Count consecutive 1900 Hz windows until minimum
leader duration is met.
- **Break/start bit**: Count consecutive 1200 Hz windows. The break is
short; the start bit is one VIS bit duration.
- **Data/parity bits**: Accumulate audio for one bit duration, then
compare 1100 vs 1300 Hz energy to determine bit value.
- **Stop bit**: Count 1200 Hz windows for one bit duration.
Usage::
detector = VISDetector()
for chunk in audio_chunks:
result = detector.feed(chunk)
if result is not None:
vis_code, mode_name = result
"""
def __init__(self, sample_rate: int = SAMPLE_RATE):
self._sample_rate = sample_rate
self._window = VIS_WINDOW
self._bit_samples = samples_for_duration(VIS_BIT_DURATION, sample_rate)
self._leader_min_samples = samples_for_duration(VIS_LEADER_MIN, sample_rate)
self._leader_max_samples = samples_for_duration(VIS_LEADER_MAX, sample_rate)
# Pre-calculate window counts
self._leader_min_windows = max(1, self._leader_min_samples // self._window)
self._leader_max_windows = max(1, self._leader_max_samples // self._window)
self._bit_windows = max(1, self._bit_samples // self._window)
self._state = VISState.IDLE
self._buffer = np.array([], dtype=np.float64)
self._tone_counter = 0
self._data_bits: list[int] = []
self._parity_bit: int = 0
self._bit_accumulator: list[np.ndarray] = []
def reset(self) -> None:
"""Reset the detector to scan for a new VIS header."""
self._state = VISState.IDLE
self._buffer = np.array([], dtype=np.float64)
self._tone_counter = 0
self._data_bits = []
self._parity_bit = 0
self._bit_accumulator = []
@property
def state(self) -> VISState:
return self._state
def feed(self, samples: np.ndarray) -> tuple[int, str] | None:
"""Feed audio samples and attempt VIS detection.
Args:
samples: Float64 audio samples (normalized to -1..1).
Returns:
(vis_code, mode_name) tuple when a valid VIS header is detected,
or None if still scanning.
"""
self._buffer = np.concatenate([self._buffer, samples])
while len(self._buffer) >= self._window:
result = self._process_window(self._buffer[:self._window])
self._buffer = self._buffer[self._window:]
if result is not None:
return result
return None
def _process_window(self, window: np.ndarray) -> tuple[int, str] | None:
"""Process a single analysis window through the state machine.
The key design: when a state transition occurs due to a tone change,
the window that triggers the transition counts as the first window
of the new state (tone_counter = 1).
"""
tone = _classify_tone(window, self._sample_rate)
if self._state == VISState.IDLE:
if tone == FREQ_LEADER:
self._tone_counter += 1
if self._tone_counter >= self._leader_min_windows:
self._state = VISState.LEADER_1
else:
self._tone_counter = 0
elif self._state == VISState.LEADER_1:
if tone == FREQ_LEADER:
self._tone_counter += 1
if self._tone_counter > self._leader_max_windows * 3:
self._tone_counter = 0
self._state = VISState.IDLE
elif tone == FREQ_SYNC:
# Transition to BREAK; this window counts as break window 1
self._tone_counter = 1
self._state = VISState.BREAK
else:
self._tone_counter = 0
self._state = VISState.IDLE
elif self._state == VISState.BREAK:
if tone == FREQ_SYNC:
self._tone_counter += 1
if self._tone_counter > 10:
self._tone_counter = 0
self._state = VISState.IDLE
elif tone == FREQ_LEADER:
# Transition to LEADER_2; this window counts
self._tone_counter = 1
self._state = VISState.LEADER_2
else:
self._tone_counter = 0
self._state = VISState.IDLE
elif self._state == VISState.LEADER_2:
if tone == FREQ_LEADER:
self._tone_counter += 1
if self._tone_counter > self._leader_max_windows * 3:
self._tone_counter = 0
self._state = VISState.IDLE
elif tone == FREQ_SYNC:
# Transition to START_BIT; this window counts
self._tone_counter = 1
self._state = VISState.START_BIT
# Check if start bit is already complete (1-window bit)
if self._tone_counter >= self._bit_windows:
self._tone_counter = 0
self._data_bits = []
self._bit_accumulator = []
self._state = VISState.DATA_BITS
else:
self._tone_counter = 0
self._state = VISState.IDLE
elif self._state == VISState.START_BIT:
if tone == FREQ_SYNC:
self._tone_counter += 1
if self._tone_counter >= self._bit_windows:
self._tone_counter = 0
self._data_bits = []
self._bit_accumulator = []
self._state = VISState.DATA_BITS
else:
# Non-sync during start bit: check if we had enough sync
# windows already (tolerant: accept if within 1 window)
if self._tone_counter >= self._bit_windows - 1:
# Close enough - accept and process this window as data
self._data_bits = []
self._bit_accumulator = [window]
self._tone_counter = 1
self._state = VISState.DATA_BITS
else:
self._tone_counter = 0
self._state = VISState.IDLE
elif self._state == VISState.DATA_BITS:
self._tone_counter += 1
self._bit_accumulator.append(window)
if self._tone_counter >= self._bit_windows:
bit_audio = np.concatenate(self._bit_accumulator)
bit_val = self._decode_bit(bit_audio)
self._data_bits.append(bit_val)
self._tone_counter = 0
self._bit_accumulator = []
if len(self._data_bits) == 8:
self._state = VISState.PARITY
elif self._state == VISState.PARITY:
self._tone_counter += 1
self._bit_accumulator.append(window)
if self._tone_counter >= self._bit_windows:
bit_audio = np.concatenate(self._bit_accumulator)
self._parity_bit = self._decode_bit(bit_audio)
self._tone_counter = 0
self._bit_accumulator = []
self._state = VISState.STOP_BIT
elif self._state == VISState.STOP_BIT:
self._tone_counter += 1
if self._tone_counter >= self._bit_windows:
result = self._validate_and_decode()
self.reset()
return result
return None
def _decode_bit(self, samples: np.ndarray) -> int:
"""Decode a single VIS data bit from its audio samples.
Compares Goertzel energy at 1100 Hz (bit=1) vs 1300 Hz (bit=0).
"""
e1 = goertzel(samples, FREQ_VIS_BIT_1, self._sample_rate)
e0 = goertzel(samples, FREQ_VIS_BIT_0, self._sample_rate)
return 1 if e1 > e0 else 0
def _validate_and_decode(self) -> tuple[int, str] | None:
"""Validate parity and decode the VIS code.
Returns:
(vis_code, mode_name) or None if validation fails.
"""
if len(self._data_bits) != 8:
return None
# Decode VIS code (LSB first)
vis_code = 0
for i, bit in enumerate(self._data_bits):
vis_code |= bit << i
# Look up mode
mode_name = VIS_CODES.get(vis_code)
if mode_name is not None:
return vis_code, mode_name
return None