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

[style](trx-rs): reformat codebase

Browse Source 
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: Stanislaw Grams <stanislawgrams@gmail.com>
This commit is contained in:
Stan Grams
2026-03-17 22:36:11 +01:00
parent a823e66816
commit b533d704a1
22 changed files with 348 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/trx-client/src/audio_client.rs
+9 -7
View File
@@ -25,11 +25,11 @@ use trx_core::audio::{
parse_vchan_audio_frame, parse_vchan_uuid_msg, read_audio_msg, write_audio_msg,
write_vchan_uuid_msg, AudioStreamInfo, AUDIO_MSG_AIS_DECODE, AUDIO_MSG_APRS_DECODE,
AUDIO_MSG_CW_DECODE, AUDIO_MSG_FT2_DECODE, AUDIO_MSG_FT4_DECODE, AUDIO_MSG_FT8_DECODE,
AUDIO_MSG_HF_APRS_DECODE,
AUDIO_MSG_HISTORY_COMPRESSED, AUDIO_MSG_RX_FRAME, AUDIO_MSG_RX_FRAME_CH,
AUDIO_MSG_STREAM_INFO, AUDIO_MSG_TX_FRAME, AUDIO_MSG_VCHAN_ALLOCATED, AUDIO_MSG_VCHAN_BW,
AUDIO_MSG_VCHAN_DESTROYED, AUDIO_MSG_VCHAN_FREQ, AUDIO_MSG_VCHAN_MODE, AUDIO_MSG_VCHAN_REMOVE,
AUDIO_MSG_VCHAN_SUB, AUDIO_MSG_VCHAN_UNSUB, AUDIO_MSG_VDES_DECODE, AUDIO_MSG_WSPR_DECODE,
AUDIO_MSG_HF_APRS_DECODE, AUDIO_MSG_HISTORY_COMPRESSED, AUDIO_MSG_RX_FRAME,
AUDIO_MSG_RX_FRAME_CH, AUDIO_MSG_STREAM_INFO, AUDIO_MSG_TX_FRAME, AUDIO_MSG_VCHAN_ALLOCATED,
AUDIO_MSG_VCHAN_BW, AUDIO_MSG_VCHAN_DESTROYED, AUDIO_MSG_VCHAN_FREQ, AUDIO_MSG_VCHAN_MODE,
AUDIO_MSG_VCHAN_REMOVE, AUDIO_MSG_VCHAN_SUB, AUDIO_MSG_VCHAN_UNSUB, AUDIO_MSG_VDES_DECODE,
AUDIO_MSG_WSPR_DECODE,
};
use trx_core::decode::DecodedMessage;
use trx_frontend::VChanAudioCmd;
@@ -195,7 +195,8 @@ async fn handle_audio_connection(
}
// Re-apply non-default bandwidth after re-subscribing.
if sub.bandwidth_hz > 0 {
let bw_json = serde_json::json!({ "uuid": uuid.to_string(), "bandwidth_hz": sub.bandwidth_hz });
let bw_json =
serde_json::json!({ "uuid": uuid.to_string(), "bandwidth_hz": sub.bandwidth_hz });
if let Ok(payload) = serde_json::to_vec(&bw_json) {
if let Err(e) = write_audio_msg(&mut writer, AUDIO_MSG_VCHAN_BW, &payload).await {
warn!("Audio vchan reconnect BW write failed: {}", e);
@@ -209,7 +210,8 @@ async fn handle_audio_connection(
// Spawn RX read task
let rx_tx = rx_tx.clone();
let decode_tx = decode_tx.clone();
let vchan_audio_rx: Arc<RwLock<HashMap<Uuid, broadcast::Sender<Bytes>>>> = Arc::clone(vchan_audio);
let vchan_audio_rx: Arc<RwLock<HashMap<Uuid, broadcast::Sender<Bytes>>>> =
Arc::clone(vchan_audio);
let vchan_destroyed_for_rx = vchan_destroyed_tx.clone();
let mut rx_handle = tokio::spawn(async move {
loop {
src/trx-client/src/config.rs
+6 -10
View File
@@ -395,9 +395,7 @@ impl ClientConfig {
);
}
if self.frontends.http.decode_history_retention_min == 0 {
return Err(
"[frontends.http].decode_history_retention_min must be > 0".to_string(),
);
return Err("[frontends.http].decode_history_retention_min must be > 0".to_string());
}
for (rig_id, minutes) in &self.frontends.http.decode_history_retention_min_by_rig {
if rig_id.trim().is_empty() {
@@ -616,13 +614,11 @@ mod tests {
assert_eq!(config.frontends.http.spectrum_coverage_margin_hz, 50_000);
assert_eq!(config.frontends.http.spectrum_usable_span_ratio, 0.92);
assert_eq!(config.frontends.http.decode_history_retention_min, 1440);
assert!(
config
.frontends
.http
.decode_history_retention_min_by_rig
.is_empty()
);
assert!(config
.frontends
.http
.decode_history_retention_min_by_rig
.is_empty());
assert_eq!(config.frontends.rigctl.port, 4532);
assert!(config.frontends.http_json.enabled);
assert_eq!(config.frontends.http_json.port, 0);
src/trx-client/src/main.rs
+61 -60
View File
@@ -185,8 +185,11 @@ async fn async_init() -> DynResult<AppState> {
cfg.frontends.http.spectrum_usable_span_ratio;
frontend_runtime.http_decode_history_retention_min =
cfg.frontends.http.decode_history_retention_min;
frontend_runtime.http_decode_history_retention_min_by_rig =
cfg.frontends.http.decode_history_retention_min_by_rig.clone();
frontend_runtime.http_decode_history_retention_min_by_rig = cfg
.frontends
.http
.decode_history_retention_min_by_rig
.clone();
// Resolve remote URL: CLI > config [remote] section > error
let remote_url = cli
@@ -305,8 +308,7 @@ async fn async_init() -> DynResult<AppState> {
frontend_runtime.decode_rx = Some(decode_tx.clone());
// Virtual-channel audio: shared broadcaster map + command channel.
let (vchan_cmd_tx, vchan_cmd_rx) =
mpsc::unbounded_channel::<trx_frontend::VChanAudioCmd>();
let (vchan_cmd_tx, vchan_cmd_rx) = mpsc::unbounded_channel::<trx_frontend::VChanAudioCmd>();
*frontend_runtime.vchan_audio_cmd.lock().unwrap() = Some(vchan_cmd_tx);
let (vchan_destroyed_tx, _) = broadcast::channel::<uuid::Uuid>(64);
@@ -318,65 +320,64 @@ async fn async_init() -> DynResult<AppState> {
let cw_history = frontend_runtime.cw_history.clone();
let ft8_history = frontend_runtime.ft8_history.clone();
let wspr_history = frontend_runtime.wspr_history.clone();
let replay_history_sink: Arc<dyn Fn(DecodedMessage) + Send + Sync> =
Arc::new(move |msg| {
let now = std::time::Instant::now();
match msg {
DecodedMessage::Ais(mut message) => {
if message.ts_ms.is_none() {
message.ts_ms = Some(current_timestamp_ms());
}
if let Ok(mut history) = ais_history.lock() {
history.push_back((now, message));
}
let replay_history_sink: Arc<dyn Fn(DecodedMessage) + Send + Sync> = Arc::new(move |msg| {
let now = std::time::Instant::now();
match msg {
DecodedMessage::Ais(mut message) => {
if message.ts_ms.is_none() {
message.ts_ms = Some(current_timestamp_ms());
}
DecodedMessage::Vdes(mut message) => {
if message.ts_ms.is_none() {
message.ts_ms = Some(current_timestamp_ms());
}
if let Ok(mut history) = vdes_history.lock() {
history.push_back((now, message));
}
}
DecodedMessage::Aprs(mut packet) => {
if packet.ts_ms.is_none() {
packet.ts_ms = Some(current_timestamp_ms());
}
if let Ok(mut history) = aprs_history.lock() {
history.push_back((now, packet));
}
}
DecodedMessage::HfAprs(mut packet) => {
if packet.ts_ms.is_none() {
packet.ts_ms = Some(current_timestamp_ms());
}
if let Ok(mut history) = hf_aprs_history.lock() {
history.push_back((now, packet));
}
}
DecodedMessage::Cw(event) => {
if let Ok(mut history) = cw_history.lock() {
history.push_back((now, event));
}
}
DecodedMessage::Ft8(message) => {
if let Ok(mut history) = ft8_history.lock() {
history.push_back((now, message));
}
}
DecodedMessage::Ft4(_) => {
// FT4 history is managed by the frontend HTTP audio collector
}
DecodedMessage::Ft2(_) => {
// FT2 history is managed by the frontend HTTP audio collector
}
DecodedMessage::Wspr(message) => {
if let Ok(mut history) = wspr_history.lock() {
history.push_back((now, message));
}
if let Ok(mut history) = ais_history.lock() {
history.push_back((now, message));
}
}
});
DecodedMessage::Vdes(mut message) => {
if message.ts_ms.is_none() {
message.ts_ms = Some(current_timestamp_ms());
}
if let Ok(mut history) = vdes_history.lock() {
history.push_back((now, message));
}
}
DecodedMessage::Aprs(mut packet) => {
if packet.ts_ms.is_none() {
packet.ts_ms = Some(current_timestamp_ms());
}
if let Ok(mut history) = aprs_history.lock() {
history.push_back((now, packet));
}
}
DecodedMessage::HfAprs(mut packet) => {
if packet.ts_ms.is_none() {
packet.ts_ms = Some(current_timestamp_ms());
}
if let Ok(mut history) = hf_aprs_history.lock() {
history.push_back((now, packet));
}
}
DecodedMessage::Cw(event) => {
if let Ok(mut history) = cw_history.lock() {
history.push_back((now, event));
}
}
DecodedMessage::Ft8(message) => {
if let Ok(mut history) = ft8_history.lock() {
history.push_back((now, message));
}
}
DecodedMessage::Ft4(_) => {
// FT4 history is managed by the frontend HTTP audio collector
}
DecodedMessage::Ft2(_) => {
// FT2 history is managed by the frontend HTTP audio collector
}
DecodedMessage::Wspr(message) => {
if let Ok(mut history) = wspr_history.lock() {
history.push_back((now, message));
}
}
}
});
info!(
"Audio enabled: default port {}, decode channel set",
src/trx-client/src/remote_client.rs
+16 -29
View File
@@ -68,10 +68,7 @@ pub async fn run_remote_client(
) -> RigResult<()> {
// Spectrum polling runs on its own dedicated TCP connection so it never
// blocks state polls or user commands on the main connection.
let spectrum_task = tokio::spawn(run_spectrum_connection(
config.clone(),
shutdown_rx.clone(),
));
let spectrum_task = tokio::spawn(run_spectrum_connection(config.clone(), shutdown_rx.clone()));
let mut reconnect_delay = Duration::from_secs(1);
@@ -147,8 +144,7 @@ async fn run_spectrum_connection(
if let Err(e) = stream.set_nodelay(true) {
warn!("Spectrum TCP_NODELAY failed: {}", e);
}
if let Err(e) =
handle_spectrum_connection(&config, stream, &mut shutdown_rx).await
if let Err(e) = handle_spectrum_connection(&config, stream, &mut shutdown_rx).await
{
warn!("Spectrum connection dropped: {}", e);
}
@@ -301,13 +297,10 @@ async fn send_command(
.map_err(|e| RigError::communication(format!("JSON serialize failed: {e}")))?;
payload.push('\n');
time::timeout(
IO_TIMEOUT,
writer.write_all(payload.as_bytes()),
)
.await
.map_err(|_| RigError::communication(format!("write timed out after {:?}", IO_TIMEOUT)))?
.map_err(|e| RigError::communication(format!("write failed: {e}")))?;
time::timeout(IO_TIMEOUT, writer.write_all(payload.as_bytes()))
.await
.map_err(|_| RigError::communication(format!("write timed out after {:?}", IO_TIMEOUT)))?
.map_err(|e| RigError::communication(format!("write failed: {e}")))?;
time::timeout(IO_TIMEOUT, writer.flush())
.await
.map_err(|_| RigError::communication(format!("flush timed out after {:?}", IO_TIMEOUT)))?
@@ -347,15 +340,12 @@ async fn send_command_no_state_update(
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(format!("write timed out after {:?}", SPECTRUM_IO_TIMEOUT))
})?
.map_err(|e| RigError::communication(format!("write failed: {e}")))?;
time::timeout(SPECTRUM_IO_TIMEOUT, writer.write_all(payload.as_bytes()))
.await
.map_err(|_| {
RigError::communication(format!("write timed out after {:?}", SPECTRUM_IO_TIMEOUT))
})?
.map_err(|e| RigError::communication(format!("write failed: {e}")))?;
time::timeout(SPECTRUM_IO_TIMEOUT, writer.flush())
.await
.map_err(|_| {
@@ -443,13 +433,10 @@ async fn send_get_rigs(
.map_err(|e| RigError::communication(format!("JSON serialize failed: {e}")))?;
payload.push('\n');
time::timeout(
IO_TIMEOUT,
writer.write_all(payload.as_bytes()),
)
.await
.map_err(|_| RigError::communication(format!("write timed out after {:?}", IO_TIMEOUT)))?
.map_err(|e| RigError::communication(format!("write failed: {e}")))?;
time::timeout(IO_TIMEOUT, writer.write_all(payload.as_bytes()))
.await
.map_err(|_| RigError::communication(format!("write timed out after {:?}", IO_TIMEOUT)))?
.map_err(|e| RigError::communication(format!("write failed: {e}")))?;
time::timeout(IO_TIMEOUT, writer.flush())
.await
.map_err(|_| RigError::communication(format!("flush timed out after {:?}", IO_TIMEOUT)))?
src/trx-client/trx-frontend/src/lib.rs
+6 -2
View File
@@ -3,9 +3,9 @@
// SPDX-License-Identifier: BSD-2-Clause
use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::RwLock;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, AtomicUsize};
use std::sync::RwLock;
use std::sync::{Arc, Mutex};
use std::time::Instant;
@@ -95,7 +95,11 @@ pub struct SharedSpectrum {
impl SharedSpectrum {
/// Replace the stored frame, pre-serialising RDS in one pass.
pub fn set(&mut self, frame: Option<SpectrumData>, vchan_rds: Option<Vec<trx_core::rig::state::VchanRdsEntry>>) {
pub fn set(
&mut self,
frame: Option<SpectrumData>,
vchan_rds: Option<Vec<trx_core::rig::state::VchanRdsEntry>>,
) {
self.rds_json = frame
.as_ref()
.and_then(|f| f.rds.as_ref())
src/trx-client/trx-frontend/trx-frontend-http/src/api.rs
+78 -39
View File
@@ -47,8 +47,16 @@ fn base64_encode(data: &[u8]) -> String {
let n = (b0 << 16) | (b1 << 8) | b2;
out.push(T[((n >> 18) & 63) as usize]);
out.push(T[((n >> 12) & 63) as usize]);
out.push(if chunk.len() > 1 { T[((n >> 6) & 63) as usize] } else { b'=' });
out.push(if chunk.len() > 2 { T[(n & 63) as usize] } else { b'=' });
out.push(if chunk.len() > 1 {
T[((n >> 6) & 63) as usize]
} else {
b'='
});
out.push(if chunk.len() > 2 {
T[(n & 63) as usize]
} else {
b'='
});
}
// SAFETY: output contains only ASCII base64 characters.
unsafe { String::from_utf8_unchecked(out) }
@@ -120,23 +128,53 @@ fn inject_frontend_meta(json: &str, meta: FrontendMeta) -> String {
// Build only the extra key-value pairs as a JSON fragment.
let mut extra = serde_json::Map::new();
extra.insert("clients".into(), serde_json::json!(meta.http_clients));
extra.insert("rigctl_clients".into(), serde_json::json!(meta.rigctl_clients));
if let Some(v) = meta.rigctl_addr { extra.insert("rigctl_addr".into(), serde_json::json!(v)); }
if let Some(v) = meta.active_rig_id { extra.insert("active_rig_id".into(), serde_json::json!(v)); }
extra.insert(
"rigctl_clients".into(),
serde_json::json!(meta.rigctl_clients),
);
if let Some(v) = meta.rigctl_addr {
extra.insert("rigctl_addr".into(), serde_json::json!(v));
}
if let Some(v) = meta.active_rig_id {
extra.insert("active_rig_id".into(), serde_json::json!(v));
}
extra.insert("rig_ids".into(), serde_json::json!(meta.rig_ids));
if let Some(v) = meta.owner_callsign { extra.insert("owner_callsign".into(), serde_json::json!(v)); }
if let Some(v) = meta.owner_website_url { extra.insert("owner_website_url".into(), serde_json::json!(v)); }
if let Some(v) = meta.owner_website_name { extra.insert("owner_website_name".into(), serde_json::json!(v)); }
if let Some(v) = meta.ais_vessel_url_base { extra.insert("ais_vessel_url_base".into(), serde_json::json!(v)); }
extra.insert("show_sdr_gain_control".into(), serde_json::json!(meta.show_sdr_gain_control));
extra.insert("initial_map_zoom".into(), serde_json::json!(meta.initial_map_zoom));
extra.insert("spectrum_coverage_margin_hz".into(), serde_json::json!(meta.spectrum_coverage_margin_hz));
extra.insert("spectrum_usable_span_ratio".into(), serde_json::json!(meta.spectrum_usable_span_ratio));
if let Some(v) = meta.owner_callsign {
extra.insert("owner_callsign".into(), serde_json::json!(v));
}
if let Some(v) = meta.owner_website_url {
extra.insert("owner_website_url".into(), serde_json::json!(v));
}
if let Some(v) = meta.owner_website_name {
extra.insert("owner_website_name".into(), serde_json::json!(v));
}
if let Some(v) = meta.ais_vessel_url_base {
extra.insert("ais_vessel_url_base".into(), serde_json::json!(v));
}
extra.insert(
"show_sdr_gain_control".into(),
serde_json::json!(meta.show_sdr_gain_control),
);
extra.insert(
"initial_map_zoom".into(),
serde_json::json!(meta.initial_map_zoom),
);
extra.insert(
"spectrum_coverage_margin_hz".into(),
serde_json::json!(meta.spectrum_coverage_margin_hz),
);
extra.insert(
"spectrum_usable_span_ratio".into(),
serde_json::json!(meta.spectrum_usable_span_ratio),
);
extra.insert(
"decode_history_retention_min".into(),
serde_json::json!(meta.decode_history_retention_min),
);
extra.insert("server_connected".into(), serde_json::json!(meta.server_connected));
extra.insert(
"server_connected".into(),
serde_json::json!(meta.server_connected),
);
// Serialize the extra map, strip its outer braces, and splice in.
let extra_json = match serde_json::to_string(&extra) {
@@ -328,9 +366,7 @@ pub async fn events(
let scheduler_control = scheduler_control_updates.clone();
async move {
state.snapshot().and_then(|v| {
if let Ok(Some(rig_id)) =
context.remote_active_rig_id.lock().map(|g| g.clone())
{
if let Ok(Some(rig_id)) = context.remote_active_rig_id.lock().map(|g| g.clone()) {
vchan.update_primary(
&rig_id,
v.status.freq.hz,
@@ -367,9 +403,8 @@ pub async fn events(
if let Some(colon) = msg.find(':') {
let rig_id = &msg[..colon];
let channels_json = &msg[colon + 1..];
let payload = format!(
"{{\"rig_id\":\"{rig_id}\",\"channels\":{channels_json}}}"
);
let payload =
format!("{{\"rig_id\":\"{rig_id}\",\"channels\":{channels_json}}}");
return Some((
Ok::<Bytes, Error>(Bytes::from(format!(
"event: channels\ndata: {payload}\n\n"
@@ -573,9 +608,7 @@ fn gzip_bytes(payload: &[u8]) -> std::io::Result<Vec<u8>> {
/// not block real-time messages: the client fetches this endpoint in parallel
/// with opening the SSE connection and drains it in the background.
#[get("/decode/history")]
pub async fn decode_history(
context: web::Data<Arc<FrontendRuntimeContext>>,
) -> impl Responder {
pub async fn decode_history(context: web::Data<Arc<FrontendRuntimeContext>>) -> impl Responder {
if context.decode_rx.is_none() {
return HttpResponse::NotFound().body("decode not enabled");
}
@@ -1414,9 +1447,7 @@ pub async fn delete_channel_route(
let (rig_id, channel_id) = path.into_inner();
match vchan_mgr.delete_channel(&rig_id, channel_id) {
Ok(()) => HttpResponse::Ok().finish(),
Err(crate::server::vchan::VChanClientError::NotFound) => {
HttpResponse::NotFound().finish()
}
Err(crate::server::vchan::VChanClientError::NotFound) => HttpResponse::NotFound().finish(),
Err(crate::server::vchan::VChanClientError::Permanent) => {
HttpResponse::BadRequest().body("cannot remove the primary channel")
}
@@ -1476,9 +1507,7 @@ pub async fn set_vchan_freq(
let (rig_id, channel_id) = path.into_inner();
match vchan_mgr.set_channel_freq(&rig_id, channel_id, body.freq_hz) {
Ok(()) => HttpResponse::Ok().finish(),
Err(crate::server::vchan::VChanClientError::NotFound) => {
HttpResponse::NotFound().finish()
}
Err(crate::server::vchan::VChanClientError::NotFound) => HttpResponse::NotFound().finish(),
Err(e) => HttpResponse::BadRequest().body(e.to_string()),
}
}
@@ -1497,9 +1526,7 @@ pub async fn set_vchan_bw(
let (rig_id, channel_id) = path.into_inner();
match vchan_mgr.set_channel_bandwidth(&rig_id, channel_id, body.bandwidth_hz) {
Ok(()) => HttpResponse::Ok().finish(),
Err(crate::server::vchan::VChanClientError::NotFound) => {
HttpResponse::NotFound().finish()
}
Err(crate::server::vchan::VChanClientError::NotFound) => HttpResponse::NotFound().finish(),
Err(e) => HttpResponse::BadRequest().body(e.to_string()),
}
}
@@ -1518,9 +1545,7 @@ pub async fn set_vchan_mode(
let (rig_id, channel_id) = path.into_inner();
match vchan_mgr.set_channel_mode(&rig_id, channel_id, &body.mode) {
Ok(()) => HttpResponse::Ok().finish(),
Err(crate::server::vchan::VChanClientError::NotFound) => {
HttpResponse::NotFound().finish()
}
Err(crate::server::vchan::VChanClientError::NotFound) => HttpResponse::NotFound().finish(),
Err(e) => HttpResponse::BadRequest().body(e.to_string()),
}
}
@@ -1783,14 +1808,20 @@ async fn ft8_js() -> impl Responder {
#[get("/ft4.js")]
async fn ft4_js() -> impl Responder {
HttpResponse::Ok()
.insert_header((header::CONTENT_TYPE, "application/javascript; charset=utf-8"))
.insert_header((
header::CONTENT_TYPE,
"application/javascript; charset=utf-8",
))
.body(status::FT4_JS)
}
#[get("/ft2.js")]
async fn ft2_js() -> impl Responder {
HttpResponse::Ok()
.insert_header((header::CONTENT_TYPE, "application/javascript; charset=utf-8"))
.insert_header((
header::CONTENT_TYPE,
"application/javascript; charset=utf-8",
))
.body(status::FT2_JS)
}
@@ -1951,7 +1982,14 @@ fn bookmark_decoder_state(
}
}
(want_aprs, want_hf_aprs, want_ft8, want_ft4, want_ft2, want_wspr)
(
want_aprs,
want_hf_aprs,
want_ft8,
want_ft4,
want_ft2,
want_wspr,
)
}
fn bookmark_decoder_kinds(bookmark: &crate::server::bookmarks::Bookmark) -> Vec<String> {
@@ -2018,7 +2056,8 @@ async fn apply_selected_channel(
let Some(bookmark) = bookmark_store.get(bookmark_id) else {
return Ok(());
};
let (want_aprs, want_hf_aprs, want_ft8, want_ft4, want_ft2, want_wspr) = bookmark_decoder_state(&bookmark);
let (want_aprs, want_hf_aprs, want_ft8, want_ft4, want_ft2, want_wspr) =
bookmark_decoder_state(&bookmark);
let desired = [
RigCommand::SetAprsDecodeEnabled(want_aprs),
RigCommand::SetHfAprsDecodeEnabled(want_hf_aprs),
src/trx-client/trx-frontend/trx-frontend-http/src/audio.rs
+20 -5
View File
@@ -57,7 +57,10 @@ fn decode_history_cutoff(context: &FrontendRuntimeContext) -> Instant {
Instant::now() - decode_history_retention(context)
}
fn prune_aprs_history(context: &FrontendRuntimeContext, history: &mut VecDeque<(Instant, AprsPacket)>) {
fn prune_aprs_history(
context: &FrontendRuntimeContext,
history: &mut VecDeque<(Instant, AprsPacket)>,
) {
let cutoff = decode_history_cutoff(context);
while let Some((ts, _)) = history.front() {
if *ts >= cutoff {
@@ -80,7 +83,10 @@ fn prune_hf_aprs_history(
}
}
fn prune_ais_history(context: &FrontendRuntimeContext, history: &mut VecDeque<(Instant, AisMessage)>) {
fn prune_ais_history(
context: &FrontendRuntimeContext,
history: &mut VecDeque<(Instant, AisMessage)>,
) {
let cutoff = decode_history_cutoff(context);
while let Some((ts, _)) = history.front() {
if *ts >= cutoff {
@@ -137,7 +143,10 @@ fn prune_cw_history(context: &FrontendRuntimeContext, history: &mut VecDeque<(In
}
}
fn prune_ft8_history(context: &FrontendRuntimeContext, history: &mut VecDeque<(Instant, Ft8Message)>) {
fn prune_ft8_history(
context: &FrontendRuntimeContext,
history: &mut VecDeque<(Instant, Ft8Message)>,
) {
let cutoff = decode_history_cutoff(context);
while let Some((ts, _)) = history.front() {
if *ts >= cutoff {
@@ -147,7 +156,10 @@ fn prune_ft8_history(context: &FrontendRuntimeContext, history: &mut VecDeque<(I
}
}
fn prune_ft4_history(context: &FrontendRuntimeContext, history: &mut VecDeque<(Instant, Ft8Message)>) {
fn prune_ft4_history(
context: &FrontendRuntimeContext,
history: &mut VecDeque<(Instant, Ft8Message)>,
) {
let cutoff = decode_history_cutoff(context);
while let Some((ts, _)) = history.front() {
if *ts >= cutoff {
@@ -157,7 +169,10 @@ fn prune_ft4_history(context: &FrontendRuntimeContext, history: &mut VecDeque<(I
}
}
fn prune_ft2_history(context: &FrontendRuntimeContext, history: &mut VecDeque<(Instant, Ft8Message)>) {
fn prune_ft2_history(
context: &FrontendRuntimeContext,
history: &mut VecDeque<(Instant, Ft8Message)>,
) {
let cutoff = decode_history_cutoff(context);
while let Some((ts, _)) = history.front() {
if *ts >= cutoff {
src/trx-client/trx-frontend/trx-frontend-http/src/background_decode.rs
+27 -15
View File
@@ -85,12 +85,24 @@ impl BackgroundDecodeStore {
let _ = std::fs::create_dir_all(parent);
}
let db = if path.exists() {
PickleDb::load(path, PickleDbDumpPolicy::AutoDump, SerializationMethod::Json)
.unwrap_or_else(|_| {
PickleDb::new(path, PickleDbDumpPolicy::AutoDump, SerializationMethod::Json)
})
PickleDb::load(
path,
PickleDbDumpPolicy::AutoDump,
SerializationMethod::Json,
)
.unwrap_or_else(|_| {
PickleDb::new(
path,
PickleDbDumpPolicy::AutoDump,
SerializationMethod::Json,
)
})
} else {
PickleDb::new(path, PickleDbDumpPolicy::AutoDump, SerializationMethod::Json)
PickleDb::new(
path,
PickleDbDumpPolicy::AutoDump,
SerializationMethod::Json,
)
};
Self {
db: Arc::new(RwLock::new(db)),
@@ -160,11 +172,13 @@ impl BackgroundDecodeManager {
}
pub fn get_config(&self, rig_id: &str) -> BackgroundDecodeConfig {
self.store.get(rig_id).unwrap_or_else(|| BackgroundDecodeConfig {
rig_id: rig_id.to_string(),
enabled: false,
bookmark_ids: Vec::new(),
})
self.store
.get(rig_id)
.unwrap_or_else(|| BackgroundDecodeConfig {
rig_id: rig_id.to_string(),
enabled: false,
bookmark_ids: Vec::new(),
})
}
pub fn put_config(&self, mut config: BackgroundDecodeConfig) -> Option<BackgroundDecodeConfig> {
@@ -268,10 +282,7 @@ impl BackgroundDecodeManager {
bookmark_id: bookmark.id.clone(),
freq_hz: bookmark.freq_hz,
mode: bookmark.mode.clone(),
bandwidth_hz: bookmark
.bandwidth_hz
.unwrap_or(0)
.min(u32::MAX as u64) as u32,
bandwidth_hz: bookmark.bandwidth_hz.unwrap_or(0).min(u32::MAX as u64) as u32,
decoder_kinds,
}
}
@@ -565,7 +576,8 @@ fn bookmark_supported_decoder_kinds(bookmark: &Bookmark) -> Vec<String> {
}
fn channel_matches_bookmark(channel: &ClientChannel, bookmark: &Bookmark) -> bool {
channel.freq_hz == bookmark.freq_hz && normalized_mode(&channel.mode) == normalized_mode(&bookmark.mode)
channel.freq_hz == bookmark.freq_hz
&& normalized_mode(&channel.mode) == normalized_mode(&bookmark.mode)
}
fn normalized_mode(mode: &str) -> String {
src/trx-client/trx-frontend/trx-frontend-http/src/scheduler.rs
+37 -17
View File
@@ -117,12 +117,24 @@ impl SchedulerStore {
let _ = std::fs::create_dir_all(parent);
}
let db = if path.exists() {
PickleDb::load(path, PickleDbDumpPolicy::AutoDump, SerializationMethod::Json)
.unwrap_or_else(|_| {
PickleDb::new(path, PickleDbDumpPolicy::AutoDump, SerializationMethod::Json)
})
PickleDb::load(
path,
PickleDbDumpPolicy::AutoDump,
SerializationMethod::Json,
)
.unwrap_or_else(|_| {
PickleDb::new(
path,
PickleDbDumpPolicy::AutoDump,
SerializationMethod::Json,
)
})
} else {
PickleDb::new(path, PickleDbDumpPolicy::AutoDump, SerializationMethod::Json)
PickleDb::new(
path,
PickleDbDumpPolicy::AutoDump,
SerializationMethod::Json,
)
};
Self {
db: Arc::new(RwLock::new(db)),
@@ -206,10 +218,8 @@ fn sunrise_sunset_today(lat_deg: f64, lon_deg: f64) -> Option<(f64, f64)> {
let lambda = sun_lon - 0.00569 - 0.00478 * omega.to_radians().sin();
// Obliquity of the ecliptic.
let eps0 = 23.0
+ (26.0
+ (21.448 - jc * (46.8150 + jc * (0.00059 - jc * 0.001813))) / 60.0)
/ 60.0;
let eps0 =
23.0 + (26.0 + (21.448 - jc * (46.8150 + jc * (0.00059 - jc * 0.001813))) / 60.0) / 60.0;
let eps = eps0 + 0.00256 * omega.to_radians().cos();
// Sun's declination.
@@ -219,8 +229,7 @@ fn sunrise_sunset_today(lat_deg: f64, lon_deg: f64) -> Option<(f64, f64)> {
let y = (eps.to_radians() / 2.0).tan().powi(2);
let l0_rad = l0.to_radians();
let eot = 4.0
* (y * (2.0 * l0_rad).sin()
- 2.0 * m_rad.sin()
* (y * (2.0 * l0_rad).sin() - 2.0 * m_rad.sin()
+ 4.0 * y * m_rad.sin() * (2.0 * l0_rad).cos()
- 0.5 * y * y * (4.0 * l0_rad).sin()
- 1.25 * (2.0 * m_rad).sin())
@@ -228,8 +237,7 @@ fn sunrise_sunset_today(lat_deg: f64, lon_deg: f64) -> Option<(f64, f64)> {
// Hour angle for sunrise/sunset (zenith = 90.833°).
let lat_rad = lat_deg.to_radians();
let cos_ha = ((PI / 2.0 + 0.833_f64.to_radians()).cos())
/ (lat_rad.cos() * decl.cos())
let cos_ha = ((PI / 2.0 + 0.833_f64.to_radians()).cos()) / (lat_rad.cos() * decl.cos())
- lat_rad.tan() * decl.tan();
if !(-1.0..=1.0).contains(&cos_ha) {
@@ -654,7 +662,10 @@ pub fn spawn_scheduler_task(
)
.await
{
warn!("scheduler: failed to apply target for '{}': {e}", config.rig_id);
warn!(
"scheduler: failed to apply target for '{}': {e}",
config.rig_id
);
continue;
}
@@ -678,7 +689,11 @@ async fn apply_scheduler_decoders(
let mut want_wspr = false;
let mut update_from = |bm: &crate::server::bookmarks::Bookmark| {
for decoder in bm.decoders.iter().map(|item| item.trim().to_ascii_lowercase()) {
for decoder in bm
.decoders
.iter()
.map(|item| item.trim().to_ascii_lowercase())
{
match decoder.as_str() {
"aprs" => want_aprs = true,
"hf-aprs" => want_hf_aprs = true,
@@ -707,7 +722,10 @@ async fn apply_scheduler_decoders(
for (label, cmd) in desired {
if let Err(e) = scheduler_send(rig_tx, cmd, rig_id.to_string()).await {
warn!("scheduler: Set{label}DecodeEnabled failed for '{}': {:?}", rig_id, e);
warn!(
"scheduler: Set{label}DecodeEnabled failed for '{}': {:?}",
rig_id, e
);
}
}
}
@@ -931,7 +949,9 @@ pub async fn put_scheduler_control(
#[cfg(test)]
mod tests {
use super::{timespan_active_entry, timespan_cycle_slot, timespan_active_entries, ScheduleEntry};
use super::{
timespan_active_entries, timespan_active_entry, timespan_cycle_slot, ScheduleEntry,
};
fn entry(
id: &str,
src/trx-client/trx-frontend/trx-frontend-http/src/server.rs
+3 -4
View File
@@ -6,10 +6,10 @@
mod api;
#[path = "audio.rs"]
pub mod audio;
#[path = "background_decode.rs"]
pub mod background_decode;
#[path = "auth.rs"]
pub mod auth;
#[path = "background_decode.rs"]
pub mod background_decode;
#[path = "bookmarks.rs"]
pub mod bookmarks;
#[path = "scheduler.rs"]
@@ -88,8 +88,7 @@ async fn serve(
);
let background_decode_path = BackgroundDecodeStore::default_path();
let background_decode_store =
Arc::new(BackgroundDecodeStore::open(&background_decode_path));
let background_decode_store = Arc::new(BackgroundDecodeStore::open(&background_decode_path));
let vchan_mgr = Arc::new(ClientChannelManager::new(4));
let background_decode_mgr = BackgroundDecodeManager::new(
background_decode_store,
src/trx-client/trx-frontend/trx-frontend-http/src/status.rs
+2 -4
View File
@@ -9,8 +9,7 @@ const CLIENT_BUILD_DATE: &str = env!("TRX_CLIENT_BUILD_DATE");
const INDEX_HTML: &str = include_str!("../assets/web/index.html");
pub const STYLE_CSS: &str = include_str!("../assets/web/style.css");
pub const APP_JS: &str = include_str!("../assets/web/app.js");
pub const DECODE_HISTORY_WORKER_JS: &str =
include_str!("../assets/web/decode-history-worker.js");
pub const DECODE_HISTORY_WORKER_JS: &str = include_str!("../assets/web/decode-history-worker.js");
pub const WEBGL_RENDERER_JS: &str = include_str!("../assets/web/webgl-renderer.js");
pub const LEAFLET_AIS_TRACKSYMBOL_JS: &str =
include_str!("../assets/web/leaflet-ais-tracksymbol.js");
@@ -25,8 +24,7 @@ pub const WSPR_JS: &str = include_str!("../assets/web/plugins/wspr.js");
pub const CW_JS: &str = include_str!("../assets/web/plugins/cw.js");
pub const BOOKMARKS_JS: &str = include_str!("../assets/web/plugins/bookmarks.js");
pub const SCHEDULER_JS: &str = include_str!("../assets/web/plugins/scheduler.js");
pub const BACKGROUND_DECODE_JS: &str =
include_str!("../assets/web/plugins/background-decode.js");
pub const BACKGROUND_DECODE_JS: &str = include_str!("../assets/web/plugins/background-decode.js");
pub const VCHAN_JS: &str = include_str!("../assets/web/plugins/vchan.js");
pub fn index_html() -> String {
src/trx-client/trx-frontend/trx-frontend-http/src/vchan.rs
+23 -15
View File
@@ -367,11 +367,7 @@ impl ClientChannelManager {
}
/// Explicitly delete a channel by UUID (any session may do this).
pub fn delete_channel(
&self,
rig_id: &str,
channel_id: Uuid,
) -> Result<(), VChanClientError> {
pub fn delete_channel(&self, rig_id: &str, channel_id: Uuid) -> Result<(), VChanClientError> {
let mut rigs = self.rigs.write().unwrap();
let channels = rigs.get_mut(rig_id).ok_or(VChanClientError::NotFound)?;
let pos = channels
@@ -450,7 +446,10 @@ impl ClientChannelManager {
ch.freq_hz = freq_hz;
self.broadcast_change(rig_id, channels);
drop(rigs);
self.send_audio_cmd(VChanAudioCmd::SetFreq { uuid: channel_id, freq_hz });
self.send_audio_cmd(VChanAudioCmd::SetFreq {
uuid: channel_id,
freq_hz,
});
Ok(())
}
@@ -469,7 +468,10 @@ impl ClientChannelManager {
ch.mode = mode.to_string();
self.broadcast_change(rig_id, channels);
drop(rigs);
self.send_audio_cmd(VChanAudioCmd::SetMode { uuid: channel_id, mode: mode.to_string() });
self.send_audio_cmd(VChanAudioCmd::SetMode {
uuid: channel_id,
mode: mode.to_string(),
});
Ok(())
}
@@ -488,7 +490,10 @@ impl ClientChannelManager {
ch.bandwidth_hz = bandwidth_hz;
self.broadcast_change(rig_id, channels);
drop(rigs);
self.send_audio_cmd(VChanAudioCmd::SetBandwidth { uuid: channel_id, bandwidth_hz });
self.send_audio_cmd(VChanAudioCmd::SetBandwidth {
uuid: channel_id,
bandwidth_hz,
});
Ok(())
}
@@ -530,12 +535,14 @@ impl ClientChannelManager {
let mut changed = false;
let desired_map: HashMap<String, (u64, String, u32, Vec<String>)> = desired
.iter()
.map(|(bookmark_id, freq_hz, mode, bandwidth_hz, decoder_kinds)| {
(
bookmark_id.clone(),
(*freq_hz, mode.clone(), *bandwidth_hz, decoder_kinds.clone()),
)
})
.map(
|(bookmark_id, freq_hz, mode, bandwidth_hz, decoder_kinds)| {
(
bookmark_id.clone(),
(*freq_hz, mode.clone(), *bandwidth_hz, decoder_kinds.clone()),
)
},
)
.collect();
let desired_ids: std::collections::HashSet<&str> =
desired_map.keys().map(String::as_str).collect();
@@ -561,7 +568,8 @@ impl ClientChannelManager {
let Some(bookmark_id) = channel.scheduler_bookmark_id.as_deref() else {
continue;
};
let Some((freq_hz, mode, bandwidth_hz, decoder_kinds)) = desired_map.get(bookmark_id) else {
let Some((freq_hz, mode, bandwidth_hz, decoder_kinds)) = desired_map.get(bookmark_id)
else {
continue;
};
if channel.freq_hz != *freq_hz {