src/coordinator/coord_state.rs

@@ -186,6 +186,7 @@ impl<'a> CoordState<'a> {
             self.tracker_connection_state.clone(),
             self.tracker_state.clone(),
             self.tracker_metrics.clone(),
+            self.rt.clone(),
         ));
     }
 

src/coordinator/perf_state.rs

@@ -1,7 +1,7 @@
 use std::{collections::VecDeque, sync, time::Duration};
 
 const MAX_RECORDED_TIMES: usize = 10;
-const DEGRADED_TRACKER_TIME: u128 = 150;
+const DEGRADED_TRACKER_TIME: u128 = 100;
 
 #[derive(Debug)]
 pub struct TrackerMetrics {

src/coordinator/remote_video_processor.rs

@@ -5,11 +5,15 @@ use std::{
 };
 
 use async_channel::Sender;
-use futures_util::{SinkExt, StreamExt, TryStreamExt};
+use futures_util::{stream::SplitStream, SinkExt, StreamExt, TryStreamExt};
 use gstreamer_app::AppSink;
 use gstreamer_video::{video_frame::Readable, VideoFrame, VideoInfo};
-use tokio::time::{sleep_until, Instant};
-use tokio_tungstenite::{connect_async, tungstenite::Message};
+use tokio::{
+    net::TcpStream,
+    runtime::Handle,
+    time::{sleep_until, Instant},
+};
+use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
 use tracing::{error, info, instrument};
 
 use crate::remote_sources::TrackerState;
@@ -27,6 +31,7 @@ pub async fn remote_video_loop(
     socket_state: Arc<SocketState>,
     tracker_state: Arc<Mutex<TrackerState>>,
     tracker_metrics: Arc<tokio::sync::Mutex<TrackerMetrics>>,
+    runtime: Handle,
 ) {
     info!(
         "Starting remote tracker processing connection to: {}",
@@ -63,14 +68,14 @@ pub async fn remote_video_loop(
                 sleep_until(Instant::now() + Duration::from_secs(1)).await;
             }
             Ok((connection, _)) => {
-                let (mut sender, mut recvr) = connection.split();
+                let (mut sender, recvr) = connection.split();
 
-                // runtime.spawn(listen_to_messages(
-                //     recvr,
-                //     to_mec.clone(),
-                //     tracker_state.clone(),
-                //     socket_state.clone(),
-                // ));
+                runtime.spawn(listen_to_messages(
+                    recvr,
+                    to_mec.clone(),
+                    tracker_state.clone(),
+                    socket_state.clone(),
+                ));
 
                 let mut last_iter: Instant;
 
@@ -95,7 +100,7 @@ pub async fn remote_video_loop(
                                 error!("Could not close socket to remote computer: {e}")
                             }
                             socket_state.is_connected.store(false, Ordering::SeqCst);
-                            break;
+                            return;
                         }
                     };
 
@@ -113,42 +118,7 @@ pub async fn remote_video_loop(
                         }
                         socket_state.is_connected.store(false, Ordering::SeqCst);
                         socket_state.stay_connected.store(false, Ordering::SeqCst);
-                        break;
-                    }
-
-                    match recvr.try_next().await {
-                        Ok(Some(message)) => {
-                            let (x_off, y_off, _do_send) =
-                                process_incoming_string(message.to_string(), &tracker_state)
-                                    .and_then(|_| calculate_tracking(&tracker_state))
-                                    .unwrap_or((0, 0, false));
-
-                            let do_send = true;
-
-                            // For some reason, this do_send is inverted from what it should be
-                            // info!("Do Send is: {}", do_send.to_string());
-                            if do_send {
-                                if let Err(e) = to_mec
-                                    .send(ApplicationEvent::MoveEvent(
-                                        super::MoveEvent { x: x_off, y: y_off },
-                                        super::ConnectionType::Automated,
-                                    ))
-                                    .await
-                                {
-                                    error!("Could not send message to MEC, assuming critical failure: {e}");
-                                    socket_state.is_connected.store(false, Ordering::SeqCst);
-                                    socket_state.stay_connected.store(false, Ordering::SeqCst);
-                                    return;
-                                }
-                            }
-                        }
-                        Ok(None) => {
-                            info!("Recieved an empty message from the remote computer: Aborting");
-                            break;
-                        }
-                        Err(e) => {
-                            error!("Got an error on while recieving from remote computer: {e}");
-                        }
+                        return;
                     }
 
                     if !socket_state.stay_connected.load(Ordering::SeqCst) {
@@ -162,7 +132,7 @@ pub async fn remote_video_loop(
                     }
 
                     // rate limit updates
-                    // sleep_until(Instant::now() + Duration::from_millis(100)).await;
+                    sleep_until(Instant::now() + Duration::from_millis(50)).await;
                 }
             }
         }
@@ -197,6 +167,55 @@ pub async fn remote_video_loop(
     socket_state.is_connected.store(false, Ordering::SeqCst);
 }
 
+#[instrument]
+async fn listen_to_messages(
+    mut recvr: SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>,
+    to_mec: Sender<ApplicationEvent>,
+    tracker_state: Arc<Mutex<TrackerState>>,
+    socket_state: Arc<SocketState>,
+) {
+    info!("Starting tracker connection listen");
+    while socket_state.stay_connected.load(Ordering::SeqCst) {
+        match recvr.try_next().await {
+            Ok(Some(message)) => {
+                let (x_off, y_off, _do_send) =
+                    process_incoming_string(message.to_string(), &tracker_state)
+                        .and_then(|_| calculate_tracking(&tracker_state))
+                        .unwrap_or((0, 0, false));
+
+                let do_send = true;
+
+                // For some reason, this do_send is inverted from what it should be
+                // info!("Do Send is: {}", do_send.to_string());
+                if do_send {
+                    if let Err(e) = to_mec
+                        .send(ApplicationEvent::MoveEvent(
+                            super::MoveEvent { x: x_off, y: y_off },
+                            super::ConnectionType::Automated,
+                        ))
+                        .await
+                    {
+                        error!("Could not send message to MEC, assuming critical failure: {e}");
+                        socket_state.is_connected.store(false, Ordering::SeqCst);
+                        socket_state.stay_connected.store(false, Ordering::SeqCst);
+                        return;
+                    }
+                }
+            }
+            Ok(None) => {
+                info!("Recieved an empty message from the remote computer: Aborting");
+                break;
+            }
+            Err(e) => {
+                error!("Got an error on while recieving from remote computer: {e}");
+            }
+        }
+    }
+    info!(
+        "Stopping tracker listen connection with keep alive: {}",
+        socket_state.stay_connected.load(Ordering::SeqCst)
+    );
+}
 
 fn get_video_frame(
     appsink: &AppSink,

src/gstreamer_pipeline.rs

@@ -16,6 +16,7 @@ pub struct WebcamPipeline {
     pub queue_app: Element,
     pub sink_paintable: Element,
 
+    pub queue: Element,
     pub resize: Element,
     pub sink_frame: Arc<Mutex<AppSink>>,
 }
@@ -46,14 +47,9 @@ impl WebcamPipeline {
             .build()
             .context(BuildSnafu { element: "tee" })?;
 
-        let queue_app = ElementFactory::make("queue")
-            .property("max-size-time", 1u64)
-            .property("max-size-buffers", 0u32)
-            .property("max-size-bytes", 0u32)
-            .build()
-            .context(BuildSnafu {
-                element: "paintable queue",
-            })?;
+        let queue_app = ElementFactory::make("queue").build().context(BuildSnafu {
+            element: "paintable queue",
+        })?;
         let sink_paintable = ElementFactory::make("gtk4paintablesink")
             .name("gtk4_output")
             .build()
@@ -61,9 +57,9 @@ impl WebcamPipeline {
                 element: "gtkpaintablesink",
             })?;
 
-        // queue.connect_closure("overrun", false, glib::closure!(|queue: Element| {
-        //     println!("The queue is full!");
-        // }));
+        let queue = ElementFactory::make("queue").build().context(BuildSnafu {
+            element: "appsink queue",
+        })?;
 
         let resize = ElementFactory::make("videoscale")
             .build()
@@ -71,7 +67,7 @@ impl WebcamPipeline {
                 element: "videoscale",
             })?;
 
-        let caps_string = "video/x-raw,format=RGB,width=640,height=480";
+        let caps_string = "video/x-raw,format=RGB,width=640,height=480,max-buffers=1,drop=true";
         // let caps_string = String::from("video/x-raw,format=RGB,max-buffers=1,drop=true");
         let appsrc_caps = gstreamer::Caps::from_str(caps_string).context(BuildSnafu {
             element: "appsink caps",
@@ -80,7 +76,7 @@ impl WebcamPipeline {
         let sink_frame = AppSink::builder()
             .name("frame_output")
             .sync(false)
-            .max_buffers(3u32)
+            .max_buffers(1u32)
             .drop(true)
             .caps(&appsrc_caps)
             .build();
@@ -96,6 +92,7 @@ impl WebcamPipeline {
                 &queue_app,
                 &sink_paintable,
                 &resize,
+                &queue,
                 &sink_frame.upcast_ref(),
             ])
             .context(LinkSnafu {
@@ -150,19 +147,23 @@ impl WebcamPipeline {
             .ok_or(PipelineError::PadRequestError {
                 element: "tee pad 2".to_string(),
             })?;
-        let appsink_resize_sinkpad =
-            resize
+        let appsink_queue_sinkpad =
+            queue
                 .static_pad("sink")
                 .ok_or(PipelineError::PadRequestError {
                     element: "appsink queue".to_string(),
                 })?;
         tee_src_2
-            .link(&appsink_resize_sinkpad)
+            .link(&appsink_queue_sinkpad)
             .context(PadLinkSnafu {
                 from: "tee src pad 2",
                 to: "appsink queue sinkpad",
             })?;
 
+        queue.link(&resize).context(LinkSnafu {
+            from: "appsink queue",
+            to: "videoscale",
+        })?;
         resize.link(&sink_frame).context(LinkSnafu {
             from: "videoscale",
             to: "appsink",
@@ -176,6 +177,7 @@ impl WebcamPipeline {
             queue_app,
             sink_paintable,
             resize,
+            queue,
             sink_frame: Arc::new(Mutex::new(sink_frame)),
         })
     }

src/ui/mod.rs

@@ -274,13 +274,7 @@ pub fn build_ui(app: &Application, config: Arc<RwLock<AppConfig>>, runtime: Hand
                 tracker_enable_toggle.set_label("Connect to Tracker Computer");
                 tracker_enable_toggle.set_active(false);
                 tracker_enable_toggle.set_sensitive(true);
-            } else if reader.contains("Degraded") {
-                tracker_status_label.set_css_classes(&["LoadingConnection"]);
-
-                tracker_enable_toggle.set_label("Disconnect Tracker Computer");
-                tracker_enable_toggle.set_active(true);
-                tracker_enable_toggle.set_sensitive(true);
-            } else if reader.contains("Connecting") {
+            } else if reader.contains("Degraded") || reader.contains("Connecting") {
                 tracker_status_label.set_css_classes(&["LoadingConnection"]);
 
                 tracker_enable_toggle.set_label("Please Wait");
@@ -360,6 +354,7 @@ fn draw_boxes(width: i32, height: i32, ctx: &Context, tracker_state: &Arc<Mutex<
     #[cfg(feature = "tracker-state-debug")]
     debug!("Getting tracker state for drawing boxes");
     if let Ok(ts) = tracker_state.lock() {
+        println!("ts currently tracking: {}", ts.tracking_id);
         let active = ts.tracking_id;
         let highlighted_id = ts.highlighted_id.unwrap_or(0);
         for nb in ts.identity_boxes.iter() {

todo.md

@@ -1,11 +0,0 @@
-# Functional
-- Up-direction maxes at -50 instead of 100
-- Tracking lag real issue
-
-## QoL
-- Fine tuning the tracking speeds to be non-linear, make sure the pi doesn't have that speed cap (remember could be expecting 6v max speed).
-- During connection waits, add loading dots
-
-
-## Future ideas
-- person recognition