diff --git a/Cargo.lock b/Cargo.lock index 449ec57d..7b8488a8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3165,6 +3165,7 @@ dependencies = [ name = "trx-core" version = "0.1.0" dependencies = [ + "base64", "flate2", "reqwest", "serde", diff --git a/docs/User-Manual.md b/docs/User-Manual.md index 62a235bc..65de98ff 100644 --- a/docs/User-Manual.md +++ b/docs/User-Manual.md @@ -153,6 +153,13 @@ When audio is enabled, at least one of `rx_enabled` or `tx_enabled` must be true | `sample_rate` | u32 | `1920000` | IQ capture rate in Hz | | `bandwidth` | u32 | `1500000` | Hardware IF filter bandwidth in Hz | | `center_offset_hz` | i64 | `100000` | Offset from dial to avoid DC spur | +| `spectrum_fft_size` | usize | `1024` | Spectrum FFT bins; power of two, 128–8192 | +| `spectrum_interval_ms` | u64 | `50` | How often a spectrum frame is pushed to subscribed clients | + +Spectrum is the largest thing on the client connection. On a slow or +high-latency link, halving `spectrum_fft_size` halves the bytes per frame (at +half the frequency resolution) and raising `spectrum_interval_ms` sends fewer of +them; see [Spectrum over a slow link](#spectrum-over-a-slow-link). #### `[sdr.gain]` @@ -295,6 +302,7 @@ Rigs without an explicit `id` get auto-generated IDs like `ft817_0`, `soapysdr_1 |-------|------|---------|-------------| | `url` | string | — | Server address (e.g. `localhost:4530`) | | `poll_interval_ms` | u64 | `750` | State poll interval | +| `spectrum_interval_ms` | u64 | `50` | Spectrum frame interval; also settable per `[[remotes]]` entry | #### `[remote.auth]` @@ -394,6 +402,41 @@ older single `port` key and `--rigctl-port` are ignored. The bridge is intended for WSJT-X integration via virtual audio devices (ALSA loopback on Linux, BlackHole on macOS). +### Spectrum over a slow link + +Spectrum dominates the server↔client connection: everything else is a few +hundred bytes, a frame is a few kilobytes. Three things govern what it costs. + +**Frames are pushed, not polled.** The client subscribes and the server sends +frames at `[sdr].spectrum_interval_ms`. Polling cost a round trip per frame, so +the rate was capped at 1/RTT — on a 200 ms link you could not exceed 5 frames a +second however often the client asked. Clients fall back to polling +automatically against a server too old to stream. + +**Bins travel as whole dBFS.** They are base64-encoded `i8` on the wire, about +an eighth of the JSON array of floats they used to be, at the resolution the +display draws anyway. + +**Both ends have a rate, and the slower one wins.** The server pushes no faster +than `[sdr].spectrum_interval_ms`; the client asks for no more than +`[[remotes]].spectrum_interval_ms`. + +For a link that struggles, start here: + +```toml +[trx-server.sdr] +spectrum_fft_size = 512 # half the bins, half the bytes +spectrum_interval_ms = 200 # 5 frames/s instead of 20 + +[[trx-client.remotes]] +name = "remote-site" +url = "radio.example.com:4530" +spectrum_interval_ms = 200 +``` + +That is roughly 0.7 KB per frame at 5 frames/s — about 3.5 KB/s, against +roughly 200 KB/s for 1024 float bins at 20 frames/s. + ### CLI Override Summary **trx-server:** diff --git a/src/trx-client/src/main.rs b/src/trx-client/src/main.rs index d2f67717..9146cd82 100644 --- a/src/trx-client/src/main.rs +++ b/src/trx-client/src/main.rs @@ -288,6 +288,7 @@ async fn async_init() -> DynResult { let token = cli.token.clone().or_else(|| cfg.remote.auth.token.clone()); let poll_interval_ms = cli.poll_interval_ms.unwrap_or(cfg.remote.poll_interval_ms); vec![RemoteEntry { + spectrum_interval_ms: cfg.remote.spectrum_interval_ms, name, url: url.clone(), rig_id, @@ -483,6 +484,13 @@ async fn async_init() -> DynResult { .map(|e| e.poll_interval_ms) .min() .unwrap_or(750); + // Entries sharing a server share its connections, so the most frequent + // request wins: whoever wants spectrum fastest sets the rate. + let spectrum_interval = entries + .iter() + .map(|e| e.spectrum_interval_ms) + .min() + .unwrap_or_else(|| remote_client::DEFAULT_SPECTRUM_INTERVAL.as_millis() as u64); let (server_tx, server_rx) = mpsc::channel::(RIG_TASK_CHANNEL_BUFFER); for entry in entries { @@ -496,6 +504,8 @@ async fn async_init() -> DynResult { known_rigs: frontend_runtime.routing.remote_rigs.clone(), rig_states: frontend_runtime.routing.rig_states.clone(), poll_interval: Duration::from_millis(poll_interval), + spectrum_interval: Duration::from_millis(spectrum_interval), + spectrum_stream_unsupported: Arc::new(std::sync::atomic::AtomicBool::new(false)), spectrum: frontend_runtime.spectrum.sender.clone(), rig_spectrums: frontend_runtime.spectrum.per_rig.clone(), server_connected: frontend_runtime.routing.server_connected.clone(), diff --git a/src/trx-client/src/remote_client.rs b/src/trx-client/src/remote_client.rs index 2f2d741d..b97a6f0a 100644 --- a/src/trx-client/src/remote_client.rs +++ b/src/trx-client/src/remote_client.rs @@ -20,7 +20,7 @@ use trx_core::{RigError, RigResult}; use trx_frontend::{RemoteRigEntry, SharedSpectrum}; use trx_protocol::rig_command_to_client; use trx_protocol::types::RigEntry; -use trx_protocol::{ClientCommand, ClientEnvelope, ClientResponse, MeterUpdate}; +use trx_protocol::{ClientCommand, ClientEnvelope, ClientResponse, MeterUpdate, SpectrumFrame}; // Endpoint parsing lives in `trx-config` so config validation and the // connection code agree on what a URL means. @@ -32,9 +32,10 @@ const SPECTRUM_IO_TIMEOUT: Duration = Duration::from_secs(3); const MAX_JSON_LINE_BYTES: usize = 256 * 1024; const MAX_CONSECUTIVE_POLL_FAILURES: u32 = 3; -// Keep remote spectrum reasonably responsive without returning to the old -// timeout churn caused by a much tighter request cadence. -const SPECTRUM_POLL_INTERVAL: Duration = Duration::from_millis(50); +// Default spectrum cadence when a config does not specify one. Both the push +// stream and the poll fallback run at the configured rate; see +// `[[remotes]].spectrum_interval_ms`. +pub const DEFAULT_SPECTRUM_INTERVAL: Duration = Duration::from_millis(50); #[derive(Clone)] pub struct RemoteClientConfig { @@ -43,6 +44,12 @@ pub struct RemoteClientConfig { pub selected_rig_id: Arc>>, pub known_rigs: Arc>>, pub poll_interval: Duration, + /// How often spectrum frames are wanted. Drives the poll fallback and is + /// the rate the client asks the server to push at. + pub spectrum_interval: Duration, + /// Set once a server has rejected `SubscribeSpectrum`, so later + /// connections to it go straight to polling instead of asking again. + pub spectrum_stream_unsupported: Arc, /// Spectrum watch sender; spectrum task publishes here, SSE clients subscribe. pub spectrum: Arc>, /// Shared flag: `true` while a TCP connection to trx-server is active. @@ -491,6 +498,100 @@ async fn send_get_sat_passes_on( )) } +/// What ended a spectrum stream attempt. +enum SpectrumStreamOutcome { + /// The stream ran and is over; the connection is spent. + Finished, + /// The server rejected the subscription. The connection is still usable, + /// so the caller can poll on it. + Unsupported, +} + +/// Subscribe to the server's spectrum push for one rig and publish frames as +/// they arrive. +/// +/// The server answers either with frames or, when it is too old to know the +/// command, with an error response — which leaves the connection usable for +/// polling, so falling back costs no reconnect. +async fn run_spectrum_stream( + config: &RemoteClientConfig, + writer: &mut (impl AsyncWriteExt + Unpin), + reader: &mut (impl AsyncBufRead + Unpin), + short_name: &str, + shutdown_rx: &mut watch::Receiver, +) -> RigResult { + let envelope = build_envelope( + config, + ClientCommand::SubscribeSpectrum, + Some(short_name.to_string()), + ); + let mut payload = serde_json::to_string(&envelope) + .map_err(|e| RigError::communication(format!("JSON serialize failed: {e}")))?; + payload.push('\n'); + time::timeout(SPECTRUM_IO_TIMEOUT, writer.write_all(payload.as_bytes())) + .await + .map_err(|_| RigError::communication("spectrum subscribe write timed out".to_string()))? + .map_err(|e| RigError::communication(format!("spectrum subscribe write failed: {e}")))?; + time::timeout(SPECTRUM_IO_TIMEOUT, writer.flush()) + .await + .map_err(|_| RigError::communication("spectrum subscribe flush timed out".to_string()))? + .map_err(|e| RigError::communication(format!("spectrum subscribe flush failed: {e}")))?; + + // Re-check what the UI wants: switching rigs has to end this stream so the + // connection can be rebuilt for the new one. + let mut supervisor = time::interval(Duration::from_millis(500)); + supervisor.tick().await; + + loop { + tokio::select! { + changed = shutdown_rx.changed() => { + match changed { + Ok(()) if *shutdown_rx.borrow() => return Ok(SpectrumStreamOutcome::Finished), + Ok(()) => {} + Err(_) => return Ok(SpectrumStreamOutcome::Finished), + } + } + _ = supervisor.tick() => { + let wanted = active_spectrum_rig_ids(config); + if wanted.len() != 1 || wanted[0] != short_name { + return Ok(SpectrumStreamOutcome::Finished); + } + } + line = read_limited_line(reader, MAX_JSON_LINE_BYTES) => { + let line = line + .map_err(|e| RigError::communication(format!("spectrum read failed: {e}")))? + .ok_or_else(|| { + RigError::communication("spectrum connection closed".to_string()) + })?; + let trimmed = line.trim_end(); + if trimmed.is_empty() { + continue; + } + match serde_json::from_str::(trimmed) { + Ok(frame) => publish_spectrum_frame(config, short_name, frame), + // Anything that is not a frame means the server would rather + // answer than stream: an older build rejecting the command. + Err(_) => return Ok(SpectrumStreamOutcome::Unsupported), + } + } + } + } +} + +/// Publish one pushed frame to the per-rig and selected-rig watch channels. +fn publish_spectrum_frame(config: &RemoteClientConfig, short_name: &str, frame: SpectrumFrame) { + if let Ok(map) = config.rig_spectrums.read() { + if let Some(tx) = map.get(short_name) { + tx.send_modify(|s| s.set(Some(frame.spectrum.clone()), frame.vchan_rds.clone())); + } + } + if selected_rig_id(config).as_deref() == Some(short_name) { + config + .spectrum + .send_modify(|s| s.set(Some(frame.spectrum), frame.vchan_rds)); + } +} + async fn handle_spectrum_connection( config: &RemoteClientConfig, stream: TcpStream, @@ -498,7 +599,33 @@ async fn handle_spectrum_connection( ) -> RigResult<()> { let (reader, mut writer) = stream.into_split(); let mut reader = BufReader::new(reader); - let mut interval = time::interval(SPECTRUM_POLL_INTERVAL); + + // Prefer the push stream: polling costs a round trip per frame, so on a + // high-latency link the frame rate is 1/RTT no matter what interval is + // configured. It only works for one rig per connection, and only against + // a server new enough to understand the command. + let streamable = active_spectrum_rig_ids(config); + if streamable.len() == 1 && !config.spectrum_stream_unsupported.load(Ordering::Relaxed) { + match run_spectrum_stream( + config, + &mut writer, + &mut reader, + &streamable[0], + shutdown_rx, + ) + .await? + { + SpectrumStreamOutcome::Finished => return Ok(()), + SpectrumStreamOutcome::Unsupported => { + info!("Server does not support spectrum streaming; falling back to polling"); + config + .spectrum_stream_unsupported + .store(true, Ordering::Relaxed); + } + } + } + + let mut interval = time::interval(config.spectrum_interval); // Cache the token outside the poll loop to avoid cloning it every 50ms. let cached_token = config.token.clone(); @@ -1187,6 +1314,7 @@ mod tests { use super::{has_short_names, resolve_server_rig_id, resolve_short_name}; use super::{ parse_audio_url, parse_remote_url, RemoteClientConfig, RemoteEndpoint, SharedSpectrum, + DEFAULT_SPECTRUM_INTERVAL, }; use std::collections::HashMap; use std::sync::atomic::AtomicBool; @@ -1381,6 +1509,8 @@ mod tests { selected_rig_id: Arc::new(Mutex::new(None)), known_rigs: Arc::new(Mutex::new(Vec::new())), poll_interval: Duration::from_millis(100), + spectrum_interval: DEFAULT_SPECTRUM_INTERVAL, + spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)), spectrum: Arc::new(spectrum_tx), server_connected: Arc::new(AtomicBool::new(false)), rig_server_connected: Arc::new(RwLock::new(HashMap::new())), @@ -1426,6 +1556,8 @@ mod tests { selected_rig_id: Arc::new(Mutex::new(Some("sdr".to_string()))), known_rigs: Arc::new(Mutex::new(Vec::new())), poll_interval: Duration::from_millis(500), + spectrum_interval: DEFAULT_SPECTRUM_INTERVAL, + spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)), spectrum: Arc::new(spectrum_tx), server_connected: Arc::new(AtomicBool::new(false)), rig_server_connected: Arc::new(RwLock::new(HashMap::new())), @@ -1441,6 +1573,130 @@ mod tests { assert_eq!(envelope.rig_id.as_deref(), Some("sdr")); } + fn stream_test_config(spectrum_tx: watch::Sender) -> super::RemoteClientConfig { + super::RemoteClientConfig { + addr: "127.0.0.1:4530".to_string(), + token: None, + selected_rig_id: Arc::new(Mutex::new(Some("sdr".to_string()))), + known_rigs: Arc::new(Mutex::new(Vec::new())), + poll_interval: Duration::from_millis(500), + spectrum_interval: DEFAULT_SPECTRUM_INTERVAL, + spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)), + spectrum: Arc::new(spectrum_tx), + server_connected: Arc::new(AtomicBool::new(false)), + rig_server_connected: Arc::new(RwLock::new(HashMap::new())), + rig_states: Arc::new(RwLock::new(HashMap::new())), + rig_spectrums: Arc::new(RwLock::new(HashMap::new())), + rig_id_to_short_name: HashMap::new(), + short_name_to_rig_id: Arc::new(RwLock::new(HashMap::new())), + sat_passes: Arc::new(RwLock::new(None)), + rig_meters: Arc::new(RwLock::new(HashMap::new())), + } + } + + /// A pushed frame reaches the watch channel the UI reads, with no request + /// from the client beyond the initial subscribe. + #[tokio::test] + async fn spectrum_stream_publishes_pushed_frames() { + use tokio::io::{AsyncBufReadExt, BufReader}; + + let (spectrum_tx, mut spectrum_rx) = watch::channel(SharedSpectrum::default()); + let config = stream_test_config(spectrum_tx); + let (shutdown_tx, mut shutdown_rx) = watch::channel(false); + + let (client_io, mut server_io) = tokio::io::duplex(64 * 1024); + let (client_read, mut client_write) = tokio::io::split(client_io); + let mut client_read = BufReader::new(client_read); + + let task = tokio::spawn(async move { + super::run_spectrum_stream( + &config, + &mut client_write, + &mut client_read, + "sdr", + &mut shutdown_rx, + ) + .await + }); + + // The server sees the subscribe, then pushes without being asked. + let mut server = BufReader::new(&mut server_io); + let mut subscribe = String::new(); + server.read_line(&mut subscribe).await.expect("subscribe"); + assert!( + subscribe.contains("subscribe_spectrum"), + "unexpected command: {subscribe}" + ); + + let frame = trx_protocol::SpectrumFrame { + rig_id: "sdr".to_string(), + spectrum: trx_core::rig::state::SpectrumData { + bins: vec![-70.0, -30.0], + center_hz: 14_200_000, + sample_rate: 1_920_000, + rds: None, + }, + vchan_rds: None, + }; + let mut line = serde_json::to_string(&frame).unwrap(); + line.push('\n'); + server_io.write_all(line.as_bytes()).await.expect("push"); + server_io.flush().await.expect("flush"); + + spectrum_rx.changed().await.expect("spectrum published"); + let published = spectrum_rx.borrow().clone(); + let spectrum = published.frame.expect("spectrum present"); + assert_eq!(spectrum.center_hz, 14_200_000); + assert_eq!(spectrum.bins, vec![-70.0, -30.0]); + + let _ = shutdown_tx.send(true); + let _ = task.await; + } + + /// An older server answers the unknown command with an error instead of + /// frames. That has to read as "poll instead", not as a dead connection. + #[tokio::test] + async fn spectrum_stream_falls_back_when_unsupported() { + use tokio::io::{AsyncBufReadExt, BufReader}; + + let (spectrum_tx, _spectrum_rx) = watch::channel(SharedSpectrum::default()); + let config = stream_test_config(spectrum_tx); + let (_shutdown_tx, mut shutdown_rx) = watch::channel(false); + + let (client_io, mut server_io) = tokio::io::duplex(64 * 1024); + let (client_read, mut client_write) = tokio::io::split(client_io); + let mut client_read = BufReader::new(client_read); + + let task = tokio::spawn(async move { + super::run_spectrum_stream( + &config, + &mut client_write, + &mut client_read, + "sdr", + &mut shutdown_rx, + ) + .await + }); + + let mut server = BufReader::new(&mut server_io); + let mut subscribe = String::new(); + server.read_line(&mut subscribe).await.expect("subscribe"); + + server_io + .write_all( + b"{\"success\":false,\"state\":null,\"error\":\"Invalid JSON: unknown variant\"}\n", + ) + .await + .expect("error response"); + server_io.flush().await.expect("flush"); + + let outcome = task.await.expect("join").expect("stream result"); + assert!( + matches!(outcome, super::SpectrumStreamOutcome::Unsupported), + "an error response should fall back to polling" + ); + } + #[test] fn build_envelope_translates_short_name_to_server_rig_id() { let (spectrum_tx, _spectrum_rx) = watch::channel(SharedSpectrum::default()); @@ -1454,6 +1710,8 @@ mod tests { selected_rig_id: Arc::new(Mutex::new(Some("home-hf".to_string()))), known_rigs: Arc::new(Mutex::new(Vec::new())), poll_interval: Duration::from_millis(500), + spectrum_interval: DEFAULT_SPECTRUM_INTERVAL, + spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)), spectrum: Arc::new(spectrum_tx), server_connected: Arc::new(AtomicBool::new(false)), rig_server_connected: Arc::new(RwLock::new(HashMap::new())), @@ -1486,6 +1744,8 @@ mod tests { selected_rig_id: Arc::new(Mutex::new(None)), known_rigs: Arc::new(Mutex::new(Vec::new())), poll_interval: Duration::from_millis(500), + spectrum_interval: DEFAULT_SPECTRUM_INTERVAL, + spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)), spectrum: Arc::new(spectrum_tx), server_connected: Arc::new(AtomicBool::new(false)), rig_server_connected: Arc::new(RwLock::new(HashMap::new())), @@ -1510,6 +1770,8 @@ mod tests { selected_rig_id: Arc::new(Mutex::new(None)), known_rigs: Arc::new(Mutex::new(Vec::new())), poll_interval: Duration::from_millis(500), + spectrum_interval: DEFAULT_SPECTRUM_INTERVAL, + spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)), spectrum: Arc::new(spectrum_tx), server_connected: Arc::new(AtomicBool::new(false)), rig_server_connected: Arc::new(RwLock::new(HashMap::new())), @@ -1550,6 +1812,8 @@ mod tests { selected_rig_id: Arc::new(Mutex::new(None)), known_rigs: known_rigs.clone(), poll_interval: Duration::from_millis(500), + spectrum_interval: DEFAULT_SPECTRUM_INTERVAL, + spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)), spectrum: Arc::new(spectrum_tx), server_connected: Arc::new(AtomicBool::new(false)), rig_server_connected: Arc::new(RwLock::new(HashMap::new())), @@ -1623,6 +1887,8 @@ mod tests { selected_rig_id, known_rigs, poll_interval: Duration::from_millis(500), + spectrum_interval: DEFAULT_SPECTRUM_INTERVAL, + spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)), spectrum: Arc::new(spectrum_tx), server_connected: Arc::new(AtomicBool::new(false)), rig_server_connected: Arc::new(RwLock::new(HashMap::new())), diff --git a/src/trx-client/trx-frontend/trx-frontend-http/frontend/src/api/generated.ts b/src/trx-client/trx-frontend/trx-frontend-http/frontend/src/api/generated.ts index 8119b243..79d3cd4d 100644 --- a/src/trx-client/trx-frontend/trx-frontend-http/frontend/src/api/generated.ts +++ b/src/trx-client/trx-frontend/trx-frontend-http/frontend/src/api/generated.ts @@ -78,6 +78,10 @@ export type RdsData = { pi?: number | null, program_service?: string | null, rad export type SpectrumData = { /** * FFT magnitude bins in dBFS, FFT-shifted so DC (centre frequency) is at index N/2. + * + * On the wire these are base64-encoded `i8` whole dBFS (see + * `spectrum_wire`), which is what the display draws anyway; the TypeScript + * type describes the decoded array the browser receives over SSE. */ bins: Array, /** diff --git a/src/trx-config/src/client.rs b/src/trx-config/src/client.rs index 27fc1504..6d7a8492 100644 --- a/src/trx-config/src/client.rs +++ b/src/trx-config/src/client.rs @@ -76,6 +76,8 @@ pub struct RemoteConfig { pub auth: RemoteAuthConfig, /// Poll interval in milliseconds. pub poll_interval_ms: u64, + /// Spectrum frame interval in milliseconds. + pub spectrum_interval_ms: u64, } impl Default for RemoteConfig { @@ -85,6 +87,7 @@ impl Default for RemoteConfig { rig_id: None, auth: RemoteAuthConfig::default(), poll_interval_ms: 750, + spectrum_interval_ms: default_spectrum_interval_ms(), } } } @@ -121,12 +124,23 @@ pub struct RemoteEntry { /// Poll interval in milliseconds. Defaults to 750. #[serde(default = "default_poll_interval_ms")] pub poll_interval_ms: u64, + /// How often spectrum frames are wanted from this server, in milliseconds. + /// Defaults to 50 (20 frames/s). + /// + /// Raise it on a slow or high-latency link: spectrum is by far the largest + /// thing on the connection, and the server pushes no faster than this. + #[serde(default = "default_spectrum_interval_ms")] + pub spectrum_interval_ms: u64, } fn default_poll_interval_ms() -> u64 { 750 } +fn default_spectrum_interval_ms() -> u64 { + 50 +} + /// Frontend configurations. #[derive(Debug, Clone, Default, Serialize, Deserialize)] #[serde(default)] @@ -421,6 +435,7 @@ impl ClientConfig { rig_id: self.remote.rig_id.clone(), auth: self.remote.auth.clone(), poll_interval_ms: self.remote.poll_interval_ms, + spectrum_interval_ms: self.remote.spectrum_interval_ms, }] } else { Vec::new() @@ -603,12 +618,21 @@ impl ClientConfig { i, entry.name )); } + if entry.spectrum_interval_ms == 0 { + return Err(format!( + "[[remotes]][{}].spectrum_interval_ms must be > 0 (name \"{}\")", + i, entry.name + )); + } } // Legacy [remote], kept for backward compatibility. if self.remote.poll_interval_ms == 0 { return Err("[remote].poll_interval_ms must be > 0".to_string()); } + if self.remote.spectrum_interval_ms == 0 { + return Err("[remote].spectrum_interval_ms must be > 0".to_string()); + } if let Some(url) = &self.remote.url { if url.trim().is_empty() { return Err("[remote].url must not be empty when set".to_string()); @@ -834,6 +858,7 @@ impl ClientConfig { token_file: None, }, poll_interval_ms: 750, + spectrum_interval_ms: 50, }, RemoteEntry { name: "home-vhf".to_string(), @@ -844,6 +869,7 @@ impl ClientConfig { token_file: None, }, poll_interval_ms: 750, + spectrum_interval_ms: 50, }, ], frontends: FrontendsConfig { @@ -994,6 +1020,7 @@ impl ConfigFile for ClientConfig { rig_id: Some("hf".to_string()), auth: RemoteAuthConfig::default(), poll_interval_ms: default_poll_interval_ms(), + spectrum_interval_ms: 50, }], ..Default::default() }; @@ -1311,6 +1338,7 @@ url = "remote.example.com:4530" rig_id: None, auth: RemoteAuthConfig::default(), poll_interval_ms: 750, + spectrum_interval_ms: 50, }], ..Default::default() }; @@ -1330,6 +1358,7 @@ url = "remote.example.com:4530" token_file: None, }, poll_interval_ms: 750, + spectrum_interval_ms: 50, }, ..Default::default() }; @@ -1368,6 +1397,7 @@ url = "remote.example.com:4530" rig_id: None, auth: RemoteAuthConfig::default(), poll_interval_ms: 750, + spectrum_interval_ms: 50, }], ..Default::default() }; @@ -1386,6 +1416,7 @@ url = "remote.example.com:4530" rig_id: None, auth: RemoteAuthConfig::default(), poll_interval_ms: 750, + spectrum_interval_ms: 50, }, RemoteEntry { name: "dup".to_string(), @@ -1393,6 +1424,7 @@ url = "remote.example.com:4530" rig_id: None, auth: RemoteAuthConfig::default(), poll_interval_ms: 750, + spectrum_interval_ms: 50, }, ], ..Default::default() @@ -1409,6 +1441,7 @@ url = "remote.example.com:4530" rig_id: None, auth: RemoteAuthConfig::default(), poll_interval_ms: 750, + spectrum_interval_ms: 50, }], ..Default::default() }; @@ -1427,6 +1460,7 @@ url = "remote.example.com:4530" rig_id: None, auth: RemoteAuthConfig::default(), poll_interval_ms: 750, + spectrum_interval_ms: 50, }], ..Default::default() }; @@ -1445,6 +1479,7 @@ url = "remote.example.com:4530" rig_id: None, auth: RemoteAuthConfig::default(), poll_interval_ms: 0, + spectrum_interval_ms: 50, }], ..Default::default() }; @@ -1499,6 +1534,45 @@ url = "remote.example.com:4530" assert!(warnings.is_empty(), "unexpected warnings: {warnings:?}"); } + // --- Spectrum cadence --- + + #[test] + fn test_spectrum_interval_defaults_and_parses() { + let config: ClientConfig = toml::from_str( + r#" +[[remotes]] +name = "slow-link" +url = "host:4530" +spectrum_interval_ms = 500 + +[[remotes]] +name = "lan" +url = "host2:4530" +"#, + ) + .unwrap(); + assert_eq!(config.remotes[0].spectrum_interval_ms, 500); + assert_eq!(config.remotes[1].spectrum_interval_ms, 50); + assert!(config.validate().is_ok()); + } + + #[test] + fn test_validate_rejects_zero_spectrum_interval() { + let config: ClientConfig = toml::from_str( + r#" +[[remotes]] +name = "hf" +url = "host:4530" +spectrum_interval_ms = 0 +"#, + ) + .unwrap(); + assert!(config + .validate() + .unwrap_err() + .contains("spectrum_interval_ms")); + } + // --- Secret indirection --- fn secret_file(content: &str) -> tempfile::NamedTempFile { @@ -1537,6 +1611,7 @@ url = "remote.example.com:4530" token_file: Some(f.path().to_str().unwrap().to_string()), }, poll_interval_ms: 750, + spectrum_interval_ms: 50, }], ..Default::default() }; @@ -1570,6 +1645,7 @@ url = "remote.example.com:4530" rig_id: None, auth: RemoteAuthConfig::default(), poll_interval_ms: 750, + spectrum_interval_ms: 50, }) .collect() } diff --git a/src/trx-config/src/server.rs b/src/trx-config/src/server.rs index 10595c51..733a16a9 100644 --- a/src/trx-config/src/server.rs +++ b/src/trx-config/src/server.rs @@ -399,12 +399,30 @@ pub struct SdrConfig { /// Default: 4. #[serde(default = "default_max_virtual_channels")] pub max_virtual_channels: usize, + /// FFT bin count for the spectrum display. Must be a power of two. + /// + /// Halving it halves both the DSP cost and the bytes each frame puts on + /// the network, at half the frequency resolution — worth it on a slow link. + #[serde(default = "default_spectrum_fft_size")] + pub spectrum_fft_size: usize, + /// How often the server pushes a spectrum frame to a subscribed client, in + /// milliseconds. Raise it on a slow or metered link. + #[serde(default = "default_spectrum_interval_ms")] + pub spectrum_interval_ms: u64, } fn default_max_virtual_channels() -> usize { 4 } +fn default_spectrum_fft_size() -> usize { + 1024 +} + +fn default_spectrum_interval_ms() -> u64 { + 50 +} + impl Default for SdrConfig { fn default() -> Self { Self { @@ -417,6 +435,8 @@ impl Default for SdrConfig { noise_blanker: SdrNoiseBlankerConfig::default(), channels: Vec::new(), max_virtual_channels: default_max_virtual_channels(), + spectrum_fft_size: default_spectrum_fft_size(), + spectrum_interval_ms: default_spectrum_interval_ms(), } } } @@ -913,6 +933,9 @@ fn validate_rig_instance( errors.push(format!("{prefix}[sdr.gain].max_value must be >= 0")); } } + if let Err(e) = validate_spectrum_config(prefix, &rig.sdr) { + errors.push(e); + } if let Err(e) = validate_sdr_squelch_config(&format!("{prefix}[sdr.squelch]"), &rig.sdr.squelch) { errors.push(e); @@ -1070,6 +1093,26 @@ fn validate_access(prefix: &str, access: &AccessConfig) -> Result<(), String> { Ok(()) } +/// A non-power-of-two FFT size still plans, but costs far more per frame; the +/// bounds keep the display useful without letting a typo allocate a huge FFT. +fn validate_spectrum_config(prefix: &str, sdr: &SdrConfig) -> Result<(), String> { + let size = sdr.spectrum_fft_size; + if !(128..=8192).contains(&size) { + return Err(format!( + "{prefix}[sdr].spectrum_fft_size must be in range 128..=8192" + )); + } + if !size.is_power_of_two() { + return Err(format!( + "{prefix}[sdr].spectrum_fft_size must be a power of two (got {size})" + )); + } + if sdr.spectrum_interval_ms == 0 { + return Err(format!("{prefix}[sdr].spectrum_interval_ms must be > 0")); + } + Ok(()) +} + fn validate_sdr_squelch_config(path: &str, squelch: &SdrSquelchConfig) -> Result<(), String> { if !squelch.threshold_db.is_finite() { return Err(format!("{path}.threshold_db must be finite")); @@ -1813,6 +1856,93 @@ port = 4531 assert!(warnings.is_empty(), "unexpected warnings: {warnings:?}"); } + // --- Spectrum transport knobs --- + + #[test] + fn test_spectrum_defaults() { + let cfg = SdrConfig::default(); + assert_eq!(cfg.spectrum_fft_size, 1024); + assert_eq!(cfg.spectrum_interval_ms, 50); + } + + fn sdr_cfg_with_spectrum(fft_size: usize, interval_ms: u64) -> ServerConfig { + let mut cfg = ServerConfig::default(); + cfg.rig.access.port = Some("/dev/ttyUSB0".to_string()); + cfg.rig.access.baud = Some(9600); + cfg.sdr.spectrum_fft_size = fft_size; + cfg.sdr.spectrum_interval_ms = interval_ms; + cfg + } + + #[test] + fn test_validate_accepts_smaller_power_of_two_fft() { + assert!(sdr_cfg_with_spectrum(256, 200).validate().is_ok()); + } + + #[test] + fn test_validate_rejects_non_power_of_two_fft() { + let err = sdr_cfg_with_spectrum(1000, 50) + .validate() + .expect_err("expected a power-of-two error"); + assert!(err.contains("power of two"), "unexpected error: {err}"); + } + + #[test] + fn test_validate_rejects_out_of_range_fft() { + assert!(sdr_cfg_with_spectrum(64, 50).validate().is_err()); + assert!(sdr_cfg_with_spectrum(16384, 50).validate().is_err()); + } + + #[test] + fn test_validate_rejects_zero_spectrum_interval() { + let err = sdr_cfg_with_spectrum(1024, 0) + .validate() + .expect_err("expected an interval error"); + assert!( + err.contains("spectrum_interval_ms"), + "unexpected error: {err}" + ); + } + + /// Per-rig validation covers the spectrum knobs too, and names the rig. + #[test] + fn test_validate_rejects_bad_spectrum_size_in_rig_entry() { + let toml_str = r#" +[[rigs]] +id = "sdr" +[rigs.rig] +model = "soapysdr" +[rigs.rig.access] +type = "sdr" +args = "driver=rtlsdr" +[rigs.audio] +port = 4532 +tx_enabled = false +[rigs.sdr] +spectrum_fft_size = 999 +"#; + let cfg: ServerConfig = toml::from_str(toml_str).unwrap(); + let err = cfg.validate().expect_err("expected a power-of-two error"); + assert!( + err.contains("power of two") && err.contains("sdr"), + "unexpected error: {err}" + ); + } + + #[test] + fn test_parse_spectrum_config_from_toml() { + let cfg: ServerConfig = toml::from_str( + r#" +[sdr] +spectrum_fft_size = 512 +spectrum_interval_ms = 250 +"#, + ) + .unwrap(); + assert_eq!(cfg.sdr.spectrum_fft_size, 512); + assert_eq!(cfg.sdr.spectrum_interval_ms, 250); + } + // --- Secret indirection --- #[test] diff --git a/src/trx-core/Cargo.toml b/src/trx-core/Cargo.toml index fab3ce3d..82a1fb47 100644 --- a/src/trx-core/Cargo.toml +++ b/src/trx-core/Cargo.toml @@ -17,3 +17,4 @@ uuid = { workspace = true } ts-rs = { version = "12.0.1", features = ["uuid-impl"] } sgp4 = "2" reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] } +base64 = "0.22" diff --git a/src/trx-core/src/rig/mod.rs b/src/trx-core/src/rig/mod.rs index 0a6cec44..a9fbd28b 100644 --- a/src/trx-core/src/rig/mod.rs +++ b/src/trx-core/src/rig/mod.rs @@ -19,6 +19,7 @@ pub mod command; pub mod controller; pub mod request; pub mod response; +pub mod spectrum_wire; pub mod state; /// How this backend communicates with the rig. diff --git a/src/trx-core/src/rig/spectrum_wire.rs b/src/trx-core/src/rig/spectrum_wire.rs new file mode 100644 index 00000000..49a186b6 --- /dev/null +++ b/src/trx-core/src/rig/spectrum_wire.rs @@ -0,0 +1,124 @@ +// SPDX-FileCopyrightText: 2026 Stan Grams +// +// SPDX-License-Identifier: GPL-2.0-or-later + +//! Compact wire encoding for spectrum bins. +//! +//! Bins are dBFS magnitudes, and the web UI has always drawn them from `i8` +//! values — the SSE hop to the browser quantizes and base64-encodes them. The +//! server→client hop, which is the one that crosses the operator's network, +//! used to send the same information as a JSON array of `f32`: around ten bytes +//! per bin instead of one, or roughly 10 KB per 1024-bin frame. +//! +//! Bins therefore travel as base64-encoded `i8` dBFS, about an eighth of the +//! size, at a resolution the display already rounds to. Decoding still accepts +//! the old array form, so a new client can read an older server. + +use base64::engine::general_purpose::STANDARD as BASE64; +use base64::Engine as _; +use serde::de::{SeqAccess, Visitor}; +use serde::{Deserializer, Serializer}; +use std::fmt; + +/// Quantize to whole dBFS and encode as base64. +pub fn serialize(bins: &[f32], serializer: S) -> Result { + let quantized: Vec = bins + .iter() + .map(|&db| { + let clamped = if db.is_finite() { db } else { -128.0 }; + clamped.round().clamp(-128.0, 127.0) as i8 as u8 + }) + .collect(); + serializer.serialize_str(&BASE64.encode(quantized)) +} + +/// Decode base64 bins, or a plain array of numbers from an older server. +pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result, D::Error> { + deserializer.deserialize_any(BinsVisitor) +} + +struct BinsVisitor; + +impl<'de> Visitor<'de> for BinsVisitor { + type Value = Vec; + + fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("base64-encoded i8 dBFS bins, or an array of numbers") + } + + fn visit_str(self, value: &str) -> Result { + let bytes = BASE64 + .decode(value) + .map_err(|e| E::custom(format!("invalid base64 spectrum bins: {e}")))?; + Ok(bytes.into_iter().map(|byte| byte as i8 as f32).collect()) + } + + fn visit_seq>(self, mut seq: A) -> Result { + let mut bins = Vec::with_capacity(seq.size_hint().unwrap_or(1024)); + while let Some(value) = seq.next_element::()? { + bins.push(value); + } + Ok(bins) + } +} + +#[cfg(test)] +mod tests { + use serde::{Deserialize, Serialize}; + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + struct Frame { + #[serde(with = "super")] + bins: Vec, + } + + #[test] + fn test_round_trip_quantizes_to_whole_db() { + let frame = Frame { + bins: vec![-73.4, -20.6, 0.0, -120.2], + }; + let json = serde_json::to_string(&frame).unwrap(); + let back: Frame = serde_json::from_str(&json).unwrap(); + assert_eq!(back.bins, vec![-73.0, -21.0, 0.0, -120.0]); + } + + #[test] + fn test_serializes_as_a_base64_string() { + let json = serde_json::to_string(&Frame { + bins: vec![-1.0, 0.0], + }) + .unwrap(); + assert!(json.contains('"'), "bins should be a string: {json}"); + assert!(!json.contains('['), "bins should not be an array: {json}"); + } + + #[test] + fn test_clamps_out_of_range_and_non_finite() { + let frame = Frame { + bins: vec![-400.0, 400.0, f32::NAN, f32::NEG_INFINITY], + }; + let json = serde_json::to_string(&frame).unwrap(); + let back: Frame = serde_json::from_str(&json).unwrap(); + assert_eq!(back.bins, vec![-128.0, 127.0, -128.0, -128.0]); + } + + #[test] + fn test_reads_the_old_array_form() { + let back: Frame = serde_json::from_str(r#"{"bins":[-73.25,-20.5]}"#).unwrap(); + assert_eq!(back.bins, vec![-73.25, -20.5]); + } + + /// The point of the change: an ordinary frame gets much smaller. + #[test] + fn test_frame_is_far_smaller_than_the_array_form() { + let bins: Vec = (0..1024).map(|i| -60.0 - (i % 40) as f32 * 0.37).collect(); + let compact = serde_json::to_string(&Frame { bins: bins.clone() }).unwrap(); + let verbose = serde_json::to_string(&bins).unwrap(); + assert!( + compact.len() * 5 < verbose.len(), + "compact {} bytes vs array {} bytes", + compact.len(), + verbose.len() + ); + } +} diff --git a/src/trx-core/src/rig/state.rs b/src/trx-core/src/rig/state.rs index 88297d30..18d39e20 100644 --- a/src/trx-core/src/rig/state.rs +++ b/src/trx-core/src/rig/state.rs @@ -394,6 +394,12 @@ fn default_wfm_denoise_level() -> WfmDenoiseLevel { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, TS)] pub struct SpectrumData { /// FFT magnitude bins in dBFS, FFT-shifted so DC (centre frequency) is at index N/2. + /// + /// On the wire these are base64-encoded `i8` whole dBFS (see + /// `spectrum_wire`), which is what the display draws anyway; the TypeScript + /// type describes the decoded array the browser receives over SSE. + #[serde(with = "crate::rig::spectrum_wire")] + #[ts(type = "Array")] pub bins: Vec, /// Centre frequency of the SDR capture in Hz. #[ts(type = "number")] diff --git a/src/trx-protocol/src/lib.rs b/src/trx-protocol/src/lib.rs index 05d2a766..3cdb3cff 100644 --- a/src/trx-protocol/src/lib.rs +++ b/src/trx-protocol/src/lib.rs @@ -18,4 +18,6 @@ pub use auth::{NoAuthValidator, SimpleTokenValidator, TokenValidator}; pub use codec::{mode_to_string, parse_envelope, parse_mode}; pub use decoders::{DecoderActivation, DecoderDescriptor, DECODER_REGISTRY}; pub use mapping::{client_command_to_rig, rig_command_to_client}; -pub use types::{ClientCommand, ClientEnvelope, ClientResponse, MeterUpdate, RigEntry}; +pub use types::{ + ClientCommand, ClientEnvelope, ClientResponse, MeterUpdate, RigEntry, SpectrumFrame, +}; diff --git a/src/trx-protocol/src/mapping.rs b/src/trx-protocol/src/mapping.rs index 75ec1678..1ee71ae9 100644 --- a/src/trx-protocol/src/mapping.rs +++ b/src/trx-protocol/src/mapping.rs @@ -102,7 +102,7 @@ macro_rules! define_command_mapping { define_command_mapping! { // ── Client-only variants (no RigCommand counterpart) ───────────── - client_only: GetRigs, GetSatPasses, SubscribeMeter; + client_only: GetRigs, GetSatPasses, SubscribeMeter, SubscribeSpectrum; // ── Unit variants (no payload) ─────────────────────────────────── unit: diff --git a/src/trx-protocol/src/types.rs b/src/trx-protocol/src/types.rs index b1ad0294..4c696a2e 100644 --- a/src/trx-protocol/src/types.rs +++ b/src/trx-protocol/src/types.rs @@ -130,6 +130,28 @@ pub enum ClientCommand { /// newline-delimited `MeterUpdate` JSON frames and no further commands or /// regular responses are sent. Intended for a dedicated TCP connection. SubscribeMeter, + /// Subscribe to a per-rig spectrum stream on this connection. Like + /// `SubscribeMeter`, the connection becomes a one-way flow of + /// newline-delimited `SpectrumFrame` JSON and no further commands or + /// regular responses are sent. + /// + /// Polling `GetSpectrum` costs a round trip per frame, which caps the frame + /// rate at 1/RTT no matter how often the client asks; the server pushes at + /// its own cadence instead. Clients fall back to polling when the server + /// rejects this command. + SubscribeSpectrum, +} + +/// One spectrum frame pushed by the server on a dedicated spectrum stream. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct SpectrumFrame { + /// Rig identifier this frame belongs to. + pub rig_id: String, + /// The frame itself; bins travel base64-encoded (see `spectrum_wire`). + pub spectrum: trx_core::rig::state::SpectrumData, + /// Virtual-channel RDS state, mirroring what `GetSpectrum` returned. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub vchan_rds: Option>, } /// Fast meter sample pushed by the server on a dedicated meter stream. diff --git a/src/trx-server/src/listener.rs b/src/trx-server/src/listener.rs index b944a019..7889353f 100644 --- a/src/trx-server/src/listener.rs +++ b/src/trx-server/src/listener.rs @@ -494,6 +494,58 @@ where } }; + // SubscribeSpectrum: turns this connection into a one-way spectrum + // stream. Polling GetSpectrum costs a round trip per frame, so a + // client on a slow link could never reach the frame rate it asked for; + // pushing decouples the rate from the latency. + if matches!(envelope.cmd, ClientCommand::SubscribeSpectrum) { + let mut spectrum_rx = handle.spectrum_tx.subscribe(); + let io_timeout = timeouts.io_timeout; + info!( + "Client {} subscribed to spectrum stream for rig '{}'", + addr, target_rig_id + ); + loop { + tokio::select! { + frame = spectrum_rx.recv() => { + match frame { + Ok(frame) => { + let Ok(mut line) = serde_json::to_string(&frame) else { continue }; + line.push('\n'); + let write = time::timeout( + io_timeout, + writer.write_all(line.as_bytes()), + ).await; + match write { + Ok(Ok(())) => {} + Ok(Err(e)) => { + info!("Client {} spectrum write failed: {}", addr, e); + break; + } + Err(_) => { + info!("Client {} spectrum write timed out", addr); + break; + } + } + } + // A client that cannot keep up skips to the newest + // frame; stale spectrum is not worth drawing. + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } + changed = shutdown_rx.changed() => { + match changed { + Ok(()) if *shutdown_rx.borrow() => break, + Ok(()) => {} + Err(_) => break, + } + } + } + } + break; + } + // SubscribeMeter: turns this connection into a one-way meter stream. // No regular responses are produced; the connection lives until the // client disconnects or shutdown fires. @@ -732,6 +784,7 @@ mod tests { let (state_tx, state_rx) = watch::channel(state); let _state_tx = state_tx; let (meter_tx, _) = tokio::sync::broadcast::channel(8); + let (spectrum_tx, _) = tokio::sync::broadcast::channel(4); let handle = RigHandle { rig_id: "default".to_string(), display_name: "Default Rig".to_string(), @@ -739,6 +792,7 @@ mod tests { state_rx, audio_port: 4531, meter_tx, + spectrum_tx, }; let mut map = HashMap::new(); map.insert("default".to_string(), handle); @@ -933,6 +987,7 @@ mod tests { state_rx: state_rx_a, audio_port: 4531, meter_tx: meter_tx_a, + spectrum_tx: tokio::sync::broadcast::channel(4).0, }; let (tx_b, rx_b) = mpsc::channel::(8); @@ -945,6 +1000,7 @@ mod tests { state_rx: state_rx_b, audio_port: 4532, meter_tx: meter_tx_b, + spectrum_tx: tokio::sync::broadcast::channel(4).0, }; let mut map = HashMap::new(); @@ -953,6 +1009,67 @@ mod tests { (Arc::new(map), "rig_hf".to_string(), rx_a, rx_b) } + /// Polling spectrum costs a round trip per frame, so the rate a client can + /// reach is capped by latency rather than by what it asked for. Subscribed + /// clients get frames pushed instead; this is that path end to end. + #[tokio::test] + async fn subscribe_spectrum_pushes_frames() { + use trx_core::rig::state::SpectrumData; + + let (rigs, default_id) = make_rigs(sample_state()); + let spectrum_tx = rigs.get("default").expect("rig").spectrum_tx.clone(); + let ctx = make_ctx(rigs, default_id, HashSet::new()); + let (mut reader, mut writer, handle, shutdown_tx) = spawn_client_io(ctx); + + writer + .write_all(br#"{"cmd":"subscribe_spectrum"}"#) + .await + .expect("write"); + writer.write_all(b"\n").await.expect("newline"); + writer.flush().await.expect("flush"); + + // The subscription is registered asynchronously; publish until it takes. + let frame = trx_protocol::SpectrumFrame { + rig_id: "default".to_string(), + spectrum: SpectrumData { + bins: vec![-73.4, -20.6, 0.0], + center_hz: 14_200_000, + sample_rate: 1_920_000, + rds: None, + }, + vchan_rds: None, + }; + let mut line = String::new(); + for _ in 0..50 { + let _ = spectrum_tx.send(frame.clone()); + tokio::select! { + read = reader.read_line(&mut line) => { + if read.expect("read") > 0 && !line.trim().is_empty() { + break; + } + line.clear(); + } + _ = tokio::time::sleep(Duration::from_millis(20)) => {} + } + } + + let received: trx_protocol::SpectrumFrame = + serde_json::from_str(line.trim()).unwrap_or_else(|e| panic!("frame {line:?}: {e}")); + assert_eq!(received.rig_id, "default"); + assert_eq!(received.spectrum.center_hz, 14_200_000); + // Bins survive the trip quantized to whole dBFS, which is the + // resolution the display draws at anyway. + assert_eq!(received.spectrum.bins, vec![-73.0, -21.0, 0.0]); + // And they travel as base64 rather than a JSON array of floats. + assert!( + !line.contains("-73"), + "bins should not be spelled out on the wire: {line}" + ); + + let _ = shutdown_tx.send(true); + let _ = handle.await; + } + #[tokio::test] async fn multi_rig_state_isolation() { let state_hf = sample_state_custom("HF-Dummy", 14_200_000, trx_core::RigMode::USB); diff --git a/src/trx-server/src/main.rs b/src/trx-server/src/main.rs index e5a048ac..045c43e9 100644 --- a/src/trx-server/src/main.rs +++ b/src/trx-server/src/main.rs @@ -360,6 +360,7 @@ fn build_sdr_rig_from_instance(rig_cfg: &RigInstanceConfig) -> SdrRigBuildResult max_virtual_channels: rig_cfg.sdr.max_virtual_channels, nb_enabled: rig_cfg.sdr.noise_blanker.enabled, nb_threshold: rig_cfg.sdr.noise_blanker.threshold, + spectrum_fft_size: rig_cfg.sdr.spectrum_fft_size, })?; let pcm_rx = sdr_rig.subscribe_pcm(); @@ -460,6 +461,7 @@ fn build_rig_task_config( prebuilt_rig: None, command_exec_timeout: Duration::from_millis(timeouts.command_exec_timeout_ms), poll_refresh_timeout: Duration::from_millis(timeouts.poll_refresh_timeout_ms), + spectrum_interval_ms: rig_cfg.sdr.spectrum_interval_ms, } } @@ -1215,6 +1217,9 @@ async fn main() -> DynResult<()> { let (state_tx, state_rx) = watch::channel(initial_state); let (meter_tx, _) = broadcast::channel::(rig_handle::METER_BROADCAST_CAPACITY); + let (spectrum_tx, _) = broadcast::channel::( + rig_handle::SPECTRUM_BROADCAST_CAPACITY, + ); let mut task_config = build_rig_task_config( rig_cfg, @@ -1242,12 +1247,14 @@ async fn main() -> DynResult<()> { let rig_shutdown_rx = shutdown_rx.clone(); let rig_id_supervisor = rig_cfg.id.clone(); let meter_tx_task = meter_tx.clone(); + let spectrum_tx_task = spectrum_tx.clone(); task_handles.push(tokio::spawn(async move { let result = rig_task::run_rig_task( task_config, rig_rx, state_tx.clone(), meter_tx_task, + spectrum_tx_task, rig_shutdown_rx, ) .await; @@ -1294,6 +1301,7 @@ async fn main() -> DynResult<()> { rig_handles.insert( rig_cfg.id.clone(), RigHandle { + spectrum_tx: spectrum_tx.clone(), rig_id: rig_cfg.id.clone(), display_name: rig_cfg.display_name().to_string(), rig_tx, diff --git a/src/trx-server/src/rig_handle.rs b/src/trx-server/src/rig_handle.rs index 8dd0e6ad..9456f86f 100644 --- a/src/trx-server/src/rig_handle.rs +++ b/src/trx-server/src/rig_handle.rs @@ -8,13 +8,18 @@ use tokio::sync::{broadcast, mpsc, watch}; use trx_core::rig::request::RigRequest; use trx_core::rig::state::RigState; -use trx_protocol::MeterUpdate; +use trx_protocol::{MeterUpdate, SpectrumFrame}; /// Bounded broadcast capacity for the meter stream. Keeps ~0.5 s of buffered /// samples at 30 Hz — more than enough slack to tolerate a scheduling blip /// without forcing the producer to block or drop silently. pub const METER_BROADCAST_CAPACITY: usize = 16; +/// Bounded broadcast capacity for the spectrum stream. Frames are large and +/// only the newest one is worth drawing, so the buffer stays shallow: a slow +/// client lags and skips rather than making the server hold stale frames. +pub const SPECTRUM_BROADCAST_CAPACITY: usize = 4; + /// A handle to a single running rig backend. /// /// One `RigHandle` is created per rig in `main.rs` and stored in the shared @@ -34,4 +39,8 @@ pub struct RigHandle { /// ~6–7 Hz (CAT). Consumed by `SubscribeMeter` clients; independent of /// the slower `state_rx` snapshot path. pub meter_tx: broadcast::Sender, + /// Per-rig spectrum frames published by `rig_task` while at least one + /// client is subscribed. Consumed by `SubscribeSpectrum` clients; the + /// producer skips the work entirely when nobody is listening. + pub spectrum_tx: broadcast::Sender, } diff --git a/src/trx-server/src/rig_task.rs b/src/trx-server/src/rig_task.rs index bb87b6e1..b51ebd46 100644 --- a/src/trx-server/src/rig_task.rs +++ b/src/trx-server/src/rig_task.rs @@ -23,7 +23,7 @@ use trx_core::rig::request::RigRequest; use trx_core::rig::state::{RigMode, RigSnapshot, RigState}; use trx_core::rig::{RigCat, RigRxStatus, RigTxStatus}; use trx_core::{DynResult, RigError, RigResult}; -use trx_protocol::MeterUpdate; +use trx_protocol::{MeterUpdate, SpectrumFrame}; use crate::audio::DecoderHistories; use crate::error::is_invalid_bcd_error; @@ -64,6 +64,8 @@ pub struct RigTaskConfig { pub command_exec_timeout: Duration, /// Maximum time for a CAT poll refresh cycle. pub poll_refresh_timeout: Duration, + /// How often to push a spectrum frame to subscribed clients, in ms. + pub spectrum_interval_ms: u64, } impl Default for RigTaskConfig { @@ -94,6 +96,7 @@ impl Default for RigTaskConfig { prebuilt_rig: None, command_exec_timeout: DEFAULT_COMMAND_EXEC_TIMEOUT, poll_refresh_timeout: DEFAULT_POLL_REFRESH_TIMEOUT, + spectrum_interval_ms: 50, } } } @@ -115,6 +118,7 @@ pub async fn run_rig_task( mut rx: mpsc::Receiver, state_tx: watch::Sender, meter_tx: broadcast::Sender, + spectrum_tx: broadcast::Sender, mut shutdown_rx: watch::Receiver, ) -> DynResult<()> { let histories = config.histories.clone(); @@ -273,6 +277,13 @@ pub async fn run_rig_task( } else { Duration::from_millis(150) }; + // Spectrum frames get their own tick so a subscribed client never pays a + // round trip per frame. The FFT is computed by the SDR thread either way; + // this only reads the latest result, and only while somebody is subscribed. + let spectrum_tick_duration = Duration::from_millis(config.spectrum_interval_ms.max(1)); + let mut spectrum_tick: std::pin::Pin> = + Box::pin(tokio::time::sleep(spectrum_tick_duration)); + let meter_task_start = Instant::now(); let meter_state_delta_db: f64 = 0.25; let rig_id = config.rig_id.clone(); @@ -302,6 +313,21 @@ pub async fn run_rig_task( Err(_) => break, } } + // Push the latest spectrum frame to subscribed clients. + _ = &mut spectrum_tick => { + spectrum_tick = Box::pin(tokio::time::sleep(spectrum_tick_duration)); + // `send` fails only when nobody is listening, but building the + // frame clones a few KB of bins, so check before doing the work. + if spectrum_tx.receiver_count() > 0 { + if let Some(spectrum) = rig.as_sdr_ref().and_then(|s| s.get_spectrum()) { + let _ = spectrum_tx.send(SpectrumFrame { + rig_id: rig_id.clone(), + spectrum, + vchan_rds: rig.as_sdr_ref().and_then(|s| s.get_vchan_rds()), + }); + } + } + } // Fast meter-only refresh between full polls. _ = &mut meter_tick => { meter_tick = Box::pin(tokio::time::sleep(meter_tick_duration)); diff --git a/src/trx-server/trx-backend/trx-backend-soapysdr/src/dsp.rs b/src/trx-server/trx-backend/trx-backend-soapysdr/src/dsp.rs index 1dc351bd..efb95a22 100644 --- a/src/trx-server/trx-backend/trx-backend-soapysdr/src/dsp.rs +++ b/src/trx-server/trx-backend/trx-backend-soapysdr/src/dsp.rs @@ -154,6 +154,7 @@ impl SdrPipeline { squelch_cfg: VirtualSquelchConfig, nb_cfg: NoiseBlankerConfig, channels: &[(f64, RigMode, u32)], + spectrum_fft_size: usize, ) -> Self { const IQ_BROADCAST_CAPACITY: usize = 64; let (iq_tx, _iq_rx) = broadcast::channel::>>(IQ_BROADCAST_CAPACITY); @@ -219,6 +220,7 @@ impl SdrPipeline { .name("sdr-iq-read".to_string()) .spawn(move || { iq_read_loop( + spectrum_fft_size, source, sdr_sample_rate, thread_dsps, @@ -317,6 +319,7 @@ pub const IQ_BLOCK_SIZE: usize = 4096; #[allow(clippy::too_many_arguments)] fn iq_read_loop( + spectrum_fft_size: usize, mut source: Box, sdr_sample_rate: u32, channel_dsps: Arc>>>>, @@ -335,7 +338,7 @@ fn iq_read_loop( }; let throttle = !source.is_blocking(); - let mut spectrum = SpectrumSnapshotter::new(); + let mut spectrum = SpectrumSnapshotter::new(spectrum_fft_size); let mut read_error_streak: u32 = 0; let mut zero_read_streak: u32 = 0; let mut overflow_log_window_start: Option = None; @@ -572,6 +575,7 @@ mod tests { VirtualSquelchConfig::default(), NoiseBlankerConfig::default(), &[(200_000.0, RigMode::USB, 3000)], + 1024, ); assert_eq!(pipeline.pcm_senders.len(), 1); assert_eq!(pipeline.channel_dsps.read().unwrap().len(), 1); @@ -590,6 +594,7 @@ mod tests { VirtualSquelchConfig::default(), NoiseBlankerConfig::default(), &[], + 1024, ); assert_eq!(pipeline.pcm_senders.len(), 0); assert_eq!(pipeline.channel_dsps.read().unwrap().len(), 0); diff --git a/src/trx-server/trx-backend/trx-backend-soapysdr/src/dsp/spectrum.rs b/src/trx-server/trx-backend/trx-backend-soapysdr/src/dsp/spectrum.rs index 6cb0ee6a..858b1e73 100644 --- a/src/trx-server/trx-backend/trx-backend-soapysdr/src/dsp/spectrum.rs +++ b/src/trx-server/trx-backend/trx-backend-soapysdr/src/dsp/spectrum.rs @@ -9,28 +9,36 @@ use num_complex::Complex; use rustfft::num_complex::Complex as FftComplex; use rustfft::FftPlanner; -/// Number of FFT bins for the spectrum display. -pub(super) const SPECTRUM_FFT_SIZE: usize = 1024; +/// Default number of FFT bins for the spectrum display, used when the config +/// does not say otherwise. +pub(super) const DEFAULT_SPECTRUM_FFT_SIZE: usize = 1024; /// Update the spectrum buffer every this many IQ blocks (~10 Hz at 1.92 MHz / 4096 block). pub(super) const SPECTRUM_UPDATE_BLOCKS: usize = 4; pub(super) struct SpectrumSnapshotter { + fft_size: usize, hann_window: Vec, fft: std::sync::Arc>, counter: usize, } impl SpectrumSnapshotter { - pub(super) fn new() -> Self { - let hann_window: Vec = (0..SPECTRUM_FFT_SIZE) - .map(|i| 0.5 * (1.0 - (2.0 * PI * i as f32 / (SPECTRUM_FFT_SIZE - 1) as f32).cos())) + pub(super) fn new(fft_size: usize) -> Self { + let fft_size = if fft_size >= 2 { + fft_size + } else { + DEFAULT_SPECTRUM_FFT_SIZE + }; + let hann_window: Vec = (0..fft_size) + .map(|i| 0.5 * (1.0 - (2.0 * PI * i as f32 / (fft_size - 1) as f32).cos())) .collect(); let mut planner = FftPlanner::::new(); - let fft = planner.plan_fft_forward(SPECTRUM_FFT_SIZE); + let fft = planner.plan_fft_forward(fft_size); Self { + fft_size, hann_window, fft, counter: 0, @@ -48,7 +56,7 @@ impl SpectrumSnapshotter { } self.counter = 0; - let take = samples.len().min(SPECTRUM_FFT_SIZE); + let take = samples.len().min(self.fft_size); let mut buf: Vec> = samples[..take] .iter() .enumerate() @@ -59,16 +67,15 @@ impl SpectrumSnapshotter { ) }) .collect(); - buf.resize(SPECTRUM_FFT_SIZE, FftComplex::new(0.0, 0.0)); + buf.resize(self.fft_size, FftComplex::new(0.0, 0.0)); self.fft.process(&mut buf); - let half = SPECTRUM_FFT_SIZE / 2; + let half = self.fft_size / 2; let bins: Vec = buf[half..] .iter() .chain(buf[..half].iter()) .map(|value| { - let mag = - (value.re * value.re + value.im * value.im).sqrt() / SPECTRUM_FFT_SIZE as f32; + let mag = (value.re * value.re + value.im * value.im).sqrt() / self.fft_size as f32; 20.0 * mag.max(1e-10_f32).log10() }) .collect(); diff --git a/src/trx-server/trx-backend/trx-backend-soapysdr/src/lib.rs b/src/trx-server/trx-backend/trx-backend-soapysdr/src/lib.rs index abf12ff6..f7ff9104 100644 --- a/src/trx-server/trx-backend/trx-backend-soapysdr/src/lib.rs +++ b/src/trx-server/trx-backend/trx-backend-soapysdr/src/lib.rs @@ -73,6 +73,11 @@ pub struct SoapySdrConfig { pub nb_enabled: bool, /// Noise blanker impulse threshold multiplier. pub nb_threshold: f64, + /// FFT bin count for the spectrum display; a power of two. + /// + /// Fewer bins cost less DSP and put fewer bytes on the network per frame, + /// which is what a slow server↔client link cares about. + pub spectrum_fft_size: usize, } impl Default for SoapySdrConfig { @@ -99,6 +104,7 @@ impl Default for SoapySdrConfig { max_virtual_channels: 4, nb_enabled: false, nb_threshold: 10.0, + spectrum_fft_size: 1024, } } } @@ -194,6 +200,7 @@ impl SoapySdrRig { let max_virtual_channels = config.max_virtual_channels; let nb_enabled = config.nb_enabled; let nb_threshold = config.nb_threshold; + let spectrum_fft_size = config.spectrum_fft_size; tracing::info!( "initialising SoapySDR backend (args={:?}, gain_mode={:?}, gain_db={}, max_gain_db={:?})", args, @@ -293,6 +300,7 @@ impl SoapySdrRig { threshold: nb_threshold as f32, }, &all_channels, + spectrum_fft_size, )); let info = RigInfo { @@ -418,6 +426,7 @@ impl SoapySdrRig { nb_threshold: f64, ) -> DynResult { Self::new_from_config(SoapySdrConfig { + spectrum_fft_size: 1024, args: args.to_string(), channels: channels.to_vec(), gain_mode: gain_mode.to_string(), diff --git a/src/trx-server/trx-backend/trx-backend-soapysdr/src/vchan_impl.rs b/src/trx-server/trx-backend/trx-backend-soapysdr/src/vchan_impl.rs index 127c8931..c052978b 100644 --- a/src/trx-server/trx-backend/trx-backend-soapysdr/src/vchan_impl.rs +++ b/src/trx-server/trx-backend/trx-backend-soapysdr/src/vchan_impl.rs @@ -439,6 +439,7 @@ mod tests { VirtualSquelchConfig::default(), NoiseBlankerConfig::default(), &[(0.0, RigMode::USB, 3_000)], + 1024, )) } diff --git a/trx-rs.toml.example b/trx-rs.toml.example index cd2c1f4d..dfa939ba 100644 --- a/trx-rs.toml.example +++ b/trx-rs.toml.example @@ -120,6 +120,8 @@ wfm_deemphasis_us = 50 center_offset_hz = 100000 channels = [] max_virtual_channels = 4 +spectrum_fft_size = 1024 +spectrum_interval_ms = 50 # "auto" for hardware AGC, or "manual". [trx-server.sdr.gain] @@ -160,6 +162,7 @@ log_level = "info" # Legacy single-remote form; prefer [[remotes]] below. [trx-client.remote] poll_interval_ms = 750 +spectrum_interval_ms = 50 [trx-client.remote.auth] @@ -168,6 +171,7 @@ name = "home-hf" url = "192.168.1.100:4530" rig_id = "hf" poll_interval_ms = 750 +spectrum_interval_ms = 50 [trx-client.remotes.auth] token = "my-token" @@ -177,6 +181,7 @@ name = "home-vhf" url = "192.168.1.100:4530" rig_id = "vhf" poll_interval_ms = 750 +spectrum_interval_ms = 50 [trx-client.remotes.auth] token = "my-token"