Better active but silent talkgroup switching

1 files changed, 37 insertions, 29 deletions

diff --git a/src/main.rs b/src/main.rs
index c686f0c..3ca2b2b 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -4,8 +4,8 @@ use ringbuf::HeapRb;
 use serde::Deserialize;
 use std::{
     collections::HashMap,
+    fmt::Display,
     net::UdpSocket,
-    sync::{Arc, Mutex},
     time::{Duration, Instant},
 };
 
@@ -25,6 +25,11 @@ struct TalkGroup {
     #[serde(rename(deserialize = "Priority"))]
     priority: i32,
 }
+impl Display for TalkGroup {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{} ({})", self.name, self.id)
+    }
+}
 impl TalkGroup {
     fn unknown(id: u32) -> Self {
         Self {
@@ -76,7 +81,7 @@ fn write_audio<T: cpal::Sample + std::fmt::Debug>(
 enum TalkGroupStatus {
     None,
     Complete(TalkGroup, Instant),
-    Active(TalkGroup),
+    Active(TalkGroup, Instant),
 }
 
 fn main() {
@@ -119,26 +124,16 @@ fn main() {
 
     let (device, stream_config) = init_cpal();
 
-    let buffer = HeapRb::<i16>::new(8192);
+    let buffer = HeapRb::<i16>::new(256 * 1024);
     let (mut producer, mut consumer) = buffer.split();
 
-    let listening_tg = Arc::new(Mutex::new(TalkGroupStatus::None));
-    let stream_listening_tg = listening_tg.clone();
+    let mut listening_tg = TalkGroupStatus::None;
 
     let audio_stream = match stream_config.sample_format() {
         cpal::SampleFormat::I16 => device
             .build_output_stream(
                 &stream_config.into(),
-                move |data: &mut [i16], cbinfo| {
-                    let mut listening_tg = stream_listening_tg.lock().unwrap();
-                    if let TalkGroupStatus::Complete(tg, ea) = &*listening_tg {
-                        if ea.elapsed() > Duration::from_secs(2) {
-                            info!("Stopped listening to {}", tg.name);
-                            *listening_tg = TalkGroupStatus::None;
-                        }
-                    }
-                    write_audio(data, &mut consumer, cbinfo)
-                },
+                move |data: &mut [i16], cbinfo| write_audio(data, &mut consumer, cbinfo),
                 |err| eprintln!("error occurred on the audio output stream: {}", err),
             )
             .unwrap(),
@@ -165,20 +160,32 @@ fn main() {
         }
         let audio_bytes = &filled_buf[4..];
 
-        let mut listening_tg = listening_tg.lock().unwrap();
-        match &*listening_tg {
-            TalkGroupStatus::Active(listen_tg) => {
+        match &listening_tg {
+            TalkGroupStatus::Active(listen_tg, last_packet) => {
                 if listen_tg.id != curr_tg.id {
                     // Big maybe on if we wanna do this. In theory, we can swap talkgroups
                     // immediately if a higher priority talkgroup comes in, but that might
                     // be a bit jarring. For now, we'll just wait until the current talkgroup
                     // is done.
                     if curr_tg.priority > 0 && curr_tg.priority < listen_tg.priority {
-                        println!("Switching to TG with higher priority: {}", curr_tg.name);
-                        *listening_tg = TalkGroupStatus::Active(curr_tg.to_owned());
+                        if audio_bytes.len() == 2
+                            && i16::from_le_bytes(audio_bytes.try_into().unwrap()) == 0
+                        {
+                            continue;
+                        }
+                        info!("Switching to TG with higher priority: {}", curr_tg);
+                        listening_tg = TalkGroupStatus::Active(curr_tg.to_owned(), Instant::now());
+                    } else if last_packet.elapsed() > Duration::from_secs(2) {
+                        info!(
+                            "No data from TG {} in over 2 seconds, switching to TG: {}",
+                            listen_tg, curr_tg
+                        );
+                        listening_tg = TalkGroupStatus::Active(curr_tg.to_owned(), Instant::now());
                     } else {
                         continue;
                     }
+                } else {
+                    listening_tg = TalkGroupStatus::Active(curr_tg.to_owned(), Instant::now());
                 }
             }
             TalkGroupStatus::Complete(listen_tg, completed_at) => {
@@ -189,28 +196,29 @@ fn main() {
                 }
                 if listen_tg.id != curr_tg.id {
                     if completed_at.elapsed() > Duration::from_secs(1) {
-                        info!("Switching to TG: {}", curr_tg.name);
-                        *listening_tg = TalkGroupStatus::Active(curr_tg.to_owned());
+                        info!("Switching to TG: {}", curr_tg);
+                        listening_tg = TalkGroupStatus::Active(curr_tg.to_owned(), Instant::now());
                     } else {
                         debug!("TG Swap Wait not completed, skipping packet");
                         continue;
                     }
                 } else {
                     debug!("Complete talkgroup still active, resetting");
-                    *listening_tg = TalkGroupStatus::Active(curr_tg.to_owned());
+                    listening_tg = TalkGroupStatus::Active(curr_tg.to_owned(), Instant::now());
                 }
             }
             TalkGroupStatus::None => {
-                info!("Switching to TG: {}", curr_tg.name);
-                *listening_tg = TalkGroupStatus::Active(curr_tg.to_owned());
+                info!("Switching to TG: {}", curr_tg);
+                listening_tg = TalkGroupStatus::Active(curr_tg.to_owned(), Instant::now());
             }
         }
 
         if audio_bytes.len() == 2 && i16::from_le_bytes(audio_bytes.try_into().unwrap()) == 0 {
-            // if let Some(listen_tg) = &listening_tg {
-            //     println!("End of transmission from TG {}", listen_tg.name);
-            // }
-            *listening_tg = TalkGroupStatus::Complete(curr_tg.to_owned(), Instant::now());
+            debug!(
+                "Null packet received from TG {}, marking as Complete",
+                curr_tg
+            );
+            listening_tg = TalkGroupStatus::Complete(curr_tg.to_owned(), Instant::now());
             continue;
         }