[feat](trx-rs): make spectrum affordable over a slow link
Spectrum dominates the server↔client connection, and all three things that govern its cost were working against a poor link. **It was polled, one round trip per frame.** The client asked for a frame every 50 ms on a dedicated connection and waited for the reply, so the frame rate was capped at 1/RTT — on a 200 ms link, five frames a second no matter what was configured. Add SubscribeSpectrum alongside the existing SubscribeMeter: the server pushes frames from a per-rig broadcast that rig_task fills only while somebody is subscribed. A server too old to know the command answers with an error and leaves the connection usable, so the client falls back to polling on the same connection without reconnecting. **Bins were JSON floats.** 1024 bins spelled out as decimal text is around 10 KB a frame, ~200 KB/s at full rate — while the very next hop, client to browser, already sends the same information as base64 i8 in about 1.4 KB. Bins now travel base64-encoded whole dBFS, the resolution the display draws at anyway. Decoding still accepts the old array form. **Nothing was tunable.** [sdr].spectrum_fft_size and [sdr].spectrum_interval_ms replace the compile-time FFT size and cadence; [[remotes]].spectrum_interval_ms lets the client ask for less. 512 bins at 5 frames/s is roughly 3.5 KB/s against roughly 200 KB/s before. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SyX26FCpMQxiBoC7r5K1A7 Signed-off-by: Stan Grams <sjg@haxx.space>
This commit is contained in:
Generated
+1
@@ -3165,6 +3165,7 @@ dependencies = [
|
|||||||
name = "trx-core"
|
name = "trx-core"
|
||||||
version = "0.1.0"
|
version = "0.1.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
|
"base64",
|
||||||
"flate2",
|
"flate2",
|
||||||
"reqwest",
|
"reqwest",
|
||||||
"serde",
|
"serde",
|
||||||
|
|||||||
@@ -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 |
|
| `sample_rate` | u32 | `1920000` | IQ capture rate in Hz |
|
||||||
| `bandwidth` | u32 | `1500000` | Hardware IF filter bandwidth in Hz |
|
| `bandwidth` | u32 | `1500000` | Hardware IF filter bandwidth in Hz |
|
||||||
| `center_offset_hz` | i64 | `100000` | Offset from dial to avoid DC spur |
|
| `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]`
|
#### `[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`) |
|
| `url` | string | — | Server address (e.g. `localhost:4530`) |
|
||||||
| `poll_interval_ms` | u64 | `750` | State poll interval |
|
| `poll_interval_ms` | u64 | `750` | State poll interval |
|
||||||
|
| `spectrum_interval_ms` | u64 | `50` | Spectrum frame interval; also settable per `[[remotes]]` entry |
|
||||||
|
|
||||||
#### `[remote.auth]`
|
#### `[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
|
The bridge is intended for WSJT-X integration via virtual audio devices (ALSA
|
||||||
loopback on Linux, BlackHole on macOS).
|
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
|
### CLI Override Summary
|
||||||
|
|
||||||
**trx-server:**
|
**trx-server:**
|
||||||
|
|||||||
@@ -288,6 +288,7 @@ async fn async_init() -> DynResult<AppState> {
|
|||||||
let token = cli.token.clone().or_else(|| cfg.remote.auth.token.clone());
|
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);
|
let poll_interval_ms = cli.poll_interval_ms.unwrap_or(cfg.remote.poll_interval_ms);
|
||||||
vec![RemoteEntry {
|
vec![RemoteEntry {
|
||||||
|
spectrum_interval_ms: cfg.remote.spectrum_interval_ms,
|
||||||
name,
|
name,
|
||||||
url: url.clone(),
|
url: url.clone(),
|
||||||
rig_id,
|
rig_id,
|
||||||
@@ -483,6 +484,13 @@ async fn async_init() -> DynResult<AppState> {
|
|||||||
.map(|e| e.poll_interval_ms)
|
.map(|e| e.poll_interval_ms)
|
||||||
.min()
|
.min()
|
||||||
.unwrap_or(750);
|
.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::<RigRequest>(RIG_TASK_CHANNEL_BUFFER);
|
let (server_tx, server_rx) = mpsc::channel::<RigRequest>(RIG_TASK_CHANNEL_BUFFER);
|
||||||
for entry in entries {
|
for entry in entries {
|
||||||
@@ -496,6 +504,8 @@ async fn async_init() -> DynResult<AppState> {
|
|||||||
known_rigs: frontend_runtime.routing.remote_rigs.clone(),
|
known_rigs: frontend_runtime.routing.remote_rigs.clone(),
|
||||||
rig_states: frontend_runtime.routing.rig_states.clone(),
|
rig_states: frontend_runtime.routing.rig_states.clone(),
|
||||||
poll_interval: Duration::from_millis(poll_interval),
|
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(),
|
spectrum: frontend_runtime.spectrum.sender.clone(),
|
||||||
rig_spectrums: frontend_runtime.spectrum.per_rig.clone(),
|
rig_spectrums: frontend_runtime.spectrum.per_rig.clone(),
|
||||||
server_connected: frontend_runtime.routing.server_connected.clone(),
|
server_connected: frontend_runtime.routing.server_connected.clone(),
|
||||||
|
|||||||
@@ -20,7 +20,7 @@ use trx_core::{RigError, RigResult};
|
|||||||
use trx_frontend::{RemoteRigEntry, SharedSpectrum};
|
use trx_frontend::{RemoteRigEntry, SharedSpectrum};
|
||||||
use trx_protocol::rig_command_to_client;
|
use trx_protocol::rig_command_to_client;
|
||||||
use trx_protocol::types::RigEntry;
|
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
|
// Endpoint parsing lives in `trx-config` so config validation and the
|
||||||
// connection code agree on what a URL means.
|
// 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_JSON_LINE_BYTES: usize = 256 * 1024;
|
||||||
const MAX_CONSECUTIVE_POLL_FAILURES: u32 = 3;
|
const MAX_CONSECUTIVE_POLL_FAILURES: u32 = 3;
|
||||||
|
|
||||||
// Keep remote spectrum reasonably responsive without returning to the old
|
// Default spectrum cadence when a config does not specify one. Both the push
|
||||||
// timeout churn caused by a much tighter request cadence.
|
// stream and the poll fallback run at the configured rate; see
|
||||||
const SPECTRUM_POLL_INTERVAL: Duration = Duration::from_millis(50);
|
// `[[remotes]].spectrum_interval_ms`.
|
||||||
|
pub const DEFAULT_SPECTRUM_INTERVAL: Duration = Duration::from_millis(50);
|
||||||
|
|
||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
pub struct RemoteClientConfig {
|
pub struct RemoteClientConfig {
|
||||||
@@ -43,6 +44,12 @@ pub struct RemoteClientConfig {
|
|||||||
pub selected_rig_id: Arc<Mutex<Option<String>>>,
|
pub selected_rig_id: Arc<Mutex<Option<String>>>,
|
||||||
pub known_rigs: Arc<Mutex<Vec<RemoteRigEntry>>>,
|
pub known_rigs: Arc<Mutex<Vec<RemoteRigEntry>>>,
|
||||||
pub poll_interval: Duration,
|
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<AtomicBool>,
|
||||||
/// Spectrum watch sender; spectrum task publishes here, SSE clients subscribe.
|
/// Spectrum watch sender; spectrum task publishes here, SSE clients subscribe.
|
||||||
pub spectrum: Arc<watch::Sender<SharedSpectrum>>,
|
pub spectrum: Arc<watch::Sender<SharedSpectrum>>,
|
||||||
/// Shared flag: `true` while a TCP connection to trx-server is active.
|
/// 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<bool>,
|
||||||
|
) -> RigResult<SpectrumStreamOutcome> {
|
||||||
|
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::<SpectrumFrame>(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(
|
async fn handle_spectrum_connection(
|
||||||
config: &RemoteClientConfig,
|
config: &RemoteClientConfig,
|
||||||
stream: TcpStream,
|
stream: TcpStream,
|
||||||
@@ -498,7 +599,33 @@ async fn handle_spectrum_connection(
|
|||||||
) -> RigResult<()> {
|
) -> RigResult<()> {
|
||||||
let (reader, mut writer) = stream.into_split();
|
let (reader, mut writer) = stream.into_split();
|
||||||
let mut reader = BufReader::new(reader);
|
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.
|
// Cache the token outside the poll loop to avoid cloning it every 50ms.
|
||||||
let cached_token = config.token.clone();
|
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::{has_short_names, resolve_server_rig_id, resolve_short_name};
|
||||||
use super::{
|
use super::{
|
||||||
parse_audio_url, parse_remote_url, RemoteClientConfig, RemoteEndpoint, SharedSpectrum,
|
parse_audio_url, parse_remote_url, RemoteClientConfig, RemoteEndpoint, SharedSpectrum,
|
||||||
|
DEFAULT_SPECTRUM_INTERVAL,
|
||||||
};
|
};
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
use std::sync::atomic::AtomicBool;
|
use std::sync::atomic::AtomicBool;
|
||||||
@@ -1381,6 +1509,8 @@ mod tests {
|
|||||||
selected_rig_id: Arc::new(Mutex::new(None)),
|
selected_rig_id: Arc::new(Mutex::new(None)),
|
||||||
known_rigs: Arc::new(Mutex::new(Vec::new())),
|
known_rigs: Arc::new(Mutex::new(Vec::new())),
|
||||||
poll_interval: Duration::from_millis(100),
|
poll_interval: Duration::from_millis(100),
|
||||||
|
spectrum_interval: DEFAULT_SPECTRUM_INTERVAL,
|
||||||
|
spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)),
|
||||||
spectrum: Arc::new(spectrum_tx),
|
spectrum: Arc::new(spectrum_tx),
|
||||||
server_connected: Arc::new(AtomicBool::new(false)),
|
server_connected: Arc::new(AtomicBool::new(false)),
|
||||||
rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
|
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()))),
|
selected_rig_id: Arc::new(Mutex::new(Some("sdr".to_string()))),
|
||||||
known_rigs: Arc::new(Mutex::new(Vec::new())),
|
known_rigs: Arc::new(Mutex::new(Vec::new())),
|
||||||
poll_interval: Duration::from_millis(500),
|
poll_interval: Duration::from_millis(500),
|
||||||
|
spectrum_interval: DEFAULT_SPECTRUM_INTERVAL,
|
||||||
|
spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)),
|
||||||
spectrum: Arc::new(spectrum_tx),
|
spectrum: Arc::new(spectrum_tx),
|
||||||
server_connected: Arc::new(AtomicBool::new(false)),
|
server_connected: Arc::new(AtomicBool::new(false)),
|
||||||
rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
|
rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
|
||||||
@@ -1441,6 +1573,130 @@ mod tests {
|
|||||||
assert_eq!(envelope.rig_id.as_deref(), Some("sdr"));
|
assert_eq!(envelope.rig_id.as_deref(), Some("sdr"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn stream_test_config(spectrum_tx: watch::Sender<SharedSpectrum>) -> 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]
|
#[test]
|
||||||
fn build_envelope_translates_short_name_to_server_rig_id() {
|
fn build_envelope_translates_short_name_to_server_rig_id() {
|
||||||
let (spectrum_tx, _spectrum_rx) = watch::channel(SharedSpectrum::default());
|
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()))),
|
selected_rig_id: Arc::new(Mutex::new(Some("home-hf".to_string()))),
|
||||||
known_rigs: Arc::new(Mutex::new(Vec::new())),
|
known_rigs: Arc::new(Mutex::new(Vec::new())),
|
||||||
poll_interval: Duration::from_millis(500),
|
poll_interval: Duration::from_millis(500),
|
||||||
|
spectrum_interval: DEFAULT_SPECTRUM_INTERVAL,
|
||||||
|
spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)),
|
||||||
spectrum: Arc::new(spectrum_tx),
|
spectrum: Arc::new(spectrum_tx),
|
||||||
server_connected: Arc::new(AtomicBool::new(false)),
|
server_connected: Arc::new(AtomicBool::new(false)),
|
||||||
rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
|
rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
|
||||||
@@ -1486,6 +1744,8 @@ mod tests {
|
|||||||
selected_rig_id: Arc::new(Mutex::new(None)),
|
selected_rig_id: Arc::new(Mutex::new(None)),
|
||||||
known_rigs: Arc::new(Mutex::new(Vec::new())),
|
known_rigs: Arc::new(Mutex::new(Vec::new())),
|
||||||
poll_interval: Duration::from_millis(500),
|
poll_interval: Duration::from_millis(500),
|
||||||
|
spectrum_interval: DEFAULT_SPECTRUM_INTERVAL,
|
||||||
|
spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)),
|
||||||
spectrum: Arc::new(spectrum_tx),
|
spectrum: Arc::new(spectrum_tx),
|
||||||
server_connected: Arc::new(AtomicBool::new(false)),
|
server_connected: Arc::new(AtomicBool::new(false)),
|
||||||
rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
|
rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
|
||||||
@@ -1510,6 +1770,8 @@ mod tests {
|
|||||||
selected_rig_id: Arc::new(Mutex::new(None)),
|
selected_rig_id: Arc::new(Mutex::new(None)),
|
||||||
known_rigs: Arc::new(Mutex::new(Vec::new())),
|
known_rigs: Arc::new(Mutex::new(Vec::new())),
|
||||||
poll_interval: Duration::from_millis(500),
|
poll_interval: Duration::from_millis(500),
|
||||||
|
spectrum_interval: DEFAULT_SPECTRUM_INTERVAL,
|
||||||
|
spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)),
|
||||||
spectrum: Arc::new(spectrum_tx),
|
spectrum: Arc::new(spectrum_tx),
|
||||||
server_connected: Arc::new(AtomicBool::new(false)),
|
server_connected: Arc::new(AtomicBool::new(false)),
|
||||||
rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
|
rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
|
||||||
@@ -1550,6 +1812,8 @@ mod tests {
|
|||||||
selected_rig_id: Arc::new(Mutex::new(None)),
|
selected_rig_id: Arc::new(Mutex::new(None)),
|
||||||
known_rigs: known_rigs.clone(),
|
known_rigs: known_rigs.clone(),
|
||||||
poll_interval: Duration::from_millis(500),
|
poll_interval: Duration::from_millis(500),
|
||||||
|
spectrum_interval: DEFAULT_SPECTRUM_INTERVAL,
|
||||||
|
spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)),
|
||||||
spectrum: Arc::new(spectrum_tx),
|
spectrum: Arc::new(spectrum_tx),
|
||||||
server_connected: Arc::new(AtomicBool::new(false)),
|
server_connected: Arc::new(AtomicBool::new(false)),
|
||||||
rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
|
rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
|
||||||
@@ -1623,6 +1887,8 @@ mod tests {
|
|||||||
selected_rig_id,
|
selected_rig_id,
|
||||||
known_rigs,
|
known_rigs,
|
||||||
poll_interval: Duration::from_millis(500),
|
poll_interval: Duration::from_millis(500),
|
||||||
|
spectrum_interval: DEFAULT_SPECTRUM_INTERVAL,
|
||||||
|
spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)),
|
||||||
spectrum: Arc::new(spectrum_tx),
|
spectrum: Arc::new(spectrum_tx),
|
||||||
server_connected: Arc::new(AtomicBool::new(false)),
|
server_connected: Arc::new(AtomicBool::new(false)),
|
||||||
rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
|
rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
|
||||||
|
|||||||
@@ -78,6 +78,10 @@ export type RdsData = { pi?: number | null, program_service?: string | null, rad
|
|||||||
export type SpectrumData = {
|
export type SpectrumData = {
|
||||||
/**
|
/**
|
||||||
* FFT magnitude bins in dBFS, FFT-shifted so DC (centre frequency) is at index N/2.
|
* 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<number>,
|
bins: Array<number>,
|
||||||
/**
|
/**
|
||||||
|
|||||||
@@ -76,6 +76,8 @@ pub struct RemoteConfig {
|
|||||||
pub auth: RemoteAuthConfig,
|
pub auth: RemoteAuthConfig,
|
||||||
/// Poll interval in milliseconds.
|
/// Poll interval in milliseconds.
|
||||||
pub poll_interval_ms: u64,
|
pub poll_interval_ms: u64,
|
||||||
|
/// Spectrum frame interval in milliseconds.
|
||||||
|
pub spectrum_interval_ms: u64,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Default for RemoteConfig {
|
impl Default for RemoteConfig {
|
||||||
@@ -85,6 +87,7 @@ impl Default for RemoteConfig {
|
|||||||
rig_id: None,
|
rig_id: None,
|
||||||
auth: RemoteAuthConfig::default(),
|
auth: RemoteAuthConfig::default(),
|
||||||
poll_interval_ms: 750,
|
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.
|
/// Poll interval in milliseconds. Defaults to 750.
|
||||||
#[serde(default = "default_poll_interval_ms")]
|
#[serde(default = "default_poll_interval_ms")]
|
||||||
pub poll_interval_ms: u64,
|
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 {
|
fn default_poll_interval_ms() -> u64 {
|
||||||
750
|
750
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn default_spectrum_interval_ms() -> u64 {
|
||||||
|
50
|
||||||
|
}
|
||||||
|
|
||||||
/// Frontend configurations.
|
/// Frontend configurations.
|
||||||
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
|
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
@@ -421,6 +435,7 @@ impl ClientConfig {
|
|||||||
rig_id: self.remote.rig_id.clone(),
|
rig_id: self.remote.rig_id.clone(),
|
||||||
auth: self.remote.auth.clone(),
|
auth: self.remote.auth.clone(),
|
||||||
poll_interval_ms: self.remote.poll_interval_ms,
|
poll_interval_ms: self.remote.poll_interval_ms,
|
||||||
|
spectrum_interval_ms: self.remote.spectrum_interval_ms,
|
||||||
}]
|
}]
|
||||||
} else {
|
} else {
|
||||||
Vec::new()
|
Vec::new()
|
||||||
@@ -603,12 +618,21 @@ impl ClientConfig {
|
|||||||
i, entry.name
|
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.
|
// Legacy [remote], kept for backward compatibility.
|
||||||
if self.remote.poll_interval_ms == 0 {
|
if self.remote.poll_interval_ms == 0 {
|
||||||
return Err("[remote].poll_interval_ms must be > 0".to_string());
|
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 let Some(url) = &self.remote.url {
|
||||||
if url.trim().is_empty() {
|
if url.trim().is_empty() {
|
||||||
return Err("[remote].url must not be empty when set".to_string());
|
return Err("[remote].url must not be empty when set".to_string());
|
||||||
@@ -834,6 +858,7 @@ impl ClientConfig {
|
|||||||
token_file: None,
|
token_file: None,
|
||||||
},
|
},
|
||||||
poll_interval_ms: 750,
|
poll_interval_ms: 750,
|
||||||
|
spectrum_interval_ms: 50,
|
||||||
},
|
},
|
||||||
RemoteEntry {
|
RemoteEntry {
|
||||||
name: "home-vhf".to_string(),
|
name: "home-vhf".to_string(),
|
||||||
@@ -844,6 +869,7 @@ impl ClientConfig {
|
|||||||
token_file: None,
|
token_file: None,
|
||||||
},
|
},
|
||||||
poll_interval_ms: 750,
|
poll_interval_ms: 750,
|
||||||
|
spectrum_interval_ms: 50,
|
||||||
},
|
},
|
||||||
],
|
],
|
||||||
frontends: FrontendsConfig {
|
frontends: FrontendsConfig {
|
||||||
@@ -994,6 +1020,7 @@ impl ConfigFile for ClientConfig {
|
|||||||
rig_id: Some("hf".to_string()),
|
rig_id: Some("hf".to_string()),
|
||||||
auth: RemoteAuthConfig::default(),
|
auth: RemoteAuthConfig::default(),
|
||||||
poll_interval_ms: default_poll_interval_ms(),
|
poll_interval_ms: default_poll_interval_ms(),
|
||||||
|
spectrum_interval_ms: 50,
|
||||||
}],
|
}],
|
||||||
..Default::default()
|
..Default::default()
|
||||||
};
|
};
|
||||||
@@ -1311,6 +1338,7 @@ url = "remote.example.com:4530"
|
|||||||
rig_id: None,
|
rig_id: None,
|
||||||
auth: RemoteAuthConfig::default(),
|
auth: RemoteAuthConfig::default(),
|
||||||
poll_interval_ms: 750,
|
poll_interval_ms: 750,
|
||||||
|
spectrum_interval_ms: 50,
|
||||||
}],
|
}],
|
||||||
..Default::default()
|
..Default::default()
|
||||||
};
|
};
|
||||||
@@ -1330,6 +1358,7 @@ url = "remote.example.com:4530"
|
|||||||
token_file: None,
|
token_file: None,
|
||||||
},
|
},
|
||||||
poll_interval_ms: 750,
|
poll_interval_ms: 750,
|
||||||
|
spectrum_interval_ms: 50,
|
||||||
},
|
},
|
||||||
..Default::default()
|
..Default::default()
|
||||||
};
|
};
|
||||||
@@ -1368,6 +1397,7 @@ url = "remote.example.com:4530"
|
|||||||
rig_id: None,
|
rig_id: None,
|
||||||
auth: RemoteAuthConfig::default(),
|
auth: RemoteAuthConfig::default(),
|
||||||
poll_interval_ms: 750,
|
poll_interval_ms: 750,
|
||||||
|
spectrum_interval_ms: 50,
|
||||||
}],
|
}],
|
||||||
..Default::default()
|
..Default::default()
|
||||||
};
|
};
|
||||||
@@ -1386,6 +1416,7 @@ url = "remote.example.com:4530"
|
|||||||
rig_id: None,
|
rig_id: None,
|
||||||
auth: RemoteAuthConfig::default(),
|
auth: RemoteAuthConfig::default(),
|
||||||
poll_interval_ms: 750,
|
poll_interval_ms: 750,
|
||||||
|
spectrum_interval_ms: 50,
|
||||||
},
|
},
|
||||||
RemoteEntry {
|
RemoteEntry {
|
||||||
name: "dup".to_string(),
|
name: "dup".to_string(),
|
||||||
@@ -1393,6 +1424,7 @@ url = "remote.example.com:4530"
|
|||||||
rig_id: None,
|
rig_id: None,
|
||||||
auth: RemoteAuthConfig::default(),
|
auth: RemoteAuthConfig::default(),
|
||||||
poll_interval_ms: 750,
|
poll_interval_ms: 750,
|
||||||
|
spectrum_interval_ms: 50,
|
||||||
},
|
},
|
||||||
],
|
],
|
||||||
..Default::default()
|
..Default::default()
|
||||||
@@ -1409,6 +1441,7 @@ url = "remote.example.com:4530"
|
|||||||
rig_id: None,
|
rig_id: None,
|
||||||
auth: RemoteAuthConfig::default(),
|
auth: RemoteAuthConfig::default(),
|
||||||
poll_interval_ms: 750,
|
poll_interval_ms: 750,
|
||||||
|
spectrum_interval_ms: 50,
|
||||||
}],
|
}],
|
||||||
..Default::default()
|
..Default::default()
|
||||||
};
|
};
|
||||||
@@ -1427,6 +1460,7 @@ url = "remote.example.com:4530"
|
|||||||
rig_id: None,
|
rig_id: None,
|
||||||
auth: RemoteAuthConfig::default(),
|
auth: RemoteAuthConfig::default(),
|
||||||
poll_interval_ms: 750,
|
poll_interval_ms: 750,
|
||||||
|
spectrum_interval_ms: 50,
|
||||||
}],
|
}],
|
||||||
..Default::default()
|
..Default::default()
|
||||||
};
|
};
|
||||||
@@ -1445,6 +1479,7 @@ url = "remote.example.com:4530"
|
|||||||
rig_id: None,
|
rig_id: None,
|
||||||
auth: RemoteAuthConfig::default(),
|
auth: RemoteAuthConfig::default(),
|
||||||
poll_interval_ms: 0,
|
poll_interval_ms: 0,
|
||||||
|
spectrum_interval_ms: 50,
|
||||||
}],
|
}],
|
||||||
..Default::default()
|
..Default::default()
|
||||||
};
|
};
|
||||||
@@ -1499,6 +1534,45 @@ url = "remote.example.com:4530"
|
|||||||
assert!(warnings.is_empty(), "unexpected warnings: {warnings:?}");
|
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 ---
|
// --- Secret indirection ---
|
||||||
|
|
||||||
fn secret_file(content: &str) -> tempfile::NamedTempFile {
|
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()),
|
token_file: Some(f.path().to_str().unwrap().to_string()),
|
||||||
},
|
},
|
||||||
poll_interval_ms: 750,
|
poll_interval_ms: 750,
|
||||||
|
spectrum_interval_ms: 50,
|
||||||
}],
|
}],
|
||||||
..Default::default()
|
..Default::default()
|
||||||
};
|
};
|
||||||
@@ -1570,6 +1645,7 @@ url = "remote.example.com:4530"
|
|||||||
rig_id: None,
|
rig_id: None,
|
||||||
auth: RemoteAuthConfig::default(),
|
auth: RemoteAuthConfig::default(),
|
||||||
poll_interval_ms: 750,
|
poll_interval_ms: 750,
|
||||||
|
spectrum_interval_ms: 50,
|
||||||
})
|
})
|
||||||
.collect()
|
.collect()
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -399,12 +399,30 @@ pub struct SdrConfig {
|
|||||||
/// Default: 4.
|
/// Default: 4.
|
||||||
#[serde(default = "default_max_virtual_channels")]
|
#[serde(default = "default_max_virtual_channels")]
|
||||||
pub max_virtual_channels: usize,
|
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 {
|
fn default_max_virtual_channels() -> usize {
|
||||||
4
|
4
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn default_spectrum_fft_size() -> usize {
|
||||||
|
1024
|
||||||
|
}
|
||||||
|
|
||||||
|
fn default_spectrum_interval_ms() -> u64 {
|
||||||
|
50
|
||||||
|
}
|
||||||
|
|
||||||
impl Default for SdrConfig {
|
impl Default for SdrConfig {
|
||||||
fn default() -> Self {
|
fn default() -> Self {
|
||||||
Self {
|
Self {
|
||||||
@@ -417,6 +435,8 @@ impl Default for SdrConfig {
|
|||||||
noise_blanker: SdrNoiseBlankerConfig::default(),
|
noise_blanker: SdrNoiseBlankerConfig::default(),
|
||||||
channels: Vec::new(),
|
channels: Vec::new(),
|
||||||
max_virtual_channels: default_max_virtual_channels(),
|
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"));
|
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)
|
if let Err(e) = validate_sdr_squelch_config(&format!("{prefix}[sdr.squelch]"), &rig.sdr.squelch)
|
||||||
{
|
{
|
||||||
errors.push(e);
|
errors.push(e);
|
||||||
@@ -1070,6 +1093,26 @@ fn validate_access(prefix: &str, access: &AccessConfig) -> Result<(), String> {
|
|||||||
Ok(())
|
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> {
|
fn validate_sdr_squelch_config(path: &str, squelch: &SdrSquelchConfig) -> Result<(), String> {
|
||||||
if !squelch.threshold_db.is_finite() {
|
if !squelch.threshold_db.is_finite() {
|
||||||
return Err(format!("{path}.threshold_db must be finite"));
|
return Err(format!("{path}.threshold_db must be finite"));
|
||||||
@@ -1813,6 +1856,93 @@ port = 4531
|
|||||||
assert!(warnings.is_empty(), "unexpected warnings: {warnings:?}");
|
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 ---
|
// --- Secret indirection ---
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
|
|||||||
@@ -17,3 +17,4 @@ uuid = { workspace = true }
|
|||||||
ts-rs = { version = "12.0.1", features = ["uuid-impl"] }
|
ts-rs = { version = "12.0.1", features = ["uuid-impl"] }
|
||||||
sgp4 = "2"
|
sgp4 = "2"
|
||||||
reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] }
|
reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] }
|
||||||
|
base64 = "0.22"
|
||||||
|
|||||||
@@ -19,6 +19,7 @@ pub mod command;
|
|||||||
pub mod controller;
|
pub mod controller;
|
||||||
pub mod request;
|
pub mod request;
|
||||||
pub mod response;
|
pub mod response;
|
||||||
|
pub mod spectrum_wire;
|
||||||
pub mod state;
|
pub mod state;
|
||||||
|
|
||||||
/// How this backend communicates with the rig.
|
/// How this backend communicates with the rig.
|
||||||
|
|||||||
@@ -0,0 +1,124 @@
|
|||||||
|
// SPDX-FileCopyrightText: 2026 Stan Grams <sjg@haxx.space>
|
||||||
|
//
|
||||||
|
// 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<S: Serializer>(bins: &[f32], serializer: S) -> Result<S::Ok, S::Error> {
|
||||||
|
let quantized: Vec<u8> = 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<Vec<f32>, D::Error> {
|
||||||
|
deserializer.deserialize_any(BinsVisitor)
|
||||||
|
}
|
||||||
|
|
||||||
|
struct BinsVisitor;
|
||||||
|
|
||||||
|
impl<'de> Visitor<'de> for BinsVisitor {
|
||||||
|
type Value = Vec<f32>;
|
||||||
|
|
||||||
|
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<E: serde::de::Error>(self, value: &str) -> Result<Self::Value, E> {
|
||||||
|
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<A: SeqAccess<'de>>(self, mut seq: A) -> Result<Self::Value, A::Error> {
|
||||||
|
let mut bins = Vec::with_capacity(seq.size_hint().unwrap_or(1024));
|
||||||
|
while let Some(value) = seq.next_element::<f32>()? {
|
||||||
|
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<f32>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[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<f32> = (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()
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -394,6 +394,12 @@ fn default_wfm_denoise_level() -> WfmDenoiseLevel {
|
|||||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, TS)]
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, TS)]
|
||||||
pub struct SpectrumData {
|
pub struct SpectrumData {
|
||||||
/// FFT magnitude bins in dBFS, FFT-shifted so DC (centre frequency) is at index N/2.
|
/// 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<number>")]
|
||||||
pub bins: Vec<f32>,
|
pub bins: Vec<f32>,
|
||||||
/// Centre frequency of the SDR capture in Hz.
|
/// Centre frequency of the SDR capture in Hz.
|
||||||
#[ts(type = "number")]
|
#[ts(type = "number")]
|
||||||
|
|||||||
@@ -18,4 +18,6 @@ pub use auth::{NoAuthValidator, SimpleTokenValidator, TokenValidator};
|
|||||||
pub use codec::{mode_to_string, parse_envelope, parse_mode};
|
pub use codec::{mode_to_string, parse_envelope, parse_mode};
|
||||||
pub use decoders::{DecoderActivation, DecoderDescriptor, DECODER_REGISTRY};
|
pub use decoders::{DecoderActivation, DecoderDescriptor, DECODER_REGISTRY};
|
||||||
pub use mapping::{client_command_to_rig, rig_command_to_client};
|
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,
|
||||||
|
};
|
||||||
|
|||||||
@@ -102,7 +102,7 @@ macro_rules! define_command_mapping {
|
|||||||
|
|
||||||
define_command_mapping! {
|
define_command_mapping! {
|
||||||
// ── Client-only variants (no RigCommand counterpart) ─────────────
|
// ── Client-only variants (no RigCommand counterpart) ─────────────
|
||||||
client_only: GetRigs, GetSatPasses, SubscribeMeter;
|
client_only: GetRigs, GetSatPasses, SubscribeMeter, SubscribeSpectrum;
|
||||||
|
|
||||||
// ── Unit variants (no payload) ───────────────────────────────────
|
// ── Unit variants (no payload) ───────────────────────────────────
|
||||||
unit:
|
unit:
|
||||||
|
|||||||
@@ -130,6 +130,28 @@ pub enum ClientCommand {
|
|||||||
/// newline-delimited `MeterUpdate` JSON frames and no further commands or
|
/// newline-delimited `MeterUpdate` JSON frames and no further commands or
|
||||||
/// regular responses are sent. Intended for a dedicated TCP connection.
|
/// regular responses are sent. Intended for a dedicated TCP connection.
|
||||||
SubscribeMeter,
|
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<Vec<trx_core::rig::state::VchanRdsEntry>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Fast meter sample pushed by the server on a dedicated meter stream.
|
/// Fast meter sample pushed by the server on a dedicated meter stream.
|
||||||
|
|||||||
@@ -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.
|
// SubscribeMeter: turns this connection into a one-way meter stream.
|
||||||
// No regular responses are produced; the connection lives until the
|
// No regular responses are produced; the connection lives until the
|
||||||
// client disconnects or shutdown fires.
|
// client disconnects or shutdown fires.
|
||||||
@@ -732,6 +784,7 @@ mod tests {
|
|||||||
let (state_tx, state_rx) = watch::channel(state);
|
let (state_tx, state_rx) = watch::channel(state);
|
||||||
let _state_tx = state_tx;
|
let _state_tx = state_tx;
|
||||||
let (meter_tx, _) = tokio::sync::broadcast::channel(8);
|
let (meter_tx, _) = tokio::sync::broadcast::channel(8);
|
||||||
|
let (spectrum_tx, _) = tokio::sync::broadcast::channel(4);
|
||||||
let handle = RigHandle {
|
let handle = RigHandle {
|
||||||
rig_id: "default".to_string(),
|
rig_id: "default".to_string(),
|
||||||
display_name: "Default Rig".to_string(),
|
display_name: "Default Rig".to_string(),
|
||||||
@@ -739,6 +792,7 @@ mod tests {
|
|||||||
state_rx,
|
state_rx,
|
||||||
audio_port: 4531,
|
audio_port: 4531,
|
||||||
meter_tx,
|
meter_tx,
|
||||||
|
spectrum_tx,
|
||||||
};
|
};
|
||||||
let mut map = HashMap::new();
|
let mut map = HashMap::new();
|
||||||
map.insert("default".to_string(), handle);
|
map.insert("default".to_string(), handle);
|
||||||
@@ -933,6 +987,7 @@ mod tests {
|
|||||||
state_rx: state_rx_a,
|
state_rx: state_rx_a,
|
||||||
audio_port: 4531,
|
audio_port: 4531,
|
||||||
meter_tx: meter_tx_a,
|
meter_tx: meter_tx_a,
|
||||||
|
spectrum_tx: tokio::sync::broadcast::channel(4).0,
|
||||||
};
|
};
|
||||||
|
|
||||||
let (tx_b, rx_b) = mpsc::channel::<RigRequest>(8);
|
let (tx_b, rx_b) = mpsc::channel::<RigRequest>(8);
|
||||||
@@ -945,6 +1000,7 @@ mod tests {
|
|||||||
state_rx: state_rx_b,
|
state_rx: state_rx_b,
|
||||||
audio_port: 4532,
|
audio_port: 4532,
|
||||||
meter_tx: meter_tx_b,
|
meter_tx: meter_tx_b,
|
||||||
|
spectrum_tx: tokio::sync::broadcast::channel(4).0,
|
||||||
};
|
};
|
||||||
|
|
||||||
let mut map = HashMap::new();
|
let mut map = HashMap::new();
|
||||||
@@ -953,6 +1009,67 @@ mod tests {
|
|||||||
(Arc::new(map), "rig_hf".to_string(), rx_a, rx_b)
|
(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]
|
#[tokio::test]
|
||||||
async fn multi_rig_state_isolation() {
|
async fn multi_rig_state_isolation() {
|
||||||
let state_hf = sample_state_custom("HF-Dummy", 14_200_000, trx_core::RigMode::USB);
|
let state_hf = sample_state_custom("HF-Dummy", 14_200_000, trx_core::RigMode::USB);
|
||||||
|
|||||||
@@ -360,6 +360,7 @@ fn build_sdr_rig_from_instance(rig_cfg: &RigInstanceConfig) -> SdrRigBuildResult
|
|||||||
max_virtual_channels: rig_cfg.sdr.max_virtual_channels,
|
max_virtual_channels: rig_cfg.sdr.max_virtual_channels,
|
||||||
nb_enabled: rig_cfg.sdr.noise_blanker.enabled,
|
nb_enabled: rig_cfg.sdr.noise_blanker.enabled,
|
||||||
nb_threshold: rig_cfg.sdr.noise_blanker.threshold,
|
nb_threshold: rig_cfg.sdr.noise_blanker.threshold,
|
||||||
|
spectrum_fft_size: rig_cfg.sdr.spectrum_fft_size,
|
||||||
})?;
|
})?;
|
||||||
|
|
||||||
let pcm_rx = sdr_rig.subscribe_pcm();
|
let pcm_rx = sdr_rig.subscribe_pcm();
|
||||||
@@ -460,6 +461,7 @@ fn build_rig_task_config(
|
|||||||
prebuilt_rig: None,
|
prebuilt_rig: None,
|
||||||
command_exec_timeout: Duration::from_millis(timeouts.command_exec_timeout_ms),
|
command_exec_timeout: Duration::from_millis(timeouts.command_exec_timeout_ms),
|
||||||
poll_refresh_timeout: Duration::from_millis(timeouts.poll_refresh_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 (state_tx, state_rx) = watch::channel(initial_state);
|
||||||
let (meter_tx, _) =
|
let (meter_tx, _) =
|
||||||
broadcast::channel::<trx_protocol::MeterUpdate>(rig_handle::METER_BROADCAST_CAPACITY);
|
broadcast::channel::<trx_protocol::MeterUpdate>(rig_handle::METER_BROADCAST_CAPACITY);
|
||||||
|
let (spectrum_tx, _) = broadcast::channel::<trx_protocol::SpectrumFrame>(
|
||||||
|
rig_handle::SPECTRUM_BROADCAST_CAPACITY,
|
||||||
|
);
|
||||||
|
|
||||||
let mut task_config = build_rig_task_config(
|
let mut task_config = build_rig_task_config(
|
||||||
rig_cfg,
|
rig_cfg,
|
||||||
@@ -1242,12 +1247,14 @@ async fn main() -> DynResult<()> {
|
|||||||
let rig_shutdown_rx = shutdown_rx.clone();
|
let rig_shutdown_rx = shutdown_rx.clone();
|
||||||
let rig_id_supervisor = rig_cfg.id.clone();
|
let rig_id_supervisor = rig_cfg.id.clone();
|
||||||
let meter_tx_task = meter_tx.clone();
|
let meter_tx_task = meter_tx.clone();
|
||||||
|
let spectrum_tx_task = spectrum_tx.clone();
|
||||||
task_handles.push(tokio::spawn(async move {
|
task_handles.push(tokio::spawn(async move {
|
||||||
let result = rig_task::run_rig_task(
|
let result = rig_task::run_rig_task(
|
||||||
task_config,
|
task_config,
|
||||||
rig_rx,
|
rig_rx,
|
||||||
state_tx.clone(),
|
state_tx.clone(),
|
||||||
meter_tx_task,
|
meter_tx_task,
|
||||||
|
spectrum_tx_task,
|
||||||
rig_shutdown_rx,
|
rig_shutdown_rx,
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
@@ -1294,6 +1301,7 @@ async fn main() -> DynResult<()> {
|
|||||||
rig_handles.insert(
|
rig_handles.insert(
|
||||||
rig_cfg.id.clone(),
|
rig_cfg.id.clone(),
|
||||||
RigHandle {
|
RigHandle {
|
||||||
|
spectrum_tx: spectrum_tx.clone(),
|
||||||
rig_id: rig_cfg.id.clone(),
|
rig_id: rig_cfg.id.clone(),
|
||||||
display_name: rig_cfg.display_name().to_string(),
|
display_name: rig_cfg.display_name().to_string(),
|
||||||
rig_tx,
|
rig_tx,
|
||||||
|
|||||||
@@ -8,13 +8,18 @@ use tokio::sync::{broadcast, mpsc, watch};
|
|||||||
|
|
||||||
use trx_core::rig::request::RigRequest;
|
use trx_core::rig::request::RigRequest;
|
||||||
use trx_core::rig::state::RigState;
|
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
|
/// 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
|
/// samples at 30 Hz — more than enough slack to tolerate a scheduling blip
|
||||||
/// without forcing the producer to block or drop silently.
|
/// without forcing the producer to block or drop silently.
|
||||||
pub const METER_BROADCAST_CAPACITY: usize = 16;
|
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.
|
/// A handle to a single running rig backend.
|
||||||
///
|
///
|
||||||
/// One `RigHandle` is created per rig in `main.rs` and stored in the shared
|
/// 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
|
/// ~6–7 Hz (CAT). Consumed by `SubscribeMeter` clients; independent of
|
||||||
/// the slower `state_rx` snapshot path.
|
/// the slower `state_rx` snapshot path.
|
||||||
pub meter_tx: broadcast::Sender<MeterUpdate>,
|
pub meter_tx: broadcast::Sender<MeterUpdate>,
|
||||||
|
/// 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<SpectrumFrame>,
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -23,7 +23,7 @@ use trx_core::rig::request::RigRequest;
|
|||||||
use trx_core::rig::state::{RigMode, RigSnapshot, RigState};
|
use trx_core::rig::state::{RigMode, RigSnapshot, RigState};
|
||||||
use trx_core::rig::{RigCat, RigRxStatus, RigTxStatus};
|
use trx_core::rig::{RigCat, RigRxStatus, RigTxStatus};
|
||||||
use trx_core::{DynResult, RigError, RigResult};
|
use trx_core::{DynResult, RigError, RigResult};
|
||||||
use trx_protocol::MeterUpdate;
|
use trx_protocol::{MeterUpdate, SpectrumFrame};
|
||||||
|
|
||||||
use crate::audio::DecoderHistories;
|
use crate::audio::DecoderHistories;
|
||||||
use crate::error::is_invalid_bcd_error;
|
use crate::error::is_invalid_bcd_error;
|
||||||
@@ -64,6 +64,8 @@ pub struct RigTaskConfig {
|
|||||||
pub command_exec_timeout: Duration,
|
pub command_exec_timeout: Duration,
|
||||||
/// Maximum time for a CAT poll refresh cycle.
|
/// Maximum time for a CAT poll refresh cycle.
|
||||||
pub poll_refresh_timeout: Duration,
|
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 {
|
impl Default for RigTaskConfig {
|
||||||
@@ -94,6 +96,7 @@ impl Default for RigTaskConfig {
|
|||||||
prebuilt_rig: None,
|
prebuilt_rig: None,
|
||||||
command_exec_timeout: DEFAULT_COMMAND_EXEC_TIMEOUT,
|
command_exec_timeout: DEFAULT_COMMAND_EXEC_TIMEOUT,
|
||||||
poll_refresh_timeout: DEFAULT_POLL_REFRESH_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<RigRequest>,
|
mut rx: mpsc::Receiver<RigRequest>,
|
||||||
state_tx: watch::Sender<RigState>,
|
state_tx: watch::Sender<RigState>,
|
||||||
meter_tx: broadcast::Sender<MeterUpdate>,
|
meter_tx: broadcast::Sender<MeterUpdate>,
|
||||||
|
spectrum_tx: broadcast::Sender<SpectrumFrame>,
|
||||||
mut shutdown_rx: watch::Receiver<bool>,
|
mut shutdown_rx: watch::Receiver<bool>,
|
||||||
) -> DynResult<()> {
|
) -> DynResult<()> {
|
||||||
let histories = config.histories.clone();
|
let histories = config.histories.clone();
|
||||||
@@ -273,6 +277,13 @@ pub async fn run_rig_task(
|
|||||||
} else {
|
} else {
|
||||||
Duration::from_millis(150)
|
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<tokio::time::Sleep>> =
|
||||||
|
Box::pin(tokio::time::sleep(spectrum_tick_duration));
|
||||||
|
|
||||||
let meter_task_start = Instant::now();
|
let meter_task_start = Instant::now();
|
||||||
let meter_state_delta_db: f64 = 0.25;
|
let meter_state_delta_db: f64 = 0.25;
|
||||||
let rig_id = config.rig_id.clone();
|
let rig_id = config.rig_id.clone();
|
||||||
@@ -302,6 +313,21 @@ pub async fn run_rig_task(
|
|||||||
Err(_) => break,
|
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.
|
// Fast meter-only refresh between full polls.
|
||||||
_ = &mut meter_tick => {
|
_ = &mut meter_tick => {
|
||||||
meter_tick = Box::pin(tokio::time::sleep(meter_tick_duration));
|
meter_tick = Box::pin(tokio::time::sleep(meter_tick_duration));
|
||||||
|
|||||||
@@ -154,6 +154,7 @@ impl SdrPipeline {
|
|||||||
squelch_cfg: VirtualSquelchConfig,
|
squelch_cfg: VirtualSquelchConfig,
|
||||||
nb_cfg: NoiseBlankerConfig,
|
nb_cfg: NoiseBlankerConfig,
|
||||||
channels: &[(f64, RigMode, u32)],
|
channels: &[(f64, RigMode, u32)],
|
||||||
|
spectrum_fft_size: usize,
|
||||||
) -> Self {
|
) -> Self {
|
||||||
const IQ_BROADCAST_CAPACITY: usize = 64;
|
const IQ_BROADCAST_CAPACITY: usize = 64;
|
||||||
let (iq_tx, _iq_rx) = broadcast::channel::<Vec<Complex<f32>>>(IQ_BROADCAST_CAPACITY);
|
let (iq_tx, _iq_rx) = broadcast::channel::<Vec<Complex<f32>>>(IQ_BROADCAST_CAPACITY);
|
||||||
@@ -219,6 +220,7 @@ impl SdrPipeline {
|
|||||||
.name("sdr-iq-read".to_string())
|
.name("sdr-iq-read".to_string())
|
||||||
.spawn(move || {
|
.spawn(move || {
|
||||||
iq_read_loop(
|
iq_read_loop(
|
||||||
|
spectrum_fft_size,
|
||||||
source,
|
source,
|
||||||
sdr_sample_rate,
|
sdr_sample_rate,
|
||||||
thread_dsps,
|
thread_dsps,
|
||||||
@@ -317,6 +319,7 @@ pub const IQ_BLOCK_SIZE: usize = 4096;
|
|||||||
|
|
||||||
#[allow(clippy::too_many_arguments)]
|
#[allow(clippy::too_many_arguments)]
|
||||||
fn iq_read_loop(
|
fn iq_read_loop(
|
||||||
|
spectrum_fft_size: usize,
|
||||||
mut source: Box<dyn IqSource>,
|
mut source: Box<dyn IqSource>,
|
||||||
sdr_sample_rate: u32,
|
sdr_sample_rate: u32,
|
||||||
channel_dsps: Arc<RwLock<Vec<Arc<Mutex<ChannelDsp>>>>>,
|
channel_dsps: Arc<RwLock<Vec<Arc<Mutex<ChannelDsp>>>>>,
|
||||||
@@ -335,7 +338,7 @@ fn iq_read_loop(
|
|||||||
};
|
};
|
||||||
let throttle = !source.is_blocking();
|
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 read_error_streak: u32 = 0;
|
||||||
let mut zero_read_streak: u32 = 0;
|
let mut zero_read_streak: u32 = 0;
|
||||||
let mut overflow_log_window_start: Option<Instant> = None;
|
let mut overflow_log_window_start: Option<Instant> = None;
|
||||||
@@ -572,6 +575,7 @@ mod tests {
|
|||||||
VirtualSquelchConfig::default(),
|
VirtualSquelchConfig::default(),
|
||||||
NoiseBlankerConfig::default(),
|
NoiseBlankerConfig::default(),
|
||||||
&[(200_000.0, RigMode::USB, 3000)],
|
&[(200_000.0, RigMode::USB, 3000)],
|
||||||
|
1024,
|
||||||
);
|
);
|
||||||
assert_eq!(pipeline.pcm_senders.len(), 1);
|
assert_eq!(pipeline.pcm_senders.len(), 1);
|
||||||
assert_eq!(pipeline.channel_dsps.read().unwrap().len(), 1);
|
assert_eq!(pipeline.channel_dsps.read().unwrap().len(), 1);
|
||||||
@@ -590,6 +594,7 @@ mod tests {
|
|||||||
VirtualSquelchConfig::default(),
|
VirtualSquelchConfig::default(),
|
||||||
NoiseBlankerConfig::default(),
|
NoiseBlankerConfig::default(),
|
||||||
&[],
|
&[],
|
||||||
|
1024,
|
||||||
);
|
);
|
||||||
assert_eq!(pipeline.pcm_senders.len(), 0);
|
assert_eq!(pipeline.pcm_senders.len(), 0);
|
||||||
assert_eq!(pipeline.channel_dsps.read().unwrap().len(), 0);
|
assert_eq!(pipeline.channel_dsps.read().unwrap().len(), 0);
|
||||||
|
|||||||
@@ -9,28 +9,36 @@ use num_complex::Complex;
|
|||||||
use rustfft::num_complex::Complex as FftComplex;
|
use rustfft::num_complex::Complex as FftComplex;
|
||||||
use rustfft::FftPlanner;
|
use rustfft::FftPlanner;
|
||||||
|
|
||||||
/// Number of FFT bins for the spectrum display.
|
/// Default number of FFT bins for the spectrum display, used when the config
|
||||||
pub(super) const SPECTRUM_FFT_SIZE: usize = 1024;
|
/// 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).
|
/// 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) const SPECTRUM_UPDATE_BLOCKS: usize = 4;
|
||||||
|
|
||||||
pub(super) struct SpectrumSnapshotter {
|
pub(super) struct SpectrumSnapshotter {
|
||||||
|
fft_size: usize,
|
||||||
hann_window: Vec<f32>,
|
hann_window: Vec<f32>,
|
||||||
fft: std::sync::Arc<dyn rustfft::Fft<f32>>,
|
fft: std::sync::Arc<dyn rustfft::Fft<f32>>,
|
||||||
counter: usize,
|
counter: usize,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl SpectrumSnapshotter {
|
impl SpectrumSnapshotter {
|
||||||
pub(super) fn new() -> Self {
|
pub(super) fn new(fft_size: usize) -> Self {
|
||||||
let hann_window: Vec<f32> = (0..SPECTRUM_FFT_SIZE)
|
let fft_size = if fft_size >= 2 {
|
||||||
.map(|i| 0.5 * (1.0 - (2.0 * PI * i as f32 / (SPECTRUM_FFT_SIZE - 1) as f32).cos()))
|
fft_size
|
||||||
|
} else {
|
||||||
|
DEFAULT_SPECTRUM_FFT_SIZE
|
||||||
|
};
|
||||||
|
let hann_window: Vec<f32> = (0..fft_size)
|
||||||
|
.map(|i| 0.5 * (1.0 - (2.0 * PI * i as f32 / (fft_size - 1) as f32).cos()))
|
||||||
.collect();
|
.collect();
|
||||||
|
|
||||||
let mut planner = FftPlanner::<f32>::new();
|
let mut planner = FftPlanner::<f32>::new();
|
||||||
let fft = planner.plan_fft_forward(SPECTRUM_FFT_SIZE);
|
let fft = planner.plan_fft_forward(fft_size);
|
||||||
|
|
||||||
Self {
|
Self {
|
||||||
|
fft_size,
|
||||||
hann_window,
|
hann_window,
|
||||||
fft,
|
fft,
|
||||||
counter: 0,
|
counter: 0,
|
||||||
@@ -48,7 +56,7 @@ impl SpectrumSnapshotter {
|
|||||||
}
|
}
|
||||||
self.counter = 0;
|
self.counter = 0;
|
||||||
|
|
||||||
let take = samples.len().min(SPECTRUM_FFT_SIZE);
|
let take = samples.len().min(self.fft_size);
|
||||||
let mut buf: Vec<FftComplex<f32>> = samples[..take]
|
let mut buf: Vec<FftComplex<f32>> = samples[..take]
|
||||||
.iter()
|
.iter()
|
||||||
.enumerate()
|
.enumerate()
|
||||||
@@ -59,16 +67,15 @@ impl SpectrumSnapshotter {
|
|||||||
)
|
)
|
||||||
})
|
})
|
||||||
.collect();
|
.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);
|
self.fft.process(&mut buf);
|
||||||
|
|
||||||
let half = SPECTRUM_FFT_SIZE / 2;
|
let half = self.fft_size / 2;
|
||||||
let bins: Vec<f32> = buf[half..]
|
let bins: Vec<f32> = buf[half..]
|
||||||
.iter()
|
.iter()
|
||||||
.chain(buf[..half].iter())
|
.chain(buf[..half].iter())
|
||||||
.map(|value| {
|
.map(|value| {
|
||||||
let mag =
|
let mag = (value.re * value.re + value.im * value.im).sqrt() / self.fft_size as f32;
|
||||||
(value.re * value.re + value.im * value.im).sqrt() / SPECTRUM_FFT_SIZE as f32;
|
|
||||||
20.0 * mag.max(1e-10_f32).log10()
|
20.0 * mag.max(1e-10_f32).log10()
|
||||||
})
|
})
|
||||||
.collect();
|
.collect();
|
||||||
|
|||||||
@@ -73,6 +73,11 @@ pub struct SoapySdrConfig {
|
|||||||
pub nb_enabled: bool,
|
pub nb_enabled: bool,
|
||||||
/// Noise blanker impulse threshold multiplier.
|
/// Noise blanker impulse threshold multiplier.
|
||||||
pub nb_threshold: f64,
|
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 {
|
impl Default for SoapySdrConfig {
|
||||||
@@ -99,6 +104,7 @@ impl Default for SoapySdrConfig {
|
|||||||
max_virtual_channels: 4,
|
max_virtual_channels: 4,
|
||||||
nb_enabled: false,
|
nb_enabled: false,
|
||||||
nb_threshold: 10.0,
|
nb_threshold: 10.0,
|
||||||
|
spectrum_fft_size: 1024,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -194,6 +200,7 @@ impl SoapySdrRig {
|
|||||||
let max_virtual_channels = config.max_virtual_channels;
|
let max_virtual_channels = config.max_virtual_channels;
|
||||||
let nb_enabled = config.nb_enabled;
|
let nb_enabled = config.nb_enabled;
|
||||||
let nb_threshold = config.nb_threshold;
|
let nb_threshold = config.nb_threshold;
|
||||||
|
let spectrum_fft_size = config.spectrum_fft_size;
|
||||||
tracing::info!(
|
tracing::info!(
|
||||||
"initialising SoapySDR backend (args={:?}, gain_mode={:?}, gain_db={}, max_gain_db={:?})",
|
"initialising SoapySDR backend (args={:?}, gain_mode={:?}, gain_db={}, max_gain_db={:?})",
|
||||||
args,
|
args,
|
||||||
@@ -293,6 +300,7 @@ impl SoapySdrRig {
|
|||||||
threshold: nb_threshold as f32,
|
threshold: nb_threshold as f32,
|
||||||
},
|
},
|
||||||
&all_channels,
|
&all_channels,
|
||||||
|
spectrum_fft_size,
|
||||||
));
|
));
|
||||||
|
|
||||||
let info = RigInfo {
|
let info = RigInfo {
|
||||||
@@ -418,6 +426,7 @@ impl SoapySdrRig {
|
|||||||
nb_threshold: f64,
|
nb_threshold: f64,
|
||||||
) -> DynResult<Self> {
|
) -> DynResult<Self> {
|
||||||
Self::new_from_config(SoapySdrConfig {
|
Self::new_from_config(SoapySdrConfig {
|
||||||
|
spectrum_fft_size: 1024,
|
||||||
args: args.to_string(),
|
args: args.to_string(),
|
||||||
channels: channels.to_vec(),
|
channels: channels.to_vec(),
|
||||||
gain_mode: gain_mode.to_string(),
|
gain_mode: gain_mode.to_string(),
|
||||||
|
|||||||
@@ -439,6 +439,7 @@ mod tests {
|
|||||||
VirtualSquelchConfig::default(),
|
VirtualSquelchConfig::default(),
|
||||||
NoiseBlankerConfig::default(),
|
NoiseBlankerConfig::default(),
|
||||||
&[(0.0, RigMode::USB, 3_000)],
|
&[(0.0, RigMode::USB, 3_000)],
|
||||||
|
1024,
|
||||||
))
|
))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -120,6 +120,8 @@ wfm_deemphasis_us = 50
|
|||||||
center_offset_hz = 100000
|
center_offset_hz = 100000
|
||||||
channels = []
|
channels = []
|
||||||
max_virtual_channels = 4
|
max_virtual_channels = 4
|
||||||
|
spectrum_fft_size = 1024
|
||||||
|
spectrum_interval_ms = 50
|
||||||
|
|
||||||
# "auto" for hardware AGC, or "manual".
|
# "auto" for hardware AGC, or "manual".
|
||||||
[trx-server.sdr.gain]
|
[trx-server.sdr.gain]
|
||||||
@@ -160,6 +162,7 @@ log_level = "info"
|
|||||||
# Legacy single-remote form; prefer [[remotes]] below.
|
# Legacy single-remote form; prefer [[remotes]] below.
|
||||||
[trx-client.remote]
|
[trx-client.remote]
|
||||||
poll_interval_ms = 750
|
poll_interval_ms = 750
|
||||||
|
spectrum_interval_ms = 50
|
||||||
|
|
||||||
[trx-client.remote.auth]
|
[trx-client.remote.auth]
|
||||||
|
|
||||||
@@ -168,6 +171,7 @@ name = "home-hf"
|
|||||||
url = "192.168.1.100:4530"
|
url = "192.168.1.100:4530"
|
||||||
rig_id = "hf"
|
rig_id = "hf"
|
||||||
poll_interval_ms = 750
|
poll_interval_ms = 750
|
||||||
|
spectrum_interval_ms = 50
|
||||||
|
|
||||||
[trx-client.remotes.auth]
|
[trx-client.remotes.auth]
|
||||||
token = "my-token"
|
token = "my-token"
|
||||||
@@ -177,6 +181,7 @@ name = "home-vhf"
|
|||||||
url = "192.168.1.100:4530"
|
url = "192.168.1.100:4530"
|
||||||
rig_id = "vhf"
|
rig_id = "vhf"
|
||||||
poll_interval_ms = 750
|
poll_interval_ms = 750
|
||||||
|
spectrum_interval_ms = 50
|
||||||
|
|
||||||
[trx-client.remotes.auth]
|
[trx-client.remotes.auth]
|
||||||
token = "my-token"
|
token = "my-token"
|
||||||
|
|||||||
Reference in New Issue
Block a user