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

Merge upstream/main and resolve weather-satellite.js conflict

Browse Source 
Resolved conflict in static/js/modes/weather-satellite.js:
- Kept allPasses state variable and applyPassFilter() for satellite pass filtering
- Kept satellite select dropdown listener for filter feature
- Adopted upstream's optimistic stop() UI pattern for better responsiveness
- Kept optional chaining (pass?.trajectory) since drawPolarPlot can receive null

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
mitchross
2026-02-26 00:37:02 -05:00
parent 57b1b3a3ae 477506f902
commit c75c8f8821
71 changed files with 13181 additions and 3658 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/dsc/constants.py
+17 -21
View File
@@ -14,30 +14,26 @@ from __future__ import annotations
# =============================================================================
FORMAT_CODES = {
100: 'DISTRESS', # All ships distress alert
102: 'ALL_SHIPS', # All ships call
104: 'GROUP', # Group call
106: 'DISTRESS_ACK', # Distress acknowledgement
108: 'DISTRESS_RELAY', # Distress relay
110: 'GEOGRAPHIC', # Geographic area call
112: 'INDIVIDUAL', # Individual call
114: 'INDIVIDUAL_ACK', # Individual acknowledgement
116: 'ROUTINE', # Routine call
118: 'SAFETY', # Safety call
120: 'URGENCY', # Urgency call
102: 'ALL_SHIPS', # All ships call
112: 'INDIVIDUAL', # Individual call
114: 'INDIVIDUAL_ACK', # Individual acknowledgement
116: 'GROUP', # Group call (including geographic area)
120: 'DISTRESS', # Distress alert
123: 'ALL_SHIPS_URGENCY_SAFETY', # All ships urgency/safety
}
# Valid ITU-R M.493 format specifiers
VALID_FORMAT_SPECIFIERS = {102, 112, 114, 116, 120, 123}
# Valid EOS (End of Sequence) symbols per ITU-R M.493
VALID_EOS = {117, 122, 127}
# Category priority (lower = higher priority)
CATEGORY_PRIORITY = {
'DISTRESS': 0,
'DISTRESS_ACK': 1,
'DISTRESS_RELAY': 2,
'URGENCY': 3,
'SAFETY': 4,
'ROUTINE': 5,
'ALL_SHIPS_URGENCY_SAFETY': 2,
'ALL_SHIPS': 5,
'GROUP': 5,
'GEOGRAPHIC': 5,
'INDIVIDUAL': 5,
'INDIVIDUAL_ACK': 5,
}
@@ -453,11 +449,11 @@ VHF_CHANNELS = {
# DSC Modulation Parameters
# =============================================================================
DSC_BAUD_RATE = 100 # 100 baud per ITU-R M.493
DSC_BAUD_RATE = 1200 # 1200 bps per ITU-R M.493
# FSK tone frequencies (Hz)
DSC_MARK_FREQ = 1800 # B (mark) - binary 1
DSC_SPACE_FREQ = 1200 # Y (space) - binary 0
# FSK tone frequencies (Hz) on 1700 Hz subcarrier
DSC_MARK_FREQ = 2100 # B (mark) - binary 1
DSC_SPACE_FREQ = 1300 # Y (space) - binary 0
# Audio sample rate for decoding
DSC_AUDIO_SAMPLE_RATE = 48000
utils/dsc/decoder.py
+23 -21
View File
@@ -5,9 +5,9 @@ DSC (Digital Selective Calling) decoder.
Decodes VHF DSC signals per ITU-R M.493. Reads 48kHz 16-bit signed
audio from stdin (from rtl_fm) and outputs JSON messages to stdout.
DSC uses 100 baud FSK with:
- Mark (1): 1800 Hz
- Space (0): 1200 Hz
DSC uses 1200 bps FSK on a 1700 Hz subcarrier with:
- Mark (1): 2100 Hz
- Space (0): 1300 Hz
Frame structure:
1. Dot pattern: 200 bits alternating 1/0 for synchronization
@@ -42,6 +42,7 @@ from .constants import (
DSC_AUDIO_SAMPLE_RATE,
FORMAT_CODES,
DISTRESS_NATURE_CODES,
VALID_EOS,
)
# Configure logging
@@ -57,7 +58,7 @@ class DSCDecoder:
"""
DSC FSK decoder.
Demodulates 100 baud FSK audio and decodes DSC protocol.
Demodulates 1200 bps FSK audio and decodes DSC protocol.
"""
def __init__(self, sample_rate: int = DSC_AUDIO_SAMPLE_RATE):
@@ -66,13 +67,13 @@ class DSCDecoder:
self.samples_per_bit = sample_rate // self.baud_rate
# FSK frequencies
self.mark_freq = DSC_MARK_FREQ # 1800 Hz = binary 1
self.space_freq = DSC_SPACE_FREQ # 1200 Hz = binary 0
self.mark_freq = DSC_MARK_FREQ # 2100 Hz = binary 1
self.space_freq = DSC_SPACE_FREQ # 1300 Hz = binary 0
# Bandpass filter for DSC band (1100-1900 Hz)
# Bandpass filter for DSC band (1100-2300 Hz)
nyq = sample_rate / 2
low = 1100 / nyq
high = 1900 / nyq
high = 2300 / nyq
self.bp_b, self.bp_a = scipy_signal.butter(4, [low, high], btype='band')
# Build FSK correlators
@@ -278,11 +279,11 @@ class DSCDecoder:
if len(symbols) < 5:
return None
# Look for EOS (End of Sequence) - symbol 127
# Look for EOS (End of Sequence) - symbols 117, 122, or 127
eos_found = False
eos_index = -1
for i, sym in enumerate(symbols):
if sym == 127: # EOS symbol
if sym in VALID_EOS:
eos_found = True
eos_index = i
break
@@ -337,20 +338,21 @@ class DSCDecoder:
format_code = symbols[0]
format_text = FORMAT_CODES.get(format_code, f'UNKNOWN-{format_code}')
# Determine category from format
category = 'ROUTINE'
if format_code == 100:
# Derive category from format specifier per ITU-R M.493
if format_code == 120:
category = 'DISTRESS'
elif format_code == 106:
category = 'DISTRESS_ACK'
elif format_code == 108:
category = 'DISTRESS_RELAY'
elif format_code == 118:
category = 'SAFETY'
elif format_code == 120:
category = 'URGENCY'
elif format_code == 123:
category = 'ALL_SHIPS_URGENCY_SAFETY'
elif format_code == 102:
category = 'ALL_SHIPS'
elif format_code == 116:
category = 'GROUP'
elif format_code == 112:
category = 'INDIVIDUAL'
elif format_code == 114:
category = 'INDIVIDUAL_ACK'
else:
category = FORMAT_CODES.get(format_code, 'UNKNOWN')
# Decode MMSI from symbols 1-5 (destination/address)
dest_mmsi = self._decode_mmsi(symbols[1:6])
utils/dsc/parser.py
+60 -9
View File
@@ -19,6 +19,8 @@ from .constants import (
TELECOMMAND_CODES,
CATEGORY_PRIORITY,
MID_COUNTRY_MAP,
VALID_FORMAT_SPECIFIERS,
VALID_EOS,
)
logger = logging.getLogger('intercept.dsc.parser')
@@ -139,13 +141,62 @@ def parse_dsc_message(raw_line: str) -> dict[str, Any] | None:
if 'source_mmsi' not in data:
return None
# ITU-R M.493 validation: format specifier must be valid
format_code = data.get('format')
if format_code not in VALID_FORMAT_SPECIFIERS:
logger.debug(f"Rejected DSC: invalid format specifier {format_code}")
return None
# Validate MMSIs
source_mmsi = str(data.get('source_mmsi', ''))
if not validate_mmsi(source_mmsi):
logger.debug(f"Rejected DSC: invalid source MMSI {source_mmsi}")
return None
dest_mmsi_val = data.get('dest_mmsi')
if dest_mmsi_val is not None:
dest_mmsi_str = str(dest_mmsi_val)
if not validate_mmsi(dest_mmsi_str):
logger.debug(f"Rejected DSC: invalid dest MMSI {dest_mmsi_str}")
return None
# Validate raw field structure if present
raw = data.get('raw')
if raw is not None:
raw_str = str(raw)
if not re.match(r'^\d+$', raw_str):
logger.debug("Rejected DSC: raw field contains non-digits")
return None
if len(raw_str) % 3 != 0:
logger.debug("Rejected DSC: raw field length not divisible by 3")
return None
# Last 3-digit token must be a valid EOS symbol
if len(raw_str) >= 3:
last_token = int(raw_str[-3:])
if last_token not in VALID_EOS:
logger.debug(f"Rejected DSC: raw EOS token {last_token} not valid")
return None
# Validate telecommand values if present (must be 100-127)
for tc_field in ('telecommand1', 'telecommand2'):
tc_val = data.get(tc_field)
if tc_val is not None:
try:
tc_int = int(tc_val)
except (ValueError, TypeError):
logger.debug(f"Rejected DSC: invalid {tc_field} value {tc_val}")
return None
if tc_int < 100 or tc_int > 127:
logger.debug(f"Rejected DSC: {tc_field} {tc_int} out of range 100-127")
return None
# Build parsed message
msg = {
'type': 'dsc_message',
'source_mmsi': str(data.get('source_mmsi', '')),
'dest_mmsi': str(data.get('dest_mmsi', '')) if data.get('dest_mmsi') else None,
'format_code': data.get('format'),
'format_text': get_format_text(data.get('format', 0)),
'source_mmsi': source_mmsi,
'dest_mmsi': str(data.get('dest_mmsi', '')) if data.get('dest_mmsi') is not None else None,
'format_code': format_code,
'format_text': get_format_text(format_code),
'category': data.get('category', 'UNKNOWN').upper(),
'timestamp': data.get('timestamp') or datetime.utcnow().isoformat(),
}
@@ -156,7 +207,7 @@ def parse_dsc_message(raw_line: str) -> dict[str, Any] | None:
msg['source_country'] = country
# Add distress nature if present
if 'nature' in data and data['nature']:
if data.get('nature') is not None:
msg['nature_code'] = data['nature']
msg['nature_of_distress'] = get_distress_nature_text(data['nature'])
@@ -173,16 +224,16 @@ def parse_dsc_message(raw_line: str) -> dict[str, Any] | None:
pass
# Add telecommand info
if 'telecommand1' in data and data['telecommand1']:
if data.get('telecommand1') is not None:
msg['telecommand1'] = data['telecommand1']
msg['telecommand1_text'] = get_telecommand_text(data['telecommand1'])
if 'telecommand2' in data and data['telecommand2']:
if data.get('telecommand2') is not None:
msg['telecommand2'] = data['telecommand2']
msg['telecommand2_text'] = get_telecommand_text(data['telecommand2'])
# Add channel if present
if 'channel' in data and data['channel']:
if data.get('channel') is not None:
msg['channel'] = data['channel']
# Add EOS (End of Sequence) info
@@ -197,7 +248,7 @@ def parse_dsc_message(raw_line: str) -> dict[str, Any] | None:
msg['priority'] = get_category_priority(msg['category'])
# Mark if this is a critical alert
msg['is_critical'] = msg['category'] in ('DISTRESS', 'DISTRESS_ACK', 'DISTRESS_RELAY', 'URGENCY')
msg['is_critical'] = msg['category'] in ('DISTRESS', 'ALL_SHIPS_URGENCY_SAFETY')
return msg
utils/gps.py
+8 -6
View File
@@ -194,18 +194,20 @@ class GPSDClient:
"""Return gpsd connection info."""
return f"gpsd://{self.host}:{self.port}"
def add_callback(self, callback: Callable[[GPSPosition], None]) -> None:
"""Add a callback to be called on position updates."""
self._callbacks.append(callback)
def add_callback(self, callback: Callable[[GPSPosition], None]) -> None:
"""Add a callback to be called on position updates."""
if callback not in self._callbacks:
self._callbacks.append(callback)
def remove_callback(self, callback: Callable[[GPSPosition], None]) -> None:
"""Remove a position update callback."""
if callback in self._callbacks:
self._callbacks.remove(callback)
def add_sky_callback(self, callback: Callable[[GPSSkyData], None]) -> None:
"""Add a callback to be called on sky data updates."""
self._sky_callbacks.append(callback)
def add_sky_callback(self, callback: Callable[[GPSSkyData], None]) -> None:
"""Add a callback to be called on sky data updates."""
if callback not in self._sky_callbacks:
self._sky_callbacks.append(callback)
def remove_sky_callback(self, callback: Callable[[GPSSkyData], None]) -> None:
"""Remove a sky data update callback."""
utils/meshtastic.py
+59 -40
View File
@@ -376,63 +376,82 @@ class MeshtasticClient:
self._error = "Meshtastic SDK not installed. Install with: pip install meshtastic"
return False
# Quick check under lock — bail if already running
with self._lock:
if self._running:
return True
try:
# Subscribe to message events before connecting
pub.subscribe(self._on_receive, "meshtastic.receive")
pub.subscribe(self._on_connection, "meshtastic.connection.established")
pub.subscribe(self._on_disconnect, "meshtastic.connection.lost")
# Create interface outside lock (blocking I/O: serial/TCP connect)
new_interface = None
new_device_path = None
new_connection_type = None
try:
# Subscribe to message events before connecting
pub.subscribe(self._on_receive, "meshtastic.receive")
pub.subscribe(self._on_connection, "meshtastic.connection.established")
pub.subscribe(self._on_disconnect, "meshtastic.connection.lost")
# Connect based on connection type
if connection_type == 'tcp':
if not hostname:
self._error = "Hostname is required for TCP connections"
self._cleanup_subscriptions()
return False
self._interface = meshtastic.tcp_interface.TCPInterface(hostname=hostname)
self._device_path = hostname
self._connection_type = 'tcp'
logger.info(f"Connected to Meshtastic device via TCP: {hostname}")
if connection_type == 'tcp':
if not hostname:
self._error = "Hostname is required for TCP connections"
self._cleanup_subscriptions()
return False
new_interface = meshtastic.tcp_interface.TCPInterface(hostname=hostname)
new_device_path = hostname
new_connection_type = 'tcp'
logger.info(f"Connected to Meshtastic device via TCP: {hostname}")
else:
if device:
new_interface = meshtastic.serial_interface.SerialInterface(device)
new_device_path = device
else:
# Serial connection (default)
if device:
self._interface = meshtastic.serial_interface.SerialInterface(device)
self._device_path = device
else:
# Auto-discover
self._interface = meshtastic.serial_interface.SerialInterface()
self._device_path = "auto"
self._connection_type = 'serial'
logger.info(f"Connected to Meshtastic device via serial: {self._device_path}")
new_interface = meshtastic.serial_interface.SerialInterface()
new_device_path = "auto"
new_connection_type = 'serial'
logger.info(f"Connected to Meshtastic device via serial: {new_device_path}")
except Exception as e:
self._error = str(e)
logger.error(f"Failed to connect to Meshtastic: {e}")
self._cleanup_subscriptions()
return False
self._running = True
self._error = None
# Install interface under lock
with self._lock:
if self._running:
# Another thread connected while we were connecting — discard ours
if new_interface:
try:
new_interface.close()
except Exception:
pass
return True
except Exception as e:
self._error = str(e)
logger.error(f"Failed to connect to Meshtastic: {e}")
self._cleanup_subscriptions()
return False
self._interface = new_interface
self._device_path = new_device_path
self._connection_type = new_connection_type
self._running = True
self._error = None
return True
def disconnect(self) -> None:
"""Disconnect from the Meshtastic device."""
iface_to_close = None
with self._lock:
if self._interface:
try:
self._interface.close()
except Exception as e:
logger.warning(f"Error closing Meshtastic interface: {e}")
self._interface = None
iface_to_close = self._interface
self._interface = None
self._cleanup_subscriptions()
self._running = False
self._device_path = None
self._connection_type = None
logger.info("Disconnected from Meshtastic device")
# Close interface outside lock (blocking I/O)
if iface_to_close:
try:
iface_to_close.close()
except Exception as e:
logger.warning(f"Error closing Meshtastic interface: {e}")
logger.info("Disconnected from Meshtastic device")
def _cleanup_subscriptions(self) -> None:
"""Unsubscribe from pubsub topics."""
utils/morse.py
+276
View File
@@ -0,0 +1,276 @@
"""Morse code (CW) decoder using Goertzel tone detection.
Signal chain: rtl_fm -M usb → raw PCM → Goertzel filter → timing state machine → characters.
"""
from __future__ import annotations
import contextlib
import math
import queue
import struct
import threading
import time
from datetime import datetime
from typing import Any
# International Morse Code table
MORSE_TABLE: dict[str, str] = {
'.-': 'A', '-...': 'B', '-.-.': 'C', '-..': 'D', '.': 'E',
'..-.': 'F', '--.': 'G', '....': 'H', '..': 'I', '.---': 'J',
'-.-': 'K', '.-..': 'L', '--': 'M', '-.': 'N', '---': 'O',
'.--.': 'P', '--.-': 'Q', '.-.': 'R', '...': 'S', '-': 'T',
'..-': 'U', '...-': 'V', '.--': 'W', '-..-': 'X', '-.--': 'Y',
'--..': 'Z',
'-----': '0', '.----': '1', '..---': '2', '...--': '3',
'....-': '4', '.....': '5', '-....': '6', '--...': '7',
'---..': '8', '----.': '9',
'.-.-.-': '.', '--..--': ',', '..--..': '?', '.----.': "'",
'-.-.--': '!', '-..-.': '/', '-.--.': '(', '-.--.-': ')',
'.-...': '&', '---...': ':', '-.-.-.': ';', '-...-': '=',
'.-.-.': '+', '-....-': '-', '..--.-': '_', '.-..-.': '"',
'...-..-': '$', '.--.-.': '@',
# Prosigns (unique codes only; -...- and -.--.- already mapped above)
'-.-.-': '<CT>', '.-.-': '<AA>', '...-.-': '<SK>',
}
# Reverse lookup: character → morse notation
CHAR_TO_MORSE: dict[str, str] = {v: k for k, v in MORSE_TABLE.items()}
class GoertzelFilter:
"""Single-frequency tone detector using the Goertzel algorithm.
O(N) per block, much cheaper than FFT for detecting one frequency.
"""
def __init__(self, target_freq: float, sample_rate: int, block_size: int):
self.target_freq = target_freq
self.sample_rate = sample_rate
self.block_size = block_size
# Precompute coefficient
k = round(target_freq * block_size / sample_rate)
omega = 2.0 * math.pi * k / block_size
self.coeff = 2.0 * math.cos(omega)
def magnitude(self, samples: list[float] | tuple[float, ...]) -> float:
"""Compute magnitude of the target frequency in the sample block."""
s0 = 0.0
s1 = 0.0
s2 = 0.0
coeff = self.coeff
for sample in samples:
s0 = sample + coeff * s1 - s2
s2 = s1
s1 = s0
return math.sqrt(s1 * s1 + s2 * s2 - coeff * s1 * s2)
class MorseDecoder:
"""Real-time Morse decoder with adaptive threshold.
Processes blocks of PCM audio and emits decoded characters.
Timing based on PARIS standard: dit = 1.2/WPM seconds.
"""
def __init__(
self,
sample_rate: int = 8000,
tone_freq: float = 700.0,
wpm: int = 15,
):
self.sample_rate = sample_rate
self.tone_freq = tone_freq
self.wpm = wpm
# Goertzel filter: ~50 blocks/sec at 8kHz
self._block_size = sample_rate // 50
self._filter = GoertzelFilter(tone_freq, sample_rate, self._block_size)
self._block_duration = self._block_size / sample_rate # seconds per block
# Timing thresholds (in blocks, converted from seconds)
dit_sec = 1.2 / wpm
self._dah_threshold = 2.0 * dit_sec / self._block_duration # blocks
self._dit_min = 0.3 * dit_sec / self._block_duration # min blocks for dit
self._char_gap = 3.0 * dit_sec / self._block_duration # blocks
self._word_gap = 7.0 * dit_sec / self._block_duration # blocks
# Adaptive threshold via EMA
self._noise_floor = 0.0
self._signal_peak = 0.0
self._threshold = 0.0
self._ema_alpha = 0.1 # smoothing factor
# State machine (counts in blocks, not wall-clock time)
self._tone_on = False
self._tone_blocks = 0 # blocks since tone started
self._silence_blocks = 0 # blocks since silence started
self._current_symbol = '' # accumulates dits/dahs for current char
self._pending_buffer: list[float] = []
self._blocks_processed = 0 # total blocks for warm-up tracking
def process_block(self, pcm_bytes: bytes) -> list[dict[str, Any]]:
"""Process a chunk of 16-bit LE PCM and return decoded events.
Returns list of event dicts with keys:
type: 'scope' | 'morse_char' | 'morse_space'
+ type-specific fields
"""
events: list[dict[str, Any]] = []
# Unpack PCM samples
n_samples = len(pcm_bytes) // 2
if n_samples == 0:
return events
samples = struct.unpack(f'<{n_samples}h', pcm_bytes[:n_samples * 2])
# Feed samples into pending buffer and process in blocks
self._pending_buffer.extend(samples)
amplitudes: list[float] = []
while len(self._pending_buffer) >= self._block_size:
block = self._pending_buffer[:self._block_size]
self._pending_buffer = self._pending_buffer[self._block_size:]
# Normalize to [-1, 1]
normalized = [s / 32768.0 for s in block]
mag = self._filter.magnitude(normalized)
amplitudes.append(mag)
self._blocks_processed += 1
# Update adaptive threshold
if mag < self._threshold or self._threshold == 0:
self._noise_floor += self._ema_alpha * (mag - self._noise_floor)
else:
self._signal_peak += self._ema_alpha * (mag - self._signal_peak)
self._threshold = (self._noise_floor + self._signal_peak) / 2.0
tone_detected = mag > self._threshold and self._threshold > 0
if tone_detected and not self._tone_on:
# Tone just started - check silence duration for gaps
self._tone_on = True
silence_count = self._silence_blocks
self._tone_blocks = 0
if self._current_symbol and silence_count >= self._char_gap:
# Character gap - decode accumulated symbol
char = MORSE_TABLE.get(self._current_symbol)
if char:
events.append({
'type': 'morse_char',
'char': char,
'morse': self._current_symbol,
'timestamp': datetime.now().strftime('%H:%M:%S'),
})
if silence_count >= self._word_gap:
events.append({
'type': 'morse_space',
'timestamp': datetime.now().strftime('%H:%M:%S'),
})
self._current_symbol = ''
elif not tone_detected and self._tone_on:
# Tone just ended - classify as dit or dah
self._tone_on = False
tone_count = self._tone_blocks
self._silence_blocks = 0
if tone_count >= self._dah_threshold:
self._current_symbol += '-'
elif tone_count >= self._dit_min:
self._current_symbol += '.'
elif tone_detected and self._tone_on:
self._tone_blocks += 1
elif not tone_detected and not self._tone_on:
self._silence_blocks += 1
# Emit scope data for visualization (~10 Hz is handled by caller)
if amplitudes:
events.append({
'type': 'scope',
'amplitudes': amplitudes,
'threshold': self._threshold,
'tone_on': self._tone_on,
})
return events
def flush(self) -> list[dict[str, Any]]:
"""Flush any pending symbol at end of stream."""
events: list[dict[str, Any]] = []
if self._current_symbol:
char = MORSE_TABLE.get(self._current_symbol)
if char:
events.append({
'type': 'morse_char',
'char': char,
'morse': self._current_symbol,
'timestamp': datetime.now().strftime('%H:%M:%S'),
})
self._current_symbol = ''
return events
def morse_decoder_thread(
rtl_stdout,
output_queue: queue.Queue,
stop_event: threading.Event,
sample_rate: int = 8000,
tone_freq: float = 700.0,
wpm: int = 15,
) -> None:
"""Thread function: reads PCM from rtl_fm, decodes Morse, pushes to queue.
Reads raw 16-bit LE PCM from *rtl_stdout* and feeds it through the
MorseDecoder, pushing scope and character events onto *output_queue*.
"""
import logging
logger = logging.getLogger('intercept.morse')
CHUNK = 4096 # bytes per read (2048 samples at 16-bit mono)
SCOPE_INTERVAL = 0.1 # scope updates at ~10 Hz
last_scope = time.monotonic()
decoder = MorseDecoder(
sample_rate=sample_rate,
tone_freq=tone_freq,
wpm=wpm,
)
try:
while not stop_event.is_set():
data = rtl_stdout.read(CHUNK)
if not data:
break
events = decoder.process_block(data)
for event in events:
if event['type'] == 'scope':
# Throttle scope events to ~10 Hz
now = time.monotonic()
if now - last_scope >= SCOPE_INTERVAL:
last_scope = now
with contextlib.suppress(queue.Full):
output_queue.put_nowait(event)
else:
# Character and space events always go through
with contextlib.suppress(queue.Full):
output_queue.put_nowait(event)
except Exception as e:
logger.debug(f"Morse decoder thread error: {e}")
finally:
# Flush any pending symbol
for event in decoder.flush():
with contextlib.suppress(queue.Full):
output_queue.put_nowait(event)
utils/process_monitor.py
+16 -7
View File
@@ -112,6 +112,8 @@ class ProcessMonitor:
def _check_all_processes(self) -> None:
"""Check health of all registered processes."""
# Collect crashed processes under lock, handle restarts outside
crashed: list[tuple[str, ProcessInfo]] = []
with self._lock:
for name, info in list(self.processes.items()):
if not info.enabled:
@@ -126,10 +128,14 @@ class ProcessMonitor:
logger.warning(
f"Process '{name}' terminated with code {return_code}"
)
self._handle_crash(name, info)
crashed.append((name, info))
# Handle restarts outside lock (involves sleeps and callbacks)
for name, info in crashed:
self._handle_crash(name, info)
def _handle_crash(self, name: str, info: ProcessInfo) -> None:
"""Handle a crashed process."""
"""Handle a crashed process. Must be called WITHOUT holding self._lock."""
if info.restart_callback is None:
logger.info(f"No restart callback for '{name}', skipping auto-restart")
return
@@ -139,7 +145,8 @@ class ProcessMonitor:
f"Process '{name}' exceeded max restarts ({info.max_restarts}), "
"disabling auto-restart"
)
info.enabled = False
with self._lock:
info.enabled = False
return
# Calculate backoff with exponential increase
@@ -149,18 +156,20 @@ class ProcessMonitor:
f"(attempt {info.restart_count + 1}/{info.max_restarts})"
)
# Wait for backoff period
# Wait for backoff period outside lock
time.sleep(backoff)
# Attempt restart
try:
info.restart_callback()
info.restart_count += 1
info.last_restart = datetime.now()
with self._lock:
info.restart_count += 1
info.last_restart = datetime.now()
logger.info(f"Successfully restarted '{name}'")
except Exception as e:
logger.error(f"Failed to restart '{name}': {e}")
info.restart_count += 1
with self._lock:
info.restart_count += 1
def get_status(self) -> Dict[str, Any]:
"""
utils/sdr/airspy.py
+6 -2
View File
@@ -10,6 +10,8 @@ from __future__ import annotations
from typing import Optional
from utils.dependencies import get_tool_path
from .base import CommandBuilder, SDRCapabilities, SDRDevice, SDRType
@@ -74,8 +76,9 @@ class AirspyCommandBuilder(CommandBuilder):
"""
device_str = self._build_device_string(device)
rx_fm_path = get_tool_path('rx_fm') or 'rx_fm'
cmd = [
'rx_fm',
rx_fm_path,
'-d', device_str,
'-f', f'{frequency_mhz}M',
'-M', modulation,
@@ -203,8 +206,9 @@ class AirspyCommandBuilder(CommandBuilder):
device_str = self._build_device_string(device)
freq_hz = int(frequency_mhz * 1e6)
rx_sdr_path = get_tool_path('rx_sdr') or 'rx_sdr'
cmd = [
'rx_sdr',
rx_sdr_path,
'-d', device_str,
'-f', str(freq_hz),
'-s', str(sample_rate),
utils/sdr/hackrf.py
+6 -2
View File
@@ -9,6 +9,8 @@ from __future__ import annotations
from typing import Optional
from utils.dependencies import get_tool_path
from .base import CommandBuilder, SDRCapabilities, SDRDevice, SDRType
@@ -70,8 +72,9 @@ class HackRFCommandBuilder(CommandBuilder):
"""
device_str = self._build_device_string(device)
rx_fm_path = get_tool_path('rx_fm') or 'rx_fm'
cmd = [
'rx_fm',
rx_fm_path,
'-d', device_str,
'-f', f'{frequency_mhz}M',
'-M', modulation,
@@ -203,8 +206,9 @@ class HackRFCommandBuilder(CommandBuilder):
device_str = self._build_device_string(device)
freq_hz = int(frequency_mhz * 1e6)
rx_sdr_path = get_tool_path('rx_sdr') or 'rx_sdr'
cmd = [
'rx_sdr',
rx_sdr_path,
'-d', device_str,
'-f', str(freq_hz),
'-s', str(sample_rate),
utils/sdr/limesdr.py
+6 -2
View File
@@ -9,6 +9,8 @@ from __future__ import annotations
from typing import Optional
from utils.dependencies import get_tool_path
from .base import CommandBuilder, SDRCapabilities, SDRDevice, SDRType
@@ -52,8 +54,9 @@ class LimeSDRCommandBuilder(CommandBuilder):
"""
device_str = self._build_device_string(device)
rx_fm_path = get_tool_path('rx_fm') or 'rx_fm'
cmd = [
'rx_fm',
rx_fm_path,
'-d', device_str,
'-f', f'{frequency_mhz}M',
'-M', modulation,
@@ -181,8 +184,9 @@ class LimeSDRCommandBuilder(CommandBuilder):
device_str = self._build_device_string(device)
freq_hz = int(frequency_mhz * 1e6)
rx_sdr_path = get_tool_path('rx_sdr') or 'rx_sdr'
cmd = [
'rx_sdr',
rx_sdr_path,
'-d', device_str,
'-f', str(freq_hz),
'-s', str(sample_rate),
utils/sdr/sdrplay.py
+6 -2
View File
@@ -9,6 +9,8 @@ from __future__ import annotations
from typing import Optional
from utils.dependencies import get_tool_path
from .base import CommandBuilder, SDRCapabilities, SDRDevice, SDRType
@@ -52,8 +54,9 @@ class SDRPlayCommandBuilder(CommandBuilder):
"""
device_str = self._build_device_string(device)
rx_fm_path = get_tool_path('rx_fm') or 'rx_fm'
cmd = [
'rx_fm',
rx_fm_path,
'-d', device_str,
'-f', f'{frequency_mhz}M',
'-M', modulation,
@@ -181,8 +184,9 @@ class SDRPlayCommandBuilder(CommandBuilder):
device_str = self._build_device_string(device)
freq_hz = int(frequency_mhz * 1e6)
rx_sdr_path = get_tool_path('rx_sdr') or 'rx_sdr'
cmd = [
'rx_sdr',
rx_sdr_path,
'-d', device_str,
'-f', str(freq_hz),
'-s', str(sample_rate),
utils/sse.py
+36 -6
View File
@@ -26,15 +26,33 @@ _fanout_channels_lock = threading.Lock()
def _run_fanout(channel: _QueueFanoutChannel) -> None:
"""Drain source queue and fan out each message to all subscribers."""
idle_drain_batch = 512
while True:
with channel.lock:
subscribers = tuple(channel.subscribers)
if not subscribers:
# Keep ingest pipelines responsive even if UI clients disconnect:
# drain and drop stale backlog while idle so producer threads do
# not block on full source queues.
drained = 0
for _ in range(idle_drain_batch):
try:
channel.source_queue.get_nowait()
drained += 1
except queue.Empty:
break
if drained == 0:
time.sleep(channel.source_timeout)
continue
try:
msg = channel.source_queue.get(timeout=channel.source_timeout)
except queue.Empty:
continue
with channel.lock:
subscribers = tuple(channel.subscribers)
for subscriber in subscribers:
try:
subscriber.put_nowait(msg)
@@ -52,13 +70,24 @@ def _ensure_fanout_channel(
source_queue: queue.Queue,
source_timeout: float,
) -> _QueueFanoutChannel:
"""Get/create a fanout channel and ensure distributor thread is running."""
"""Get/create a fanout channel."""
with _fanout_channels_lock:
channel = _fanout_channels.get(channel_key)
if channel is None:
channel = _QueueFanoutChannel(source_queue=source_queue, source_timeout=source_timeout)
_fanout_channels[channel_key] = channel
if channel.source_queue is not source_queue:
# Keep channel in sync if source queue object is replaced.
channel.source_queue = source_queue
channel.source_timeout = source_timeout
return channel
def _ensure_distributor_running(channel: _QueueFanoutChannel, channel_key: str) -> None:
"""Ensure fanout distributor thread is running for a channel."""
with _fanout_channels_lock:
if channel.distributor is None or not channel.distributor.is_alive():
channel.distributor = threading.Thread(
target=_run_fanout,
@@ -68,8 +97,6 @@ def _ensure_fanout_channel(
)
channel.distributor.start()
return channel
def subscribe_fanout_queue(
source_queue: queue.Queue,
@@ -89,6 +116,9 @@ def subscribe_fanout_queue(
with channel.lock:
channel.subscribers.add(subscriber)
# Start distributor only after subscriber is registered to avoid initial-loss race.
_ensure_distributor_running(channel, channel_key)
def _unsubscribe() -> None:
with channel.lock:
channel.subscribers.discard(subscriber)
utils/sstv/sstv_decoder.py
+58 -36
View File
@@ -552,15 +552,20 @@ class SSTVDecoder:
# Clean up if the thread exits while we thought we were running.
# This prevents a "ghost running" state where is_running is True
# but the thread has already died (e.g. rtl_fm exited).
orphan_proc = None
with self._lock:
was_running = self._running
self._running = False
if was_running and self._rtl_process:
with contextlib.suppress(Exception):
self._rtl_process.terminate()
self._rtl_process.wait(timeout=2)
orphan_proc = self._rtl_process
self._rtl_process = None
# Terminate outside lock to avoid blocking other operations
if orphan_proc:
with contextlib.suppress(Exception):
orphan_proc.terminate()
orphan_proc.wait(timeout=2)
if was_running:
logger.warning("Audio decode thread stopped unexpectedly")
err_detail = rtl_fm_error.split('\n')[-1] if rtl_fm_error else ''
@@ -661,38 +666,52 @@ class SSTVDecoder:
def _retune_rtl_fm(self, new_freq_hz: int) -> None:
"""Retune rtl_fm to a new frequency by restarting the process."""
old_proc = None
with self._lock:
if not self._running:
return
old_proc = self._rtl_process
self._rtl_process = None
if self._rtl_process:
try:
self._rtl_process.terminate()
self._rtl_process.wait(timeout=2)
except Exception:
with contextlib.suppress(Exception):
self._rtl_process.kill()
# Terminate old process outside lock
if old_proc:
try:
old_proc.terminate()
old_proc.wait(timeout=2)
except Exception:
with contextlib.suppress(Exception):
old_proc.kill()
rtl_cmd = [
'rtl_fm',
'-d', str(self._device_index),
'-f', str(new_freq_hz),
'-M', self._modulation,
'-s', str(SAMPLE_RATE),
'-r', str(SAMPLE_RATE),
'-l', '0',
'-'
]
# Build and start new process outside lock
rtl_cmd = [
'rtl_fm',
'-d', str(self._device_index),
'-f', str(new_freq_hz),
'-M', self._modulation,
'-s', str(SAMPLE_RATE),
'-r', str(SAMPLE_RATE),
'-l', '0',
'-'
]
logger.debug(f"Restarting rtl_fm: {' '.join(rtl_cmd)}")
logger.debug(f"Restarting rtl_fm: {' '.join(rtl_cmd)}")
self._rtl_process = subprocess.Popen(
rtl_cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
new_proc = subprocess.Popen(
rtl_cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
self._current_tuned_freq_hz = new_freq_hz
# Re-acquire lock to install new process
with self._lock:
if self._running:
self._rtl_process = new_proc
self._current_tuned_freq_hz = new_freq_hz
else:
# stop() was called during retune — clean up new process
with contextlib.suppress(Exception):
new_proc.terminate()
new_proc.wait(timeout=2)
@property
def last_doppler_info(self) -> DopplerInfo | None:
@@ -706,19 +725,22 @@ class SSTVDecoder:
def stop(self) -> None:
"""Stop SSTV decoder."""
proc_to_terminate = None
with self._lock:
self._running = False
proc_to_terminate = self._rtl_process
self._rtl_process = None
if self._rtl_process:
try:
self._rtl_process.terminate()
self._rtl_process.wait(timeout=5)
except Exception:
with contextlib.suppress(Exception):
self._rtl_process.kill()
self._rtl_process = None
# Terminate outside lock to avoid blocking other operations
if proc_to_terminate:
try:
proc_to_terminate.terminate()
proc_to_terminate.wait(timeout=5)
except Exception:
with contextlib.suppress(Exception):
proc_to_terminate.kill()
logger.info("SSTV decoder stopped")
logger.info("SSTV decoder stopped")
def get_images(self) -> list[SSTVImage]:
"""Get list of decoded images."""
utils/subghz.py
+2225 -2191
View File
File diff suppressed because it is too large Load Diff
utils/weather_sat.py
+163 -89
View File
@@ -85,7 +85,11 @@ WEATHER_SATELLITES = {
}
# Default sample rate for weather satellite reception
DEFAULT_SAMPLE_RATE = 1000000 # 1 MHz
try:
from config import WEATHER_SAT_SAMPLE_RATE as _configured_rate
DEFAULT_SAMPLE_RATE = _configured_rate
except ImportError:
DEFAULT_SAMPLE_RATE = 2400000 # 2.4 MHz — minimum for Meteor LRPT
@dataclass
@@ -169,7 +173,7 @@ class WeatherSatDecoder:
self._current_frequency: float = 0.0
self._current_mode: str = ''
self._capture_start_time: float = 0
self._device_index: int = 0
self._device_index: int = -1
self._capture_output_dir: Path | None = None
self._on_complete_callback: Callable[[], None] | None = None
self._capture_phase: str = 'idle'
@@ -237,7 +241,7 @@ class WeatherSatDecoder:
satellite: str,
input_file: str | Path,
sample_rate: int = DEFAULT_SAMPLE_RATE,
) -> bool:
) -> tuple[bool, str | None]:
"""Start weather satellite decode from a pre-recorded IQ/WAV file.
No SDR hardware is required — SatDump runs in offline mode.
@@ -248,28 +252,30 @@ class WeatherSatDecoder:
sample_rate: Sample rate of the recording in Hz
Returns:
True if started successfully
Tuple of (success, error_message). error_message is None on success.
"""
with self._lock:
if self._running:
return True
return True, None
if not self._decoder:
logger.error("No weather satellite decoder available")
msg = 'SatDump not installed. Build from source or install via package manager.'
self._emit_progress(CaptureProgress(
status='error',
message='SatDump not installed. Build from source or install via package manager.'
message=msg,
))
return False
return False, msg
sat_info = WEATHER_SATELLITES.get(satellite)
if not sat_info:
logger.error(f"Unknown satellite: {satellite}")
msg = f'Unknown satellite: {satellite}'
self._emit_progress(CaptureProgress(
status='error',
message=f'Unknown satellite: {satellite}'
message=msg,
))
return False
return False, msg
input_path = Path(input_file)
@@ -279,29 +285,33 @@ class WeatherSatDecoder:
resolved = input_path.resolve()
if not resolved.is_relative_to(allowed_base):
logger.warning(f"Path traversal blocked in start_from_file: {input_file}")
msg = 'Input file must be under the data/ directory'
self._emit_progress(CaptureProgress(
status='error',
message='Input file must be under the data/ directory'
message=msg,
))
return False
return False, msg
except (OSError, ValueError):
msg = 'Invalid file path'
self._emit_progress(CaptureProgress(
status='error',
message='Invalid file path'
message=msg,
))
return False
return False, msg
if not input_path.is_file():
logger.error(f"Input file not found: {input_file}")
msg = 'Input file not found'
self._emit_progress(CaptureProgress(
status='error',
message='Input file not found'
message=msg,
))
return False
return False, msg
self._current_satellite = satellite
self._current_frequency = sat_info['frequency']
self._current_mode = sat_info['mode']
self._device_index = -1 # Offline decode does not claim an SDR device
self._capture_start_time = time.time()
self._capture_phase = 'decoding'
self._stop_event.clear()
@@ -326,17 +336,18 @@ class WeatherSatDecoder:
capture_phase='decoding',
))
return True
return True, None
except Exception as e:
self._running = False
error_msg = str(e)
logger.error(f"Failed to start file decode: {e}")
self._emit_progress(CaptureProgress(
status='error',
satellite=satellite,
message=str(e)
message=error_msg,
))
return False
return False, error_msg
def start(
self,
@@ -345,7 +356,7 @@ class WeatherSatDecoder:
gain: float = 40.0,
sample_rate: int = DEFAULT_SAMPLE_RATE,
bias_t: bool = False,
) -> bool:
) -> tuple[bool, str | None]:
"""Start weather satellite capture and decode.
Args:
@@ -356,28 +367,35 @@ class WeatherSatDecoder:
bias_t: Enable bias-T power for LNA
Returns:
True if started successfully
Tuple of (success, error_message). error_message is None on success.
"""
# Validate satellite BEFORE acquiring the lock
sat_info = WEATHER_SATELLITES.get(satellite)
if not sat_info:
logger.error(f"Unknown satellite: {satellite}")
msg = f'Unknown satellite: {satellite}'
self._emit_progress(CaptureProgress(
status='error',
message=msg,
))
return False, msg
# Resolve device ID BEFORE lock — this runs rtl_test which can
# take up to 5s and has no side effects on instance state.
source_id = self._resolve_device_id(device_index)
with self._lock:
if self._running:
return True
return True, None
if not self._decoder:
logger.error("No weather satellite decoder available")
msg = 'SatDump not installed. Build from source or install via package manager.'
self._emit_progress(CaptureProgress(
status='error',
message='SatDump not installed. Build from source or install via package manager.'
message=msg,
))
return False
sat_info = WEATHER_SATELLITES.get(satellite)
if not sat_info:
logger.error(f"Unknown satellite: {satellite}")
self._emit_progress(CaptureProgress(
status='error',
message=f'Unknown satellite: {satellite}'
))
return False
return False, msg
self._current_satellite = satellite
self._current_frequency = sat_info['frequency']
@@ -389,7 +407,7 @@ class WeatherSatDecoder:
try:
self._running = True
self._start_satdump(sat_info, device_index, gain, sample_rate, bias_t)
self._start_satdump(sat_info, device_index, gain, sample_rate, bias_t, source_id)
logger.info(
f"Weather satellite capture started: {satellite} "
@@ -405,17 +423,18 @@ class WeatherSatDecoder:
capture_phase=self._capture_phase,
))
return True
return True, None
except Exception as e:
self._running = False
error_msg = str(e)
logger.error(f"Failed to start weather satellite capture: {e}")
self._emit_progress(CaptureProgress(
status='error',
satellite=satellite,
message=str(e)
message=error_msg,
))
return False
return False, error_msg
def _start_satdump(
self,
@@ -424,6 +443,7 @@ class WeatherSatDecoder:
gain: float,
sample_rate: int,
bias_t: bool,
source_id: str | None = None,
) -> None:
"""Start SatDump live capture and decode."""
# Create timestamped output directory for this capture
@@ -434,9 +454,9 @@ class WeatherSatDecoder:
freq_hz = int(sat_info['frequency'] * 1_000_000)
# SatDump v1.2+ uses string source_id (device serial) not numeric index.
# Auto-detect serial by querying rtl_eeprom, fall back to string index.
source_id = self._resolve_device_id(device_index)
# Use pre-resolved source_id, or fall back to resolving now
if source_id is None:
source_id = self._resolve_device_id(device_index)
cmd = [
'satdump', 'live',
@@ -446,9 +466,14 @@ class WeatherSatDecoder:
'--samplerate', str(sample_rate),
'--frequency', str(freq_hz),
'--gain', str(int(gain)),
'--source_id', source_id,
]
# Only pass --source_id if we have a real serial number.
# When _resolve_device_id returns None (no serial found),
# omit the flag so SatDump uses the first available device.
if source_id is not None:
cmd.extend(['--source_id', source_id])
if bias_t:
cmd.append('--bias')
@@ -468,36 +493,33 @@ class WeatherSatDecoder:
close_fds=True,
)
register_process(self._process)
os.close(slave_fd) # parent doesn't need the slave side
try:
os.close(slave_fd) # parent doesn't need the slave side
except OSError:
pass
# Check for early exit asynchronously (avoid blocking /start for 3s)
# Synchronous startup check — catch immediate failures (bad args,
# missing device) before returning to the caller.
time.sleep(0.5)
if self._process.poll() is not None:
error_output = self._drain_pty_output(master_fd)
if error_output:
logger.error(f"SatDump output:\n{error_output}")
error_msg = self._extract_error(error_output, self._process.returncode)
raise RuntimeError(error_msg)
# Backup async check for slower failures (e.g. device opens then
# fails after a second or two).
def _check_early_exit():
"""Poll once after 3s; if SatDump died, emit an error event."""
time.sleep(3)
"""Poll once after 2s; if SatDump died, emit an error event."""
time.sleep(2)
process = self._process
if process is None or process.poll() is None:
return # still running or already cleaned up
retcode = process.returncode
output = b''
try:
while True:
r, _, _ = select.select([master_fd], [], [], 0.1)
if not r:
break
chunk = os.read(master_fd, 4096)
if not chunk:
break
output += chunk
except OSError:
pass
output_str = output.decode('utf-8', errors='replace')
error_msg = f"SatDump exited immediately (code {retcode})"
if output_str:
for line in output_str.strip().splitlines():
if 'error' in line.lower() or 'could not' in line.lower() or 'cannot' in line.lower():
error_msg = line.strip()
break
logger.error(f"SatDump output:\n{output_str}")
error_output = self._drain_pty_output(master_fd)
if error_output:
logger.error(f"SatDump output:\n{error_output}")
error_msg = self._extract_error(error_output, process.returncode)
self._emit_progress(CaptureProgress(
status='error',
satellite=self._current_satellite,
@@ -568,11 +590,21 @@ class WeatherSatDecoder:
close_fds=True,
)
register_process(self._process)
os.close(slave_fd) # parent doesn't need the slave side
try:
os.close(slave_fd) # parent doesn't need the slave side
except OSError:
pass
# For offline mode, don't check for early exit — file decoding
# may complete very quickly and exit code 0 is normal success.
# The reader thread will handle output and detect errors.
# Synchronous startup check — catch immediate failures (bad args,
# missing pipeline). For offline mode, exit code 0 is normal success
# (file decoding can finish quickly), so only raise on non-zero.
time.sleep(0.5)
if self._process.poll() is not None and self._process.returncode != 0:
error_output = self._drain_pty_output(master_fd)
if error_output:
logger.error(f"SatDump offline output:\n{error_output}")
error_msg = self._extract_error(error_output, self._process.returncode)
raise RuntimeError(error_msg)
# Start reader thread to monitor output
self._reader_thread = threading.Thread(
@@ -605,12 +637,12 @@ class WeatherSatDecoder:
return 'info'
@staticmethod
def _resolve_device_id(device_index: int) -> str:
def _resolve_device_id(device_index: int) -> str | None:
"""Resolve RTL-SDR device index to serial number string for SatDump v1.2+.
SatDump v1.2+ expects --source_id as a device serial string, not a
numeric index. Try to look up the serial via rtl_test, fall back to
the string representation of the index.
numeric index. Try to look up the serial via rtl_test, return None
if no serial can be found (caller should omit --source_id).
"""
try:
result = subprocess.run(
@@ -636,8 +668,35 @@ class WeatherSatDecoder:
except (subprocess.TimeoutExpired, FileNotFoundError, OSError) as e:
logger.debug(f"Could not detect device serial: {e}")
# Fall back to string index
return str(device_index)
# No serial found — caller should omit --source_id
return None
@staticmethod
def _drain_pty_output(master_fd: int) -> str:
"""Read all available output from a PTY master fd."""
output = b''
try:
while True:
r, _, _ = select.select([master_fd], [], [], 0.1)
if not r:
break
chunk = os.read(master_fd, 4096)
if not chunk:
break
output += chunk
except OSError:
pass
return output.decode('utf-8', errors='replace')
@staticmethod
def _extract_error(output: str, returncode: int) -> str:
"""Extract a meaningful error message from SatDump output."""
if output:
for line in output.strip().splitlines():
lower = line.lower()
if 'error' in lower or 'could not' in lower or 'cannot' in lower or 'failed' in lower:
return line.strip()
return f"SatDump exited immediately (code {returncode})"
def _read_pty_lines(self):
"""Read lines from the PTY master fd, splitting on \\n and \\r.
@@ -801,20 +860,23 @@ class WeatherSatDecoder:
# Signal watcher thread to do final scan and exit
self._stop_event.set()
# Process ended — release resources
was_running = self._running
self._running = False
# Acquire lock when modifying shared state to avoid racing
# with stop() which may have already cleaned up.
with self._lock:
was_running = self._running
self._running = False
process = self._process
elapsed = int(time.time() - self._capture_start_time) if self._capture_start_time else 0
if was_running:
# Collect exit status (returncode is only set after poll/wait)
if self._process and self._process.returncode is None:
if process and process.returncode is None:
try:
self._process.wait(timeout=5)
process.wait(timeout=5)
except subprocess.TimeoutExpired:
self._process.kill()
self._process.wait()
retcode = self._process.returncode if self._process else None
process.kill()
process.wait()
retcode = process.returncode if process else None
if retcode and retcode != 0:
self._capture_phase = 'error'
self._emit_progress(CaptureProgress(
@@ -892,7 +954,15 @@ class WeatherSatDecoder:
product = self._parse_product_name(filepath)
# Copy image to main output dir for serving
serve_name = f"{self._current_satellite}_{filepath.stem}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
safe_sat = re.sub(r'[^A-Za-z0-9_-]+', '_', self._current_satellite).strip('_') or 'satellite'
safe_stem = re.sub(r'[^A-Za-z0-9_-]+', '_', filepath.stem).strip('_') or 'image'
suffix = filepath.suffix.lower()
if suffix not in ('.png', '.jpg', '.jpeg'):
suffix = '.png'
serve_name = (
f"{safe_sat}_{safe_stem}_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')}"
f"{suffix}"
)
serve_path = self._output_dir / serve_name
try:
shutil.copy2(filepath, serve_path)
@@ -944,7 +1014,7 @@ class WeatherSatDecoder:
if 'ndvi' in name:
return 'NDVI Vegetation'
if 'channel' in name or 'ch' in name:
match = re.search(r'(?:channel|ch)\s*(\d+)', name)
match = re.search(r'(?:channel|ch)[\s_-]*(\d+)', name)
if match:
return f'Channel {match.group(1)}'
if 'avhrr' in name:
@@ -967,13 +1037,16 @@ class WeatherSatDecoder:
self._running = False
self._stop_event.set()
self._close_pty()
if self._process:
safe_terminate(self._process)
self._process = None
process = self._process
self._process = None
elapsed = int(time.time() - self._capture_start_time) if self._capture_start_time else 0
logger.info(f"Weather satellite capture stopped after {elapsed}s")
self._device_index = -1
# Terminate outside the lock so stop() returns quickly
# and doesn't block start() or other lock acquisitions
if process:
safe_terminate(process)
def get_images(self) -> list[WeatherSatImage]:
"""Get list of decoded images."""
@@ -1020,6 +1093,7 @@ class WeatherSatDecoder:
product=self._parse_product_name(filepath),
)
self._images.append(image)
known_filenames.add(filepath.name)
def delete_image(self, filename: str) -> bool:
"""Delete a decoded image."""
utils/weather_sat_scheduler.py
+95 -78
View File
@@ -4,12 +4,12 @@ Automatically captures satellite passes based on predicted pass times.
Uses threading.Timer for scheduling — no external dependencies required.
"""
from __future__ import annotations
import threading
import uuid
from datetime import datetime, timezone, timedelta
from typing import Any, Callable
from __future__ import annotations
import threading
import uuid
from datetime import datetime, timezone, timedelta
from typing import Any, Callable
from utils.logging import get_logger
from utils.weather_sat import get_weather_sat_decoder, WEATHER_SATELLITES, CaptureProgress
@@ -21,13 +21,15 @@ try:
from config import (
WEATHER_SAT_SCHEDULE_REFRESH_MINUTES,
WEATHER_SAT_CAPTURE_BUFFER_SECONDS,
WEATHER_SAT_SAMPLE_RATE,
)
except ImportError:
WEATHER_SAT_SCHEDULE_REFRESH_MINUTES = 30
WEATHER_SAT_CAPTURE_BUFFER_SECONDS = 30
WEATHER_SAT_SAMPLE_RATE = 2400000
class ScheduledPass:
class ScheduledPass:
"""A pass scheduled for automatic capture."""
def __init__(self, pass_data: dict[str, Any]):
@@ -46,13 +48,13 @@ class ScheduledPass:
self._timer: threading.Timer | None = None
self._stop_timer: threading.Timer | None = None
@property
def start_dt(self) -> datetime:
return _parse_utc_iso(self.start_time)
@property
def end_dt(self) -> datetime:
return _parse_utc_iso(self.end_time)
@property
def start_dt(self) -> datetime:
return _parse_utc_iso(self.start_time)
@property
def end_dt(self) -> datetime:
return _parse_utc_iso(self.end_time)
def to_dict(self) -> dict[str, Any]:
return {
@@ -71,7 +73,7 @@ class ScheduledPass:
}
class WeatherSatScheduler:
class WeatherSatScheduler:
"""Auto-scheduler for weather satellite captures."""
def __init__(self):
@@ -200,10 +202,10 @@ class WeatherSatScheduler:
with self._lock:
return [p.to_dict() for p in self._passes]
def _refresh_passes(self) -> None:
"""Recompute passes and schedule timers."""
if not self._enabled:
return
def _refresh_passes(self) -> None:
"""Recompute passes and schedule timers."""
if not self._enabled:
return
try:
from utils.weather_sat_predict import predict_passes
@@ -227,39 +229,39 @@ class WeatherSatScheduler:
p._stop_timer.cancel()
# Keep completed/skipped for history, replace scheduled
history = [p for p in self._passes if p.status in ('complete', 'skipped', 'capturing')]
self._passes = history
now = datetime.now(timezone.utc)
buffer = WEATHER_SAT_CAPTURE_BUFFER_SECONDS
for pass_data in passes:
try:
sp = ScheduledPass(pass_data)
start_dt = sp.start_dt
end_dt = sp.end_dt
except Exception as e:
logger.warning(f"Skipping invalid pass data: {e}")
continue
capture_start = start_dt - timedelta(seconds=buffer)
capture_end = end_dt + timedelta(seconds=buffer)
# Skip passes that are already over
if capture_end <= now:
continue
# Check if already in history
if any(h.id == sp.id for h in history):
continue
# Schedule capture timer. If we're already inside the capture
# window, trigger immediately instead of skipping the pass.
delay = max(0.0, (capture_start - now).total_seconds())
sp._timer = threading.Timer(delay, self._execute_capture, args=[sp])
sp._timer.daemon = True
sp._timer.start()
self._passes.append(sp)
history = [p for p in self._passes if p.status in ('complete', 'skipped', 'capturing')]
self._passes = history
now = datetime.now(timezone.utc)
buffer = WEATHER_SAT_CAPTURE_BUFFER_SECONDS
for pass_data in passes:
try:
sp = ScheduledPass(pass_data)
start_dt = sp.start_dt
end_dt = sp.end_dt
except Exception as e:
logger.warning(f"Skipping invalid pass data: {e}")
continue
capture_start = start_dt - timedelta(seconds=buffer)
capture_end = end_dt + timedelta(seconds=buffer)
# Skip passes that are already over
if capture_end <= now:
continue
# Check if already in history
if any(h.id == sp.id for h in history):
continue
# Schedule capture timer. If we're already inside the capture
# window, trigger immediately instead of skipping the pass.
delay = max(0.0, (capture_start - now).total_seconds())
sp._timer = threading.Timer(delay, self._execute_capture, args=[sp])
sp._timer.daemon = True
sp._timer.start()
self._passes.append(sp)
logger.info(
f"Scheduler refreshed: {sum(1 for p in self._passes if p.status == 'scheduled')} "
@@ -320,16 +322,31 @@ class WeatherSatScheduler:
def _release_device():
try:
import app as app_module
owner = None
get_status = getattr(app_module, 'get_sdr_device_status', None)
if callable(get_status):
try:
owner = get_status().get(self._device)
except Exception:
owner = None
if owner and owner != 'weather_sat':
logger.debug(
"Skipping SDR release for device %s owned by %s",
self._device,
owner,
)
return
app_module.release_sdr_device(self._device)
except ImportError:
pass
decoder.set_on_complete(lambda: self._on_capture_complete(sp, _release_device))
success = decoder.start(
success, _error_msg = decoder.start(
satellite=sp.satellite,
device_index=self._device,
gain=self._gain,
sample_rate=WEATHER_SAT_SAMPLE_RATE,
bias_t=self._bias_t,
)
@@ -374,31 +391,31 @@ class WeatherSatScheduler:
def _emit_event(self, event: dict[str, Any]) -> None:
"""Emit scheduler event to callback."""
if self._event_callback:
try:
self._event_callback(event)
except Exception as e:
logger.error(f"Error in scheduler event callback: {e}")
def _parse_utc_iso(value: str) -> datetime:
"""Parse UTC ISO8601 timestamp robustly across Python versions."""
if not value:
raise ValueError("missing timestamp")
text = str(value).strip()
# Backward compatibility for malformed legacy strings.
text = text.replace('+00:00Z', 'Z')
# Python <3.11 does not accept trailing 'Z' in fromisoformat.
if text.endswith('Z'):
text = text[:-1] + '+00:00'
dt = datetime.fromisoformat(text)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
else:
dt = dt.astimezone(timezone.utc)
return dt
if self._event_callback:
try:
self._event_callback(event)
except Exception as e:
logger.error(f"Error in scheduler event callback: {e}")
def _parse_utc_iso(value: str) -> datetime:
"""Parse UTC ISO8601 timestamp robustly across Python versions."""
if not value:
raise ValueError("missing timestamp")
text = str(value).strip()
# Backward compatibility for malformed legacy strings.
text = text.replace('+00:00Z', 'Z')
# Python <3.11 does not accept trailing 'Z' in fromisoformat.
if text.endswith('Z'):
text = text[:-1] + '+00:00'
dt = datetime.fromisoformat(text)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
else:
dt = dt.astimezone(timezone.utc)
return dt
# Singleton
utils/wefax.py
+844
View File
@@ -0,0 +1,844 @@
"""WeFax (Weather Fax) decoder.
Decodes HF radiofax (weather fax) transmissions using any supported SDR
(RTL-SDR, HackRF, LimeSDR, Airspy, SDRPlay) via the SDRFactory
abstraction layer. The decoder implements the standard WeFax AM protocol:
carrier 1900 Hz, deviation +/-400 Hz (black=1500, white=2300).
Pipeline: rtl_fm/rx_fm -M usb -> stdout PCM -> Python DSP state machine
State machine: SCANNING -> PHASING -> RECEIVING -> COMPLETE
"""
from __future__ import annotations
import base64
import contextlib
import io
import math
import os
import select
import subprocess
import threading
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum
from pathlib import Path
from typing import Callable
import numpy as np
from utils.dependencies import get_tool_path
from utils.logging import get_logger
from utils.sdr import SDRFactory, SDRType
logger = get_logger('intercept.wefax')
try:
from PIL import Image as PILImage
except ImportError:
PILImage = None # type: ignore[assignment,misc]
# ---------------------------------------------------------------------------
# WeFax protocol constants
# ---------------------------------------------------------------------------
CARRIER_FREQ = 1900.0 # Hz - center/carrier
BLACK_FREQ = 1500.0 # Hz - black level
WHITE_FREQ = 2300.0 # Hz - white level
START_TONE_FREQ = 300.0 # Hz - start tone
STOP_TONE_FREQ = 450.0 # Hz - stop tone
PHASING_FREQ = WHITE_FREQ # White pulse during phasing
START_TONE_DURATION = 3.0 # Minimum seconds of start tone to detect
STOP_TONE_DURATION = 3.0 # Minimum seconds of stop tone to detect
PHASING_MIN_LINES = 5 # Minimum phasing lines before image
DEFAULT_SAMPLE_RATE = 22050
DEFAULT_IOC = 576
DEFAULT_LPM = 120
class DecoderState(Enum):
"""WeFax decoder state machine states."""
SCANNING = 'scanning'
START_DETECTED = 'start_detected'
PHASING = 'phasing'
RECEIVING = 'receiving'
COMPLETE = 'complete'
# ---------------------------------------------------------------------------
# Dataclasses
# ---------------------------------------------------------------------------
@dataclass
class WeFaxImage:
"""Decoded WeFax image metadata."""
filename: str
path: Path
station: str
frequency_khz: float
timestamp: datetime
ioc: int
lpm: int
size_bytes: int = 0
def to_dict(self) -> dict:
return {
'filename': self.filename,
'path': str(self.path),
'station': self.station,
'frequency_khz': self.frequency_khz,
'timestamp': self.timestamp.isoformat(),
'ioc': self.ioc,
'lpm': self.lpm,
'size_bytes': self.size_bytes,
'url': f'/wefax/images/{self.filename}',
}
@dataclass
class WeFaxProgress:
"""WeFax decode progress update for SSE streaming."""
status: str # 'scanning', 'phasing', 'receiving', 'complete', 'error', 'stopped'
station: str = ''
message: str = ''
progress_percent: int = 0
line_count: int = 0
image: WeFaxImage | None = None
partial_image: str | None = None
def to_dict(self) -> dict:
result: dict = {
'type': 'wefax_progress',
'status': self.status,
'progress': self.progress_percent,
}
if self.station:
result['station'] = self.station
if self.message:
result['message'] = self.message
if self.line_count:
result['line_count'] = self.line_count
if self.image:
result['image'] = self.image.to_dict()
if self.partial_image:
result['partial_image'] = self.partial_image
return result
# ---------------------------------------------------------------------------
# DSP helpers (reuse Goertzel from SSTV where sensible)
# ---------------------------------------------------------------------------
def _goertzel_mag(samples: np.ndarray, target_freq: float,
sample_rate: int) -> float:
"""Compute Goertzel magnitude at a single frequency."""
n = len(samples)
if n == 0:
return 0.0
w = 2.0 * math.pi * target_freq / sample_rate
coeff = 2.0 * math.cos(w)
s1 = 0.0
s2 = 0.0
for sample in samples:
s0 = float(sample) + coeff * s1 - s2
s2 = s1
s1 = s0
energy = s1 * s1 + s2 * s2 - coeff * s1 * s2
return math.sqrt(max(0.0, energy))
def _freq_to_pixel(frequency: float) -> int:
"""Map WeFax audio frequency to pixel value (0=black, 255=white).
Linear mapping: 1500 Hz -> 0 (black), 2300 Hz -> 255 (white).
"""
normalized = (frequency - BLACK_FREQ) / (WHITE_FREQ - BLACK_FREQ)
return max(0, min(255, int(normalized * 255 + 0.5)))
def _estimate_frequency(samples: np.ndarray, sample_rate: int,
freq_low: float = 1200.0,
freq_high: float = 2500.0) -> float:
"""Estimate dominant frequency using coarse+fine Goertzel sweep."""
if len(samples) == 0:
return 0.0
best_freq = freq_low
best_energy = 0.0
# Coarse sweep (25 Hz steps)
freq = freq_low
while freq <= freq_high:
energy = _goertzel_mag(samples, freq, sample_rate) ** 2
if energy > best_energy:
best_energy = energy
best_freq = freq
freq += 25.0
# Fine sweep around peak (+/- 25 Hz, 5 Hz steps)
fine_low = max(freq_low, best_freq - 25.0)
fine_high = min(freq_high, best_freq + 25.0)
freq = fine_low
while freq <= fine_high:
energy = _goertzel_mag(samples, freq, sample_rate) ** 2
if energy > best_energy:
best_energy = energy
best_freq = freq
freq += 5.0
return best_freq
def _detect_tone(samples: np.ndarray, target_freq: float,
sample_rate: int, threshold: float = 3.0) -> bool:
"""Detect if a specific tone dominates the signal."""
target_mag = _goertzel_mag(samples, target_freq, sample_rate)
# Check against a few reference frequencies
refs = [1000.0, 1500.0, 1900.0, 2300.0]
refs = [f for f in refs if abs(f - target_freq) > 100]
if not refs:
return target_mag > 0.01
avg_ref = sum(_goertzel_mag(samples, f, sample_rate) for f in refs) / len(refs)
if avg_ref <= 0:
return target_mag > 0.01
return target_mag / avg_ref >= threshold
# ---------------------------------------------------------------------------
# WeFaxDecoder
# ---------------------------------------------------------------------------
class WeFaxDecoder:
"""WeFax decoder singleton.
Manages SDR FM demod subprocess and decodes WeFax images using a
state machine that detects start/stop tones, phasing signals, and
demodulates image lines.
"""
def __init__(self) -> None:
self._sdr_process: subprocess.Popen | None = None
self._running = False
self._lock = threading.Lock()
self._callback: Callable[[dict], None] | None = None
self._last_scope_time: float = 0.0
self._output_dir = Path('instance/wefax_images')
self._images: list[WeFaxImage] = []
self._decode_thread: threading.Thread | None = None
# Current session parameters
self._station = ''
self._frequency_khz = 0.0
self._ioc = DEFAULT_IOC
self._lpm = DEFAULT_LPM
self._sample_rate = DEFAULT_SAMPLE_RATE
self._device_index = 0
self._gain = 40.0
self._direct_sampling = True
self._output_dir.mkdir(parents=True, exist_ok=True)
self._sdr_tool_name: str = 'rtl_fm'
self._last_error: str = ''
@property
def is_running(self) -> bool:
return self._running
@property
def last_error(self) -> str:
"""Last error message from a failed start() attempt."""
return self._last_error
def set_callback(self, callback: Callable[[dict], None]) -> None:
"""Set callback for progress updates (fed to SSE queue)."""
self._callback = callback
def start(
self,
frequency_khz: float,
station: str = '',
device_index: int = 0,
gain: float = 40.0,
ioc: int = DEFAULT_IOC,
lpm: int = DEFAULT_LPM,
direct_sampling: bool = True,
sdr_type: str = 'rtlsdr',
) -> bool:
"""Start WeFax decoder.
Args:
frequency_khz: Frequency in kHz (e.g. 4298 for NOJ).
station: Station callsign for metadata.
device_index: SDR device index.
gain: Receiver gain in dB.
ioc: Index of Cooperation (576 or 288).
lpm: Lines per minute (120 or 60).
direct_sampling: Enable RTL-SDR direct sampling for HF.
sdr_type: SDR hardware type (rtlsdr, hackrf, limesdr, airspy, sdrplay).
Returns:
True if started successfully.
"""
with self._lock:
if self._running:
return True
self._station = station
self._frequency_khz = frequency_khz
self._ioc = ioc
self._lpm = lpm
self._device_index = device_index
self._gain = gain
self._direct_sampling = direct_sampling
self._sdr_type = sdr_type
self._sample_rate = DEFAULT_SAMPLE_RATE
try:
self._running = True
self._last_error = ''
self._start_pipeline_spawn()
except Exception as e:
self._running = False
self._last_error = str(e)
logger.error(f"Failed to start WeFax decoder: {e}")
self._emit_progress(WeFaxProgress(
status='error',
message=str(e),
))
return False
# Health check sleep outside lock
try:
self._start_pipeline_health_check()
logger.info(
f"WeFax decoder started: {frequency_khz} kHz, "
f"station={station}, IOC={ioc}, LPM={lpm}"
)
return True
except Exception as e:
with self._lock:
self._running = False
self._last_error = str(e)
logger.error(f"Failed to start WeFax decoder: {e}")
self._emit_progress(WeFaxProgress(
status='error',
message=str(e),
))
return False
def _start_pipeline(self) -> None:
"""Start SDR FM demod subprocess in USB mode for WeFax."""
self._start_pipeline_spawn()
self._start_pipeline_health_check()
def _start_pipeline_spawn(self) -> None:
"""Spawn the SDR FM demod subprocess. Must hold self._lock."""
try:
sdr_type_enum = SDRType(self._sdr_type)
except ValueError:
sdr_type_enum = SDRType.RTL_SDR
# Validate that the required tool is available
if sdr_type_enum == SDRType.RTL_SDR:
if not get_tool_path('rtl_fm'):
raise RuntimeError('rtl_fm not found')
else:
if not get_tool_path('rx_fm'):
raise RuntimeError('rx_fm not found (required for non-RTL-SDR devices)')
sdr_device = SDRFactory.create_default_device(
sdr_type_enum, index=self._device_index)
builder = SDRFactory.get_builder(sdr_type_enum)
rtl_cmd = builder.build_fm_demod_command(
device=sdr_device,
frequency_mhz=self._frequency_khz / 1000.0,
sample_rate=self._sample_rate,
gain=self._gain,
modulation='usb',
)
# RTL-SDR: append direct sampling flag for HF reception
if sdr_type_enum == SDRType.RTL_SDR and self._direct_sampling:
# Insert before trailing '-' stdout marker
if rtl_cmd and rtl_cmd[-1] == '-':
rtl_cmd.insert(-1, '-E')
rtl_cmd.insert(-1, 'direct2')
else:
rtl_cmd.extend(['-E', 'direct2', '-'])
self._sdr_tool_name = rtl_cmd[0] if rtl_cmd else 'sdr'
logger.info(f"Starting {self._sdr_tool_name}: {' '.join(rtl_cmd)}")
self._sdr_process = subprocess.Popen(
rtl_cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
def _start_pipeline_health_check(self) -> None:
"""Post-spawn health check and decode thread start. Called outside lock."""
time.sleep(0.3)
with self._lock:
if self._sdr_process and self._sdr_process.poll() is not None:
stderr_detail = ''
if self._sdr_process.stderr:
stderr_detail = self._sdr_process.stderr.read().decode(
errors='replace').strip()
rc = self._sdr_process.returncode
self._sdr_process = None
detail = stderr_detail.split('\n')[-1] if stderr_detail else f'exit code {rc}'
raise RuntimeError(f'{self._sdr_tool_name} failed: {detail}')
self._decode_thread = threading.Thread(
target=self._decode_audio_stream, daemon=True)
self._decode_thread.start()
def _decode_audio_stream(self) -> None:
"""Read audio from SDR FM demod and decode WeFax images.
Runs in a background thread. Processes 100ms chunks through
the start-tone / phasing / image state machine.
"""
sr = self._sample_rate
chunk_samples = sr // 10 # 100ms
chunk_bytes = chunk_samples * 2 # int16
state = DecoderState.SCANNING
start_tone_count = 0
stop_tone_count = 0
phasing_line_count = 0
# Image parameters
pixels_per_line = int(math.pi * self._ioc)
line_duration_s = 60.0 / self._lpm
samples_per_line = int(line_duration_s * sr)
# Image buffer
image_lines: list[np.ndarray] = []
line_buffer = np.zeros(0, dtype=np.float64)
max_lines = 2000 # Safety limit
sdr_error = ''
last_partial_line = -1
logger.info(
f"WeFax decode thread started: IOC={self._ioc}, "
f"LPM={self._lpm}, pixels/line={pixels_per_line}, "
f"samples/line={samples_per_line}"
)
# Emit initial scanning progress here (not in start()) so the
# frontend SSE connection is established before this event fires.
time.sleep(0.1)
self._emit_progress(WeFaxProgress(
status='scanning',
station=self._station,
message=f'Scanning {self._frequency_khz} kHz for WeFax start tone...',
))
while self._running and self._sdr_process:
try:
proc = self._sdr_process
if not proc or not proc.stdout:
break
# Non-blocking read via select() — allows checking _running
# on timeout instead of blocking indefinitely in read().
fd = proc.stdout.fileno()
ready, _, _ = select.select([fd], [], [], 0.5)
if not ready:
if not self._running:
break
continue
raw_data = os.read(fd, chunk_bytes)
if not raw_data:
if self._running:
stderr_msg = ''
if self._sdr_process and self._sdr_process.stderr:
with contextlib.suppress(Exception):
stderr_msg = self._sdr_process.stderr.read().decode(
errors='replace').strip()
rc = self._sdr_process.poll() if self._sdr_process else None
logger.warning(f"{self._sdr_tool_name} stream ended (exit code: {rc})")
if stderr_msg:
logger.warning(f"{self._sdr_tool_name} stderr: {stderr_msg}")
sdr_error = stderr_msg
break
n_samples = len(raw_data) // 2
if n_samples == 0:
continue
raw_int16 = np.frombuffer(raw_data[:n_samples * 2], dtype=np.int16)
samples = raw_int16.astype(np.float64) / 32768.0
# Emit scope waveform for frontend visualisation
self._emit_scope(raw_int16)
if state == DecoderState.SCANNING:
# Look for 300 Hz start tone
if _detect_tone(samples, START_TONE_FREQ, sr, threshold=2.5):
start_tone_count += 1
# Need sustained detection (>= START_TONE_DURATION seconds)
needed = int(START_TONE_DURATION / 0.1)
if start_tone_count >= needed:
state = DecoderState.PHASING
phasing_line_count = 0
logger.info("WeFax start tone detected, entering phasing")
self._emit_progress(WeFaxProgress(
status='phasing',
station=self._station,
message='Start tone detected, synchronising...',
))
else:
start_tone_count = max(0, start_tone_count - 1)
elif state == DecoderState.PHASING:
# Count phasing lines (alternating black/white pulses)
phasing_line_count += 1
needed_phasing = max(PHASING_MIN_LINES, int(2.0 / 0.1))
if phasing_line_count >= needed_phasing:
state = DecoderState.RECEIVING
image_lines = []
line_buffer = np.zeros(0, dtype=np.float64)
last_partial_line = -1
logger.info("Phasing complete, receiving image")
self._emit_progress(WeFaxProgress(
status='receiving',
station=self._station,
message='Receiving image...',
))
elif state == DecoderState.RECEIVING:
# Check for stop tone
if _detect_tone(samples, STOP_TONE_FREQ, sr, threshold=2.5):
stop_tone_count += 1
needed_stop = int(STOP_TONE_DURATION / 0.1)
if stop_tone_count >= needed_stop:
# Process any remaining line buffer
if len(line_buffer) >= samples_per_line * 0.5:
line_pixels = self._decode_line(
line_buffer, pixels_per_line, sr)
image_lines.append(line_pixels)
state = DecoderState.COMPLETE
logger.info(
f"Stop tone detected, image complete: "
f"{len(image_lines)} lines"
)
break
else:
stop_tone_count = max(0, stop_tone_count - 1)
# Accumulate samples into line buffer
line_buffer = np.concatenate([line_buffer, samples])
# Extract complete lines
while len(line_buffer) >= samples_per_line:
line_samples = line_buffer[:samples_per_line]
line_buffer = line_buffer[samples_per_line:]
line_pixels = self._decode_line(
line_samples, pixels_per_line, sr)
image_lines.append(line_pixels)
# Safety limit
if len(image_lines) >= max_lines:
logger.warning("WeFax max lines reached, saving image")
state = DecoderState.COMPLETE
break
# Emit progress periodically
current_lines = len(image_lines)
if current_lines > 0 and current_lines != last_partial_line and current_lines % 20 == 0:
last_partial_line = current_lines
# Rough progress estimate (typical chart ~800 lines)
pct = min(95, int(current_lines / 8))
partial_url = self._encode_partial(
image_lines, pixels_per_line)
self._emit_progress(WeFaxProgress(
status='receiving',
station=self._station,
message=f'Receiving: {current_lines} lines',
progress_percent=pct,
line_count=current_lines,
partial_image=partial_url,
))
except Exception as e:
logger.error(f"Error in WeFax decode thread: {e}")
if not self._running:
break
time.sleep(0.1)
# Save image if we got data
if state == DecoderState.COMPLETE and image_lines:
self._save_image(image_lines, pixels_per_line)
elif state == DecoderState.RECEIVING and len(image_lines) > 20:
# Save partial image if we had significant data
logger.info(f"Saving partial WeFax image: {len(image_lines)} lines")
self._save_image(image_lines, pixels_per_line)
# Clean up
with self._lock:
was_running = self._running
self._running = False
if self._sdr_process:
with contextlib.suppress(Exception):
self._sdr_process.terminate()
self._sdr_process.wait(timeout=2)
self._sdr_process = None
if was_running:
err_detail = sdr_error.split('\n')[-1] if sdr_error else ''
if state != DecoderState.COMPLETE:
msg = f'{self._sdr_tool_name} failed: {err_detail}' if err_detail else 'Decode stopped unexpectedly'
self._emit_progress(WeFaxProgress(
status='error', message=msg))
else:
self._emit_progress(WeFaxProgress(
status='stopped', message='Decoder stopped'))
logger.info("WeFax decode thread ended")
def _decode_line(self, line_samples: np.ndarray,
pixels_per_line: int, sample_rate: int) -> np.ndarray:
"""Decode one scan line from audio samples to pixel values.
Uses instantaneous frequency estimation via the analytic signal
(Hilbert transform), then maps frequency to grayscale.
"""
n = len(line_samples)
pixels = np.zeros(pixels_per_line, dtype=np.uint8)
if n < pixels_per_line:
return pixels
samples_per_pixel = n / pixels_per_line
# Use Hilbert transform for instantaneous frequency
try:
analytic = np.fft.ifft(
np.fft.fft(line_samples) * 2 * (np.arange(n) < n // 2))
inst_phase = np.unwrap(np.angle(analytic))
inst_freq = np.diff(inst_phase) / (2.0 * math.pi) * sample_rate
inst_freq = np.clip(inst_freq, BLACK_FREQ - 200, WHITE_FREQ + 200)
# Average frequency per pixel
for px in range(pixels_per_line):
start_idx = int(px * samples_per_pixel)
end_idx = int((px + 1) * samples_per_pixel)
end_idx = min(end_idx, len(inst_freq))
if start_idx >= end_idx:
continue
avg_freq = float(np.mean(inst_freq[start_idx:end_idx]))
pixels[px] = _freq_to_pixel(avg_freq)
except Exception:
# Fallback: simple Goertzel per pixel window
for px in range(pixels_per_line):
start_idx = int(px * samples_per_pixel)
end_idx = int((px + 1) * samples_per_pixel)
if start_idx >= len(line_samples) or start_idx >= end_idx:
break
window = line_samples[start_idx:end_idx]
freq = _estimate_frequency(window, sample_rate,
BLACK_FREQ - 200, WHITE_FREQ + 200)
pixels[px] = _freq_to_pixel(freq)
return pixels
def _encode_partial(self, image_lines: list[np.ndarray],
width: int) -> str | None:
"""Encode current image lines as a JPEG data URL for live preview."""
if PILImage is None or not image_lines:
return None
try:
height = len(image_lines)
img_array = np.zeros((height, width), dtype=np.uint8)
for i, line in enumerate(image_lines):
img_array[i, :len(line)] = line[:width]
img = PILImage.fromarray(img_array, mode='L')
buf = io.BytesIO()
img.save(buf, format='JPEG', quality=40)
b64 = base64.b64encode(buf.getvalue()).decode('ascii')
return f'data:image/jpeg;base64,{b64}'
except Exception:
return None
def _save_image(self, image_lines: list[np.ndarray],
width: int) -> None:
"""Save completed image to disk."""
if PILImage is None:
logger.error("Cannot save image: Pillow not installed")
self._emit_progress(WeFaxProgress(
status='error',
message='Cannot save image - Pillow not installed',
))
return
try:
height = len(image_lines)
img_array = np.zeros((height, width), dtype=np.uint8)
for i, line in enumerate(image_lines):
img_array[i, :len(line)] = line[:width]
img = PILImage.fromarray(img_array, mode='L')
timestamp = datetime.now(timezone.utc)
station_tag = self._station or 'unknown'
filename = f"wefax_{timestamp.strftime('%Y%m%d_%H%M%S')}_{station_tag}.png"
filepath = self._output_dir / filename
img.save(filepath, 'PNG')
wefax_image = WeFaxImage(
filename=filename,
path=filepath,
station=self._station,
frequency_khz=self._frequency_khz,
timestamp=timestamp,
ioc=self._ioc,
lpm=self._lpm,
size_bytes=filepath.stat().st_size,
)
self._images.append(wefax_image)
logger.info(f"WeFax image saved: {filename} ({wefax_image.size_bytes} bytes)")
self._emit_progress(WeFaxProgress(
status='complete',
station=self._station,
message=f'Image decoded: {height} lines',
progress_percent=100,
line_count=height,
image=wefax_image,
))
except Exception as e:
logger.error(f"Error saving WeFax image: {e}")
self._emit_progress(WeFaxProgress(
status='error',
message=f'Error saving image: {e}',
))
def stop(self) -> None:
"""Stop WeFax decoder.
Sets _running=False and terminates the process outside the lock,
then waits briefly for the decode thread to finish saving any
partial image before returning.
"""
with self._lock:
self._running = False
proc = self._sdr_process
self._sdr_process = None
thread = self._decode_thread
if proc:
with contextlib.suppress(Exception):
proc.terminate()
# Wait for the decode thread to save any partial image.
# With select()-based reads the thread exits within ~0.5s.
if thread:
with contextlib.suppress(Exception):
thread.join(timeout=2)
logger.info("WeFax decoder stopped")
def get_images(self) -> list[WeFaxImage]:
"""Get list of decoded images."""
self._scan_images()
return list(self._images)
def delete_image(self, filename: str) -> bool:
"""Delete a single decoded image."""
filepath = self._output_dir / filename
if not filepath.exists():
return False
filepath.unlink()
self._images = [img for img in self._images if img.filename != filename]
logger.info(f"Deleted WeFax image: {filename}")
return True
def delete_all_images(self) -> int:
"""Delete all decoded images. Returns count deleted."""
count = 0
for filepath in self._output_dir.glob('*.png'):
filepath.unlink()
count += 1
self._images.clear()
logger.info(f"Deleted all WeFax images ({count} files)")
return count
def _scan_images(self) -> None:
"""Scan output directory for images not yet tracked."""
known = {img.filename for img in self._images}
for filepath in self._output_dir.glob('*.png'):
if filepath.name not in known:
try:
stat = filepath.stat()
image = WeFaxImage(
filename=filepath.name,
path=filepath,
station='',
frequency_khz=0,
timestamp=datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc),
ioc=self._ioc,
lpm=self._lpm,
size_bytes=stat.st_size,
)
self._images.append(image)
except Exception as e:
logger.warning(f"Error scanning image {filepath}: {e}")
def _emit_progress(self, progress: WeFaxProgress) -> None:
"""Emit progress update to callback."""
if self._callback:
try:
self._callback(progress.to_dict())
except Exception as e:
logger.error(f"Error in progress callback: {e}")
def _emit_scope(self, raw_int16: np.ndarray) -> None:
"""Emit scope waveform data for frontend visualisation."""
if not self._callback:
return
now = time.monotonic()
if now - self._last_scope_time < 0.1:
return
self._last_scope_time = now
try:
peak = int(np.max(np.abs(raw_int16)))
rms = int(np.sqrt(np.mean(raw_int16.astype(np.float64) ** 2)))
# Downsample to 256 signed int8 values for lightweight transport
window = raw_int16[-256:] if len(raw_int16) > 256 else raw_int16
waveform = np.clip(window // 256, -127, 127).astype(np.int8).tolist()
self._callback({
'type': 'scope',
'rms': rms,
'peak': peak,
'waveform': waveform,
})
except Exception:
pass
# ---------------------------------------------------------------------------
# Module-level singleton
# ---------------------------------------------------------------------------
_decoder: WeFaxDecoder | None = None
def get_wefax_decoder() -> WeFaxDecoder:
"""Get or create the global WeFax decoder instance."""
global _decoder
if _decoder is None:
_decoder = WeFaxDecoder()
return _decoder
utils/wefax_scheduler.py
+543
View File
@@ -0,0 +1,543 @@
"""WeFax auto-capture scheduler.
Automatically captures WeFax broadcasts based on station broadcast schedules.
Uses threading.Timer for scheduling — no external dependencies required.
Unlike the weather satellite scheduler which uses TLE-based orbital prediction,
WeFax stations broadcast on fixed UTC schedules, making scheduling simpler.
"""
from __future__ import annotations
import threading
import uuid
from datetime import datetime, timedelta, timezone
from typing import Any, Callable
from utils.logging import get_logger
from utils.wefax import get_wefax_decoder
from utils.wefax_stations import get_station
logger = get_logger('intercept.wefax_scheduler')
# Import config defaults
try:
from config import (
WEFAX_CAPTURE_BUFFER_SECONDS,
WEFAX_SCHEDULE_REFRESH_MINUTES,
)
except ImportError:
WEFAX_SCHEDULE_REFRESH_MINUTES = 30
WEFAX_CAPTURE_BUFFER_SECONDS = 30
class ScheduledBroadcast:
"""A broadcast scheduled for automatic capture."""
def __init__(
self,
station: str,
callsign: str,
frequency_khz: float,
utc_time: str,
duration_min: int,
content: str,
occurrence_date: str = '',
):
self.id: str = str(uuid.uuid4())[:8]
self.station = station
self.callsign = callsign
self.frequency_khz = frequency_khz
self.utc_time = utc_time
self.duration_min = duration_min
self.content = content
self.occurrence_date = occurrence_date
self.status: str = 'scheduled' # scheduled, capturing, complete, skipped
self._timer: threading.Timer | None = None
self._stop_timer: threading.Timer | None = None
def to_dict(self) -> dict[str, Any]:
return {
'id': self.id,
'station': self.station,
'callsign': self.callsign,
'frequency_khz': self.frequency_khz,
'utc_time': self.utc_time,
'duration_min': self.duration_min,
'content': self.content,
'occurrence_date': self.occurrence_date,
'status': self.status,
}
class WeFaxScheduler:
"""Auto-scheduler for WeFax broadcast captures."""
def __init__(self):
self._enabled = False
self._lock = threading.Lock()
self._broadcasts: list[ScheduledBroadcast] = []
self._refresh_timer: threading.Timer | None = None
self._station: str = ''
self._callsign: str = ''
self._frequency_khz: float = 0.0
self._device: int = 0
self._gain: float = 40.0
self._ioc: int = 576
self._lpm: int = 120
self._direct_sampling: bool = True
self._progress_callback: Callable[[dict], None] | None = None
self._event_callback: Callable[[dict[str, Any]], None] | None = None
@property
def enabled(self) -> bool:
return self._enabled
def set_callbacks(
self,
progress_callback: Callable[[dict], None],
event_callback: Callable[[dict[str, Any]], None],
) -> None:
"""Set callbacks for progress and scheduler events."""
self._progress_callback = progress_callback
self._event_callback = event_callback
def enable(
self,
station: str,
frequency_khz: float,
device: int = 0,
gain: float = 40.0,
ioc: int = 576,
lpm: int = 120,
direct_sampling: bool = True,
) -> dict[str, Any]:
"""Enable auto-scheduling for a station/frequency.
Args:
station: Station callsign.
frequency_khz: Frequency in kHz.
device: RTL-SDR device index.
gain: SDR gain in dB.
ioc: Index of Cooperation (576 or 288).
lpm: Lines per minute (120 or 60).
direct_sampling: Enable direct sampling for HF.
Returns:
Status dict with scheduled broadcasts.
"""
station_data = get_station(station)
if not station_data:
return {'status': 'error', 'message': f'Station {station} not found'}
with self._lock:
self._station = station_data.get('name', station)
self._callsign = station
self._frequency_khz = frequency_khz
self._device = device
self._gain = gain
self._ioc = ioc
self._lpm = lpm
self._direct_sampling = direct_sampling
self._enabled = True
self._refresh_schedule()
return self.get_status()
def disable(self) -> dict[str, Any]:
"""Disable auto-scheduling and cancel all timers."""
with self._lock:
self._enabled = False
# Cancel refresh timer
if self._refresh_timer:
self._refresh_timer.cancel()
self._refresh_timer = None
# Cancel all broadcast timers
for b in self._broadcasts:
if b._timer:
b._timer.cancel()
b._timer = None
if b._stop_timer:
b._stop_timer.cancel()
b._stop_timer = None
self._broadcasts.clear()
logger.info("WeFax auto-scheduler disabled")
return {'status': 'disabled'}
def skip_broadcast(self, broadcast_id: str) -> bool:
"""Manually skip a scheduled broadcast."""
with self._lock:
for b in self._broadcasts:
if b.id == broadcast_id and b.status == 'scheduled':
b.status = 'skipped'
if b._timer:
b._timer.cancel()
b._timer = None
logger.info(
"Skipped broadcast: %s at %s", b.content, b.utc_time
)
self._emit_event({
'type': 'schedule_capture_skipped',
'broadcast': b.to_dict(),
'reason': 'manual',
})
return True
return False
def get_status(self) -> dict[str, Any]:
"""Get current scheduler status."""
with self._lock:
return {
'enabled': self._enabled,
'station': self._station,
'callsign': self._callsign,
'frequency_khz': self._frequency_khz,
'device': self._device,
'gain': self._gain,
'ioc': self._ioc,
'lpm': self._lpm,
'scheduled_count': sum(
1 for b in self._broadcasts if b.status == 'scheduled'
),
'total_broadcasts': len(self._broadcasts),
}
def get_broadcasts(self) -> list[dict[str, Any]]:
"""Get list of scheduled broadcasts."""
with self._lock:
return [b.to_dict() for b in self._broadcasts]
@staticmethod
def _history_key(callsign: str, utc_time: str, occurrence_date: str) -> str:
"""Build a stable key for one station UTC slot on one calendar day."""
return f'{callsign}_{utc_time}_{occurrence_date}'
def _refresh_schedule(self) -> None:
"""Recompute broadcast schedule and set timers."""
if not self._enabled:
return
station_data = get_station(self._callsign)
if not station_data:
logger.error("Station %s not found during refresh", self._callsign)
return
schedule = station_data.get('schedule', [])
with self._lock:
# Cancel existing timers
for b in self._broadcasts:
if b._timer:
b._timer.cancel()
if b._stop_timer:
b._stop_timer.cancel()
# Keep completed/skipped for history, replace scheduled
history = [
b for b in self._broadcasts
if b.status in ('complete', 'skipped', 'capturing')
]
self._broadcasts = history
now = datetime.now(timezone.utc)
buffer = WEFAX_CAPTURE_BUFFER_SECONDS
for entry in schedule:
utc_time = entry.get('utc', '')
duration_min = entry.get('duration_min', 20)
content = entry.get('content', '')
parts = utc_time.split(':')
if len(parts) != 2:
continue
try:
hour = int(parts[0])
minute = int(parts[1])
except ValueError:
continue
# Compute next occurrence (today or tomorrow)
broadcast_dt = now.replace(
hour=hour, minute=minute, second=0, microsecond=0
)
capture_end = broadcast_dt + timedelta(
minutes=duration_min, seconds=buffer
)
# If the broadcast end is already past, schedule for tomorrow
if capture_end <= now:
broadcast_dt += timedelta(days=1)
capture_end = broadcast_dt + timedelta(
minutes=duration_min, seconds=buffer
)
capture_start = broadcast_dt - timedelta(seconds=buffer)
occurrence_date = broadcast_dt.date().isoformat()
# Check if this specific day/slot was already processed.
history_key = self._history_key(
self._callsign,
utc_time,
occurrence_date,
)
if any(
self._history_key(
h.callsign,
h.utc_time,
getattr(h, 'occurrence_date', ''),
) == history_key
for h in history
):
continue
sb = ScheduledBroadcast(
station=self._station,
callsign=self._callsign,
frequency_khz=self._frequency_khz,
utc_time=utc_time,
duration_min=duration_min,
content=content,
occurrence_date=occurrence_date,
)
# Schedule capture timer
delay = max(0.0, (capture_start - now).total_seconds())
sb._timer = threading.Timer(
delay, self._execute_capture, args=[sb]
)
sb._timer.daemon = True
sb._timer.start()
logger.info(
"Scheduled capture: %s at %s UTC (fires in %.0fs)",
content, utc_time, delay,
)
self._broadcasts.append(sb)
logger.info(
"WeFax scheduler refreshed: %d broadcasts scheduled",
sum(1 for b in self._broadcasts if b.status == 'scheduled'),
)
# Schedule next refresh
if self._refresh_timer:
self._refresh_timer.cancel()
self._refresh_timer = threading.Timer(
WEFAX_SCHEDULE_REFRESH_MINUTES * 60,
self._refresh_schedule,
)
self._refresh_timer.daemon = True
self._refresh_timer.start()
def _execute_capture(self, sb: ScheduledBroadcast) -> None:
"""Execute capture for a scheduled broadcast (with error guard)."""
logger.info("Timer fired for broadcast: %s at %s", sb.content, sb.utc_time)
try:
self._execute_capture_inner(sb)
except Exception:
logger.exception(
"Unhandled exception in scheduled capture: %s at %s",
sb.content, sb.utc_time,
)
sb.status = 'skipped'
self._emit_event({
'type': 'schedule_capture_skipped',
'broadcast': sb.to_dict(),
'reason': 'error',
'detail': 'internal error — see server logs',
})
def _execute_capture_inner(self, sb: ScheduledBroadcast) -> None:
"""Execute capture for a scheduled broadcast."""
if not self._enabled or sb.status != 'scheduled':
return
decoder = get_wefax_decoder()
if decoder.is_running:
logger.info("Decoder busy, skipping scheduled broadcast: %s", sb.content)
sb.status = 'skipped'
self._emit_event({
'type': 'schedule_capture_skipped',
'broadcast': sb.to_dict(),
'reason': 'decoder_busy',
})
return
# Claim SDR device
try:
import app as app_module
error = app_module.claim_sdr_device(self._device, 'wefax')
if error:
logger.info(
"SDR device busy, skipping: %s - %s", sb.content, error
)
sb.status = 'skipped'
self._emit_event({
'type': 'schedule_capture_skipped',
'broadcast': sb.to_dict(),
'reason': 'device_busy',
})
return
except ImportError:
pass
sb.status = 'capturing'
def _release_device():
try:
import app as app_module
app_module.release_sdr_device(self._device)
except ImportError:
pass
released = False
release_lock = threading.Lock()
def _release_device_once() -> None:
nonlocal released
with release_lock:
if released:
return
released = True
_release_device()
def _scheduler_progress_callback(progress: dict) -> None:
"""Forward progress updates and release scheduler resources on terminal states."""
if self._progress_callback:
self._progress_callback(progress)
if not isinstance(progress, dict) or progress.get('type') != 'wefax_progress':
return
status = progress.get('status')
if status not in ('complete', 'error', 'stopped'):
return
if sb.status == 'capturing':
if status == 'complete':
sb.status = 'complete'
self._emit_event({
'type': 'schedule_capture_complete',
'broadcast': sb.to_dict(),
})
else:
sb.status = 'skipped'
self._emit_event({
'type': 'schedule_capture_skipped',
'broadcast': sb.to_dict(),
'reason': 'decoder_error',
'detail': progress.get('message', ''),
})
_release_device_once()
decoder.set_callback(_scheduler_progress_callback)
success = decoder.start(
frequency_khz=self._frequency_khz,
station=self._callsign,
device_index=self._device,
gain=self._gain,
ioc=self._ioc,
lpm=self._lpm,
direct_sampling=self._direct_sampling,
)
if success:
logger.info("Auto-scheduler started capture: %s", sb.content)
self._emit_event({
'type': 'schedule_capture_start',
'broadcast': sb.to_dict(),
})
# Schedule stop timer at broadcast end + buffer
now = datetime.now(timezone.utc)
parts = sb.utc_time.split(':')
broadcast_dt = now.replace(
hour=int(parts[0]), minute=int(parts[1]),
second=0, microsecond=0,
)
if broadcast_dt < now - timedelta(hours=1):
broadcast_dt += timedelta(days=1)
stop_dt = broadcast_dt + timedelta(
minutes=sb.duration_min,
seconds=WEFAX_CAPTURE_BUFFER_SECONDS,
)
stop_delay = max(0.0, (stop_dt - now).total_seconds())
if stop_delay > 0:
sb._stop_timer = threading.Timer(
stop_delay, self._stop_capture, args=[sb, _release_device_once]
)
sb._stop_timer.daemon = True
sb._stop_timer.start()
else:
# If execution was delayed beyond end-of-window, close out
# immediately so SDR allocation is never stranded.
logger.warning(
"Capture window already elapsed for %s at %s UTC; stopping immediately",
sb.content,
sb.utc_time,
)
self._stop_capture(sb, _release_device_once)
else:
sb.status = 'skipped'
_release_device_once()
self._emit_event({
'type': 'schedule_capture_skipped',
'broadcast': sb.to_dict(),
'reason': 'start_failed',
'detail': decoder.last_error or 'unknown error',
})
def _stop_capture(
self, sb: ScheduledBroadcast, release_fn: Callable
) -> None:
"""Stop capture at broadcast end."""
if sb.status != 'capturing':
release_fn()
return
sb.status = 'complete'
decoder = get_wefax_decoder()
if decoder.is_running:
decoder.stop()
logger.info("Auto-scheduler stopped capture: %s", sb.content)
release_fn()
self._emit_event({
'type': 'schedule_capture_complete',
'broadcast': sb.to_dict(),
})
def _emit_event(self, event: dict[str, Any]) -> None:
"""Emit scheduler event to callback."""
if self._event_callback:
try:
self._event_callback(event)
except Exception as e:
logger.error("Error in scheduler event callback: %s", e)
# Singleton
_scheduler: WeFaxScheduler | None = None
_scheduler_lock = threading.Lock()
def get_wefax_scheduler() -> WeFaxScheduler:
"""Get or create the global WeFax scheduler instance."""
global _scheduler
if _scheduler is None:
with _scheduler_lock:
if _scheduler is None:
_scheduler = WeFaxScheduler()
return _scheduler
utils/wefax_stations.py
+160
View File
@@ -0,0 +1,160 @@
"""WeFax station database loader.
Loads and caches station data from data/wefax_stations.json. Provides
lookup by callsign and current-broadcast filtering based on UTC time.
"""
from __future__ import annotations
import json
from datetime import datetime, timezone
from pathlib import Path
_stations_cache: list[dict] | None = None
_stations_by_callsign: dict[str, dict] = {}
_VALID_FREQUENCY_REFERENCES = {'auto', 'carrier', 'dial'}
WEFAX_USB_ALIGNMENT_OFFSET_KHZ = 1.9
_STATIONS_PATH = Path(__file__).resolve().parent.parent / 'data' / 'wefax_stations.json'
def load_stations() -> list[dict]:
"""Load all WeFax stations from JSON, caching on first call."""
global _stations_cache, _stations_by_callsign
if _stations_cache is not None:
return _stations_cache
with open(_STATIONS_PATH) as f:
data = json.load(f)
_stations_cache = data.get('stations', [])
_stations_by_callsign = {s['callsign']: s for s in _stations_cache}
return _stations_cache
def get_station(callsign: str) -> dict | None:
"""Get a single station by callsign."""
load_stations()
return _stations_by_callsign.get(callsign.upper())
def _normalize_frequency_reference(value: str | None) -> str:
"""Normalize and validate frequency reference token."""
reference = str(value or 'auto').strip().lower()
if reference not in _VALID_FREQUENCY_REFERENCES:
choices = ', '.join(sorted(_VALID_FREQUENCY_REFERENCES))
raise ValueError(f'frequency_reference must be one of: {choices}')
return reference
def _station_frequency_reference(station: dict, listed_frequency_khz: float) -> str:
"""Infer whether a station frequency entry is carrier or already USB dial."""
for entry in station.get('frequencies', []):
try:
entry_khz = float(entry.get('khz'))
except (TypeError, ValueError):
continue
if abs(entry_khz - listed_frequency_khz) > 0.001:
continue
entry_ref = str(entry.get('reference', '')).strip().lower()
if entry_ref in ('carrier', 'dial'):
return entry_ref
station_ref = str(station.get('frequency_reference', '')).strip().lower()
if station_ref in ('carrier', 'dial'):
return station_ref
# Most published marine WeFax channel lists are carrier frequencies.
return 'carrier'
def resolve_tuning_frequency_khz(
listed_frequency_khz: float,
station_callsign: str = '',
frequency_reference: str = 'auto',
) -> tuple[float, str, bool]:
"""Resolve listed frequency to the actual USB dial frequency.
Args:
listed_frequency_khz: Frequency value provided by UI/API.
station_callsign: Station callsign used for metadata lookup.
frequency_reference: One of auto/carrier/dial.
Returns:
(tuned_frequency_khz, resolved_reference, offset_applied)
"""
listed = float(listed_frequency_khz)
if listed <= 0:
raise ValueError('frequency_khz must be greater than zero')
requested_ref = _normalize_frequency_reference(frequency_reference)
resolved_ref = requested_ref
if requested_ref == 'auto':
station = get_station(station_callsign) if station_callsign else None
if station:
resolved_ref = _station_frequency_reference(station, listed)
else:
# For ad-hoc frequencies (no station metadata), treat input as dial.
resolved_ref = 'dial'
if resolved_ref == 'carrier':
tuned = round(listed - WEFAX_USB_ALIGNMENT_OFFSET_KHZ, 3)
if tuned <= 0:
raise ValueError('frequency_khz too low after USB alignment offset')
return tuned, resolved_ref, True
return listed, resolved_ref, False
def get_current_broadcasts(callsign: str) -> list[dict]:
"""Return schedule entries closest to the current UTC time.
Returns up to 3 entries: the most recent past broadcast and the
next two upcoming ones, annotated with ``minutes_until`` or
``minutes_ago`` relative to now.
"""
station = get_station(callsign)
if not station:
return []
now = datetime.now(timezone.utc)
current_minutes = now.hour * 60 + now.minute
schedule = station.get('schedule', [])
if not schedule:
return []
# Convert schedule times to minutes-since-midnight for comparison
entries: list[tuple[int, dict]] = []
for entry in schedule:
parts = entry['utc'].split(':')
mins = int(parts[0]) * 60 + int(parts[1])
entries.append((mins, entry))
entries.sort(key=lambda x: x[0])
# Find closest entries relative to now
results = []
for mins, entry in entries:
diff = mins - current_minutes
# Wrap around midnight
if diff < -720:
diff += 1440
elif diff > 720:
diff -= 1440
annotated = dict(entry)
if diff >= 0:
annotated['minutes_until'] = diff
else:
annotated['minutes_ago'] = abs(diff)
annotated['_sort_key'] = abs(diff)
results.append(annotated)
results.sort(key=lambda x: x['_sort_key'])
# Return 3 nearest entries, clean up sort key
for r in results:
r.pop('_sort_key', None)
return results[:3]