sjg/trx-rs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[feat](trx-noaa): add NOAA APT satellite image decoder

Browse Source 
New trx-noaa crate: FFT-based Hilbert transform (rustfft) for 2400 Hz
AM demodulation, sync A detection via cross-correlation, line assembly
at 4160 Hz, and JPEG output via the image crate.

- trx-core: NoaaImage type, DecodedMessage::NoaaImage variant,
  noaa_decode_enabled/noaa_decode_reset_seq on RigState/RigSnapshot,
  AUDIO_MSG_NOAA_IMAGE = 0x16
- trx-server: DecoderHistories::noaa, run_noaa_decoder task (activates
  on noaa_decode_enabled, auto-finalises after 30 s silence), saves
  JPEGs to ~/.cache/trx-rs/noaa/<YYYY-MM-DD_HH-MM-SS>.jpg, forwards
  events over TCP audio channel and history replay

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Signed-off-by: Stan Grams <sjg@haxx.space>
This commit is contained in:
Stan Grams
2026-03-28 07:00:24 +01:00
parent a8b19227d5
commit 4b40d44814
12 changed files with 852 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/trx-server/Cargo.toml
+1
View File
@@ -39,5 +39,6 @@ trx-cw = { path = "../decoders/trx-cw" }
trx-decode-log = { path = "../decoders/trx-decode-log" }
trx-ftx = { path = "../decoders/trx-ftx" }
trx-wspr = { path = "../decoders/trx-wspr" }
trx-noaa = { path = "../decoders/trx-noaa" }
trx-protocol = { path = "../trx-protocol" }
trx-reporting = { path = "../trx-reporting" }
src/trx-server/src/audio.rs
+250 -5
View File
@@ -11,6 +11,7 @@ use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use bytes::Bytes;
use chrono::Local;
use flate2::write::GzEncoder;
use flate2::Compression;
use num_complex::Complex;
@@ -25,18 +26,21 @@ use trx_core::audio::{
parse_vchan_uuid_msg, read_audio_msg, write_audio_msg, write_vchan_audio_frame,
write_vchan_uuid_msg, AudioStreamInfo, AUDIO_MSG_AIS_DECODE, AUDIO_MSG_APRS_DECODE,
AUDIO_MSG_CW_DECODE, AUDIO_MSG_FT2_DECODE, AUDIO_MSG_FT4_DECODE, AUDIO_MSG_FT8_DECODE,
AUDIO_MSG_HF_APRS_DECODE, AUDIO_MSG_HISTORY_COMPRESSED, AUDIO_MSG_RX_FRAME,
AUDIO_MSG_STREAM_INFO, AUDIO_MSG_TX_FRAME, AUDIO_MSG_VCHAN_ALLOCATED, AUDIO_MSG_VCHAN_BW,
AUDIO_MSG_VCHAN_DESTROYED, AUDIO_MSG_VCHAN_FREQ, AUDIO_MSG_VCHAN_MODE, AUDIO_MSG_VCHAN_REMOVE,
AUDIO_MSG_VCHAN_SUB, AUDIO_MSG_VCHAN_UNSUB, AUDIO_MSG_VDES_DECODE, AUDIO_MSG_WSPR_DECODE,
AUDIO_MSG_HF_APRS_DECODE, AUDIO_MSG_HISTORY_COMPRESSED, AUDIO_MSG_NOAA_IMAGE,
AUDIO_MSG_RX_FRAME, AUDIO_MSG_STREAM_INFO, AUDIO_MSG_TX_FRAME, AUDIO_MSG_VCHAN_ALLOCATED,
AUDIO_MSG_VCHAN_BW, AUDIO_MSG_VCHAN_DESTROYED, AUDIO_MSG_VCHAN_FREQ, AUDIO_MSG_VCHAN_MODE,
AUDIO_MSG_VCHAN_REMOVE, AUDIO_MSG_VCHAN_SUB, AUDIO_MSG_VCHAN_UNSUB, AUDIO_MSG_VDES_DECODE,
AUDIO_MSG_WSPR_DECODE,
};
use trx_core::decode::{
AisMessage, AprsPacket, CwEvent, DecodedMessage, Ft8Message, VdesMessage, WsprMessage,
AisMessage, AprsPacket, CwEvent, DecodedMessage, Ft8Message, NoaaImage, VdesMessage,
WsprMessage,
};
use trx_core::rig::state::{RigMode, RigState};
use trx_core::vchan::SharedVChanManager;
use trx_cw::CwDecoder;
use trx_ftx::Ft8Decoder;
use trx_noaa::AptDecoder;
use trx_vdes::VdesDecoder;
use trx_wspr::WsprDecoder;
use uuid::Uuid;
@@ -51,6 +55,9 @@ const VDES_HISTORY_RETENTION: Duration = Duration::from_secs(24 * 60 * 60);
const CW_HISTORY_RETENTION: Duration = Duration::from_secs(24 * 60 * 60);
const FT8_HISTORY_RETENTION: Duration = Duration::from_secs(24 * 60 * 60);
const WSPR_HISTORY_RETENTION: Duration = Duration::from_secs(24 * 60 * 60);
const NOAA_HISTORY_RETENTION: Duration = Duration::from_secs(24 * 60 * 60);
/// Silence timeout before auto-finalising a NOAA pass (30 s without new lines).
const NOAA_PASS_SILENCE_TIMEOUT: Duration = Duration::from_secs(30);
const FT8_SAMPLE_RATE: u32 = 12_000;
const FT2_ASYNC_BUFFER_SAMPLES: usize = 45_000;
const FT2_ASYNC_TRIGGER_SAMPLES: usize = 9_000;
@@ -207,6 +214,7 @@ pub struct DecoderHistories {
pub ft4: Mutex<VecDeque<(Instant, Ft8Message)>>,
pub ft2: Mutex<VecDeque<(Instant, Ft8Message)>>,
pub wspr: Mutex<VecDeque<(Instant, WsprMessage)>>,
pub noaa: Mutex<VecDeque<(Instant, NoaaImage)>>,
/// Approximate total entry count across all decoders, maintained
/// atomically so `estimated_total_count()` avoids 9 lock acquisitions.
total_count: AtomicUsize,
@@ -224,6 +232,7 @@ impl DecoderHistories {
ft4: Mutex::new(VecDeque::new()),
ft2: Mutex::new(VecDeque::new()),
wspr: Mutex::new(VecDeque::new()),
noaa: Mutex::new(VecDeque::new()),
total_count: AtomicUsize::new(0),
})
}
@@ -583,6 +592,38 @@ impl DecoderHistories {
self.adjust_total_count(before, 0);
}
// --- NOAA ---
fn prune_noaa(history: &mut VecDeque<(Instant, NoaaImage)>) {
let cutoff = Instant::now() - NOAA_HISTORY_RETENTION;
while let Some((ts, _)) = history.front() {
if *ts < cutoff {
history.pop_front();
} else {
break;
}
}
}
pub fn record_noaa_image(&self, mut img: NoaaImage) {
if img.ts_ms.is_none() {
img.ts_ms = Some(current_timestamp_ms());
}
let mut h = self.noaa.lock().unwrap_or_else(|e| e.into_inner());
let before = h.len();
h.push_back((Instant::now(), img));
Self::prune_noaa(&mut h);
self.adjust_total_count(before, h.len());
}
pub fn snapshot_noaa_history(&self) -> Vec<NoaaImage> {
let mut h = self.noaa.lock().unwrap_or_else(|e| e.into_inner());
let before = h.len();
Self::prune_noaa(&mut h);
self.adjust_total_count(before, h.len());
h.iter().map(|(_, img)| img.clone()).collect()
}
/// Returns a quick (non-pruning) estimate of the total number of history
/// entries across all decoders, used for pre-allocating the replay blob.
///
@@ -2353,6 +2394,203 @@ pub async fn run_wspr_decoder(
}
}
// ---------------------------------------------------------------------------
// NOAA APT decoder task
// ---------------------------------------------------------------------------
/// Decode NOAA APT satellite images from FM-demodulated audio.
///
/// The task is idle until `state.noaa_decode_enabled` becomes `true`.
/// When the user disables the decoder (or 30 s of silence elapses with no
/// new decoded lines), the accumulated image is encoded as JPEG and saved to
/// `output_dir/<YYYY-MM-DD_HH-MM-SS>.jpg`.
pub async fn run_noaa_decoder(
sample_rate: u32,
channels: u16,
mut pcm_rx: broadcast::Receiver<Vec<f32>>,
mut state_rx: watch::Receiver<RigState>,
decode_tx: broadcast::Sender<DecodedMessage>,
histories: Arc<DecoderHistories>,
output_dir: std::path::PathBuf,
) {
info!("NOAA decoder started ({}Hz, {} ch)", sample_rate, channels);
let mut decoder = AptDecoder::new(sample_rate);
let mut last_reset_seq: u64 = 0;
let mut active = state_rx.borrow().noaa_decode_enabled;
let mut pass_start_ms: i64 = 0;
// Instant of the last time new lines were decoded (for auto-finalise)
let mut last_line_at = tokio::time::Instant::now();
if active {
pass_start_ms = current_timestamp_ms();
pcm_rx = pcm_rx.resubscribe();
}
loop {
if !active {
match state_rx.changed().await {
Ok(()) => {
let state = state_rx.borrow();
active = state.noaa_decode_enabled;
if active {
decoder.reset();
pass_start_ms = current_timestamp_ms();
last_line_at = tokio::time::Instant::now();
pcm_rx = pcm_rx.resubscribe();
}
if state.noaa_decode_reset_seq != last_reset_seq {
last_reset_seq = state.noaa_decode_reset_seq;
decoder.reset();
}
}
Err(_) => break,
}
continue;
}
let silence_deadline = last_line_at + NOAA_PASS_SILENCE_TIMEOUT;
tokio::select! {
recv = pcm_rx.recv() => {
match recv {
Ok(frame) => {
let reset_seq = state_rx.borrow().noaa_decode_reset_seq;
if reset_seq != last_reset_seq {
last_reset_seq = reset_seq;
decoder.reset();
pcm_rx = pcm_rx.resubscribe();
continue;
}
let mono = downmix_mono(frame, channels);
let new_lines = tokio::task::block_in_place(|| decoder.process_samples(&mono));
if new_lines > 0 {
last_line_at = tokio::time::Instant::now();
}
}
Err(broadcast::error::RecvError::Lagged(n)) => {
warn!("NOAA decoder: dropped {} PCM frames", n);
}
Err(broadcast::error::RecvError::Closed) => break,
}
}
changed = state_rx.changed() => {
match changed {
Ok(()) => {
let state = state_rx.borrow();
let was_active = active;
active = state.noaa_decode_enabled;
if state.noaa_decode_reset_seq != last_reset_seq {
last_reset_seq = state.noaa_decode_reset_seq;
decoder.reset();
pass_start_ms = current_timestamp_ms();
}
if was_active && !active {
// User disabled — finalise whatever we have
finalize_noaa_pass(
&mut decoder,
pass_start_ms,
&output_dir,
&decode_tx,
&histories,
)
.await;
} else if !was_active && active {
decoder.reset();
pass_start_ms = current_timestamp_ms();
last_line_at = tokio::time::Instant::now();
pcm_rx = pcm_rx.resubscribe();
}
}
Err(_) => break,
}
}
// Auto-finalise after sustained silence (satellite pass ended)
_ = tokio::time::sleep_until(silence_deadline), if decoder.line_count() > 0 => {
info!(
"NOAA: no new lines for {}s — finalising pass ({} lines)",
NOAA_PASS_SILENCE_TIMEOUT.as_secs(),
decoder.line_count()
);
finalize_noaa_pass(
&mut decoder,
pass_start_ms,
&output_dir,
&decode_tx,
&histories,
)
.await;
// Remain active; ready for the next pass
pass_start_ms = current_timestamp_ms();
last_line_at = tokio::time::Instant::now();
}
}
}
}
/// Encode all accumulated lines as JPEG, write to disk, and broadcast the
/// `DecodedMessage::NoaaImage` event. No-ops if fewer than 2 lines decoded.
async fn finalize_noaa_pass(
decoder: &mut AptDecoder,
pass_start_ms: i64,
output_dir: &std::path::Path,
decode_tx: &broadcast::Sender<DecodedMessage>,
histories: &Arc<DecoderHistories>,
) {
if decoder.line_count() < 2 {
decoder.reset();
return;
}
let pass_end_ms = current_timestamp_ms();
let Some(apt_image) = decoder.finalize() else {
decoder.reset();
return;
};
// Build output path: <output_dir>/<YYYY-MM-DD_HH-MM-SS>.jpg
let dt = chrono::Local::now();
let filename = dt.format("%Y-%m-%d_%H-%M-%S.jpg").to_string();
let path = output_dir.join(&filename);
if let Err(e) = std::fs::create_dir_all(output_dir) {
warn!(
"NOAA: failed to create output directory {:?}: {}",
output_dir, e
);
decoder.reset();
return;
}
match std::fs::write(&path, &apt_image.jpeg) {
Ok(()) => {
info!(
"NOAA: saved {} ({} lines, {} bytes) to {:?}",
filename,
apt_image.line_count,
apt_image.jpeg.len(),
path
);
let img = NoaaImage {
rig_id: None,
pass_start_ms: apt_image.first_line_ms,
pass_end_ms,
line_count: apt_image.line_count,
path: path.to_string_lossy().into_owned(),
ts_ms: Some(pass_end_ms),
};
histories.record_noaa_image(img.clone());
let _ = decode_tx.send(DecodedMessage::NoaaImage(img));
}
Err(e) => {
warn!("NOAA: failed to write {:?}: {}", path, e);
}
}
decoder.reset();
}
// ---------------------------------------------------------------------------
// Virtual-channel audio support
// ---------------------------------------------------------------------------
@@ -2952,6 +3190,11 @@ async fn handle_audio_client(
DecodedMessage::Cw,
AUDIO_MSG_CW_DECODE
);
push_history!(
histories.snapshot_noaa_history(),
DecodedMessage::NoaaImage,
AUDIO_MSG_NOAA_IMAGE
);
(blob, count)
};
@@ -3035,6 +3278,7 @@ async fn handle_audio_client(
DecodedMessage::Ft4(_) => AUDIO_MSG_FT4_DECODE,
DecodedMessage::Ft2(_) => AUDIO_MSG_FT2_DECODE,
DecodedMessage::Wspr(_) => AUDIO_MSG_WSPR_DECODE,
DecodedMessage::NoaaImage(_) => AUDIO_MSG_NOAA_IMAGE,
};
if let Ok(json) = serde_json::to_vec(&msg) {
if let Err(e) = write_audio_msg(&mut writer_for_rx, msg_type, &json).await {
@@ -3062,6 +3306,7 @@ async fn handle_audio_client(
DecodedMessage::Ft4(_) => AUDIO_MSG_FT4_DECODE,
DecodedMessage::Ft2(_) => AUDIO_MSG_FT2_DECODE,
DecodedMessage::Wspr(_) => AUDIO_MSG_WSPR_DECODE,
DecodedMessage::NoaaImage(_) => AUDIO_MSG_NOAA_IMAGE,
};
if let Ok(json) = serde_json::to_vec(&msg) {
if let Err(e) = write_audio_msg(&mut writer_for_rx, msg_type, &json).await {
src/trx-server/src/main.rs
+19
View File
@@ -793,6 +793,25 @@ fn spawn_rig_audio_stack(
_ = wait_for_shutdown(wspr_shutdown_rx) => {}
}
}));
// Spawn NOAA APT decoder task
let noaa_pcm_rx = pcm_tx.subscribe();
let noaa_state_rx = state_rx.clone();
let noaa_decode_tx = decode_tx.clone();
let noaa_sr = rig_cfg.audio.sample_rate;
let noaa_ch = rig_cfg.audio.channels;
let noaa_shutdown_rx = shutdown_rx.clone();
let noaa_histories = histories.clone();
let noaa_output_dir = dirs::cache_dir()
.unwrap_or_else(|| std::path::PathBuf::from(".cache"))
.join("trx-rs")
.join("noaa");
handles.push(tokio::spawn(async move {
tokio::select! {
_ = audio::run_noaa_decoder(noaa_sr, noaa_ch as u16, noaa_pcm_rx, noaa_state_rx, noaa_decode_tx, noaa_histories, noaa_output_dir) => {}
_ = wait_for_shutdown(noaa_shutdown_rx) => {}
}
}));
}
if rig_cfg.audio.tx_enabled {