Implementation of front end and backend for window node

Author: hsn200406

backend/shared-logic/src/bc.rs

@@ -2,7 +2,7 @@ use tokio::sync::broadcast;
 use tokio::sync::broadcast::Receiver;
 
 use crate::mockeeg::{generate_mock_data};
-use crate::lsl::{EEGDataPacket, ProcessingConfig, receive_eeg};
+use crate::lsl::{EEGDataPacket, ProcessingConfig, WindowingConfig, receive_eeg};
 use crate::db::{insert_batch_eeg, get_db_client};
 use futures_util::stream::SplitSink;
 use futures_util::{SinkExt};
@@ -15,6 +15,7 @@ use tokio::sync::Mutex;
 use std::sync::Arc;
 use std::time::Instant;
 use tokio_util::sync::CancellationToken;
+use tokio::sync::watch;
 
 use log::{info, error};
 
@@ -23,6 +24,7 @@ pub async fn start_broadcast(
     write: Arc<Mutex<SplitSink<WebSocketStream<TcpStream>, Message>>>,  
     cancel_token: CancellationToken,
     processing_config: ProcessingConfig, // takes in signal processing configuration from frontend
+    windowing_rx: watch::Receiver<WindowingConfig> // takes in windowing configuration from frontend
 ) {
     let (tx, _rx) = broadcast::channel::<Arc<EEGDataPacket>>(1000); // size of the broadcast buffer, not recommand below 500, websocket will miss messages
     let rx_ws = tx.subscribe();
@@ -42,7 +44,7 @@ pub async fn start_broadcast(
     let sender_token = cancel_token.clone(); 
     let sender = tokio::spawn(async move {
         // use the ProcessingConfig provided by the client instead of default
-        receive_eeg(tx_clone, sender_token, processing_config).await;
+        receive_eeg(tx_clone, sender_token, processing_config, windowing_rx).await;
     });
 
     // Subscribe for websocket Receiver 

backend/shared-logic/src/db.rs

@@ -353,7 +353,7 @@ pub async fn delete_session(client: &DbClient, session_id: i32) -> Result<(), Er
         .execute(&**client)
         .await?;
 
-    if (res.rows_affected() == 0) {
+    if res.rows_affected() == 0 {
         info!("No rows deleted, session id {} not found", session_id);
         return Err(Error::RowNotFound);
     } else {

backend/shared-logic/src/lsl.rs

@@ -39,8 +39,23 @@ impl Default for ProcessingConfig {
     }
 }
 
+#[derive(Clone, Deserialize, Debug)]
+pub struct WindowingConfig {
+    pub chunk_size: usize,
+    pub overlap_size: usize,
+}
+
+impl Default for WindowingConfig {
+    fn default() -> Self {
+        Self {
+            chunk_size: 64,
+            overlap_size: 0,
+        }
+    }
+}
+
 // Async entry point for EEG data collection.
-pub async fn receive_eeg(tx:Sender<Arc<EEGDataPacket>>, cancel_token: CancellationToken, processing_config: ProcessingConfig) {
+pub async fn receive_eeg(tx:Sender<Arc<EEGDataPacket>>, cancel_token: CancellationToken, processing_config: ProcessingConfig, windowing_rx: tokio::sync::watch::Receiver<WindowingConfig>,) {
     info!("Starting EEG data receiver");
     let python_script_path = std::env::var("SIGNAL_PROCESSING_SCRIPT")
     .unwrap_or_else(|_| "../shared-logic/src/signal_processing/signalProcessing.py".to_string());
@@ -67,7 +82,7 @@ pub async fn receive_eeg(tx:Sender<Arc<EEGDataPacket>>, cancel_token: Cancellati
         };
 
         // Run collection loop
-        run_eeg_collection(inlet, tx, cancel_token, processing_config, sig_processor)
+        run_eeg_collection(inlet, tx, cancel_token, processing_config, sig_processor, windowing_rx)
     });
 
     // Handle results
@@ -100,16 +115,39 @@ fn setup_eeg_stream() -> Result<StreamInlet, String> {
 
 // Main EEG data collection loop.
 // Returns (successful_count, dropped_count) statistics.
-fn run_eeg_collection(inlet: StreamInlet, tx: Sender<Arc<EEGDataPacket>>, cancel_token: CancellationToken, config: ProcessingConfig, processor: SignalProcessor) -> (u32, u32) {
+fn run_eeg_collection(inlet: StreamInlet,
+    tx: Sender<Arc<EEGDataPacket>>, 
+    cancel_token: CancellationToken,
+    config: ProcessingConfig,
+    processor: SignalProcessor,
+    mut windowing_rx: tokio::sync::watch::Receiver<WindowingConfig>,
+) -> (u32, u32) {
     let mut count = 0;
     let mut drop = 0;
+
+    
+    let mut windowing = windowing_rx.borrow().clone();
+
+    // Creates a buffer that stores overlapping eeg samples
+    let mut overlap_buffer: Vec<Vec<f64>> = vec![Vec::new(); 4];
+
     let mut packet = EEGDataPacket {
-        timestamps: Vec::with_capacity(65),
-        signals: vec![Vec::with_capacity(65); 4],
+        timestamps: Vec::with_capacity(windowing.chunk_size + 1),
+        signals: vec![Vec::with_capacity(windowing.chunk_size + 1); 4],
     };
+
     // Calculate the offset between LSL clock and Unix epoch
     let lsl_to_unix_offset = Utc::now().timestamp_nanos_opt().unwrap() as f64 / 1_000_000_000.0 - lsl::local_clock();
     loop {
+        if windowing_rx.has_changed().unwrap_or(false) {
+            windowing = windowing_rx.borrow().clone();
+            info!("Windowing config updated: chunk={}, overlap={}", windowing.chunk_size, windowing.overlap_size);
+            // Discard old buffer and start fresh with new config
+            packet.timestamps.clear();
+            for ch in &mut packet.signals { ch.clear(); }
+            for ch in &mut overlap_buffer { ch.clear(); }
+        }
+
         // Check for cancellation
         if cancel_token.is_cancelled() {
             info!("EEG data receiver cancelled.");
@@ -130,10 +168,34 @@ fn run_eeg_collection(inlet: StreamInlet, tx: Sender<Arc<EEGDataPacket>>, cancel
         // Pull sample with timeout of 1 sec. If it does not see data for 1s, it returns.
         match inlet.pull_sample(1.0) {
            Ok((sample, timestamp)) => {
-                match accumulate_sample(&sample, timestamp + lsl_to_unix_offset, &mut packet) {
+                match accumulate_sample(&sample, timestamp + lsl_to_unix_offset, &mut packet, windowing.chunk_size) {
                     Ok(true) => {
+                        // Window is full. Prepend overlap from previous window if there are any
+                        if windowing.overlap_size > 0 && !overlap_buffer[0].is_empty() {
+                            // Insert overlap samples at the front of the packet
+                            for (ch_idx, ch) in packet.signals.iter_mut().enumerate() {
+                                let mut new_ch = overlap_buffer[ch_idx].clone();
+                                new_ch.extend_from_slice(ch);
+                                *ch = new_ch;
+                            }
+
+                            // Timestamps: prepend placeholders (or track overlap timestamps)
+                            // For simplicity, pad with copies of the first timestamp
+                            let first_ts = packet.timestamps[0];
+                            let mut new_ts = vec![first_ts; windowing.overlap_size];
+                            new_ts.extend_from_slice(&packet.timestamps);
+                            packet.timestamps = new_ts;
+                        }
+
+                        // Save the tail as the new overlap_buffer
+                        let n = packet.signals[0].len();
+                        let keep = windowing.overlap_size.min(n);
+                        for (ch_idx, ch) in packet.signals.iter().enumerate() {
+                            overlap_buffer[ch_idx] = ch[n - keep..].to_vec();
+                        }
+                        
                         // Packet is full, send it
-                        info!("Packet is full, send it");
+                        info!("Packet is full, sending window: {} samples (overlap: {})", packet.signals[0].len(), keep);
                          match process_and_send(&mut packet, &processor, &config, &tx) {
                             Ok(_) => count += 1,
                             Err(e) => {
@@ -146,9 +208,7 @@ fn run_eeg_collection(inlet: StreamInlet, tx: Sender<Arc<EEGDataPacket>>, cancel
                             channel.clear();  
                         }
                     }
-                    Ok(false) => {
-                        // Sample added, but packet not full yet
-                    }
+                    Ok(false) => {} // Sample added, but packet not full yet
                     Err(e) => {
                          let error_msg = e.to_string();
                         if error_msg.contains("Invalid sample length: got 0 channels") {
@@ -179,7 +239,8 @@ fn run_eeg_collection(inlet: StreamInlet, tx: Sender<Arc<EEGDataPacket>>, cancel
 fn accumulate_sample(
     sample: &[f32], 
     timestamp: f64, 
-    packet: &mut EEGDataPacket
+    packet: &mut EEGDataPacket,
+    chunk_size: usize,
 ) -> Result<bool, String> {
     // Validate sample length
     if sample.len() < 4 {
@@ -188,7 +249,7 @@ fn accumulate_sample(
 
     // Convert timestamp
     let timestamp_dt = DateTime::from_timestamp(
-        timestamp as i64, 
+        timestamp as i64,
         (timestamp.fract() * 1_000_000_000.0) as u32
     ).unwrap_or_else(|| Utc::now());
 
@@ -202,7 +263,7 @@ fn accumulate_sample(
     }
 
     // Check if packet is full
-    Ok(packet.signals[0].len() >= 65)
+    Ok(packet.signals[0].len() >= chunk_size)
 }
 
 // calls the signalProcessing.py to process the packet and sends it

backend/websocket-server/src/main.rs

@@ -1,9 +1,10 @@
-use std::os::windows::process;
+// use std::os::windows::process;
 use std::{sync::Arc};
 use futures_util::stream::SplitSink;
 use futures_util::{SinkExt, StreamExt};
 use tokio::net::{TcpListener, TcpStream};
 use tokio::sync::Mutex;
+use tokio::sync::watch;
 use tokio_tungstenite::{
     accept_async,
     tungstenite::{Message},
@@ -12,10 +13,11 @@ use tokio_tungstenite::{
 use tokio_util::sync::CancellationToken;
 use shared_logic::bc::{start_broadcast};
 use shared_logic::db::{initialize_connection};
-use shared_logic::lsl::{ProcessingConfig}; // get ProcessingConfig from lsl.rs
+use shared_logic::lsl::{ProcessingConfig, WindowingConfig}; // get ProcessingConfig from lsl.rs
 use dotenvy::dotenv;
 use log::{info, error};
 use serde_json; // used to parse ProcessingConfig from JSON sent by frontend
+use serde_json::Value; // used to parse ProcessingConfig from JSON sent by frontend
 
 
 #[tokio::main]
@@ -75,83 +77,128 @@ async fn handle_ws(stream: TcpStream) {
     }
 }
 
-// handle_connection, starts a async broadcast task, 
-// then listens for incoming websocket closing request with the read stream in order to stop the broadcast task.
 async fn handle_connection(ws_stream: WebSocketStream<TcpStream>) {
-    let ( write, mut read) = ws_stream.split();
-    // set up for the broadcast task
-    let write = Arc::new(Mutex::new(write)); 
+    let (write, mut read) = ws_stream.split();
+    let write = Arc::new(Mutex::new(write));
     let write_clone = write.clone();
     let cancel_token = CancellationToken::new();
     let cancel_clone = cancel_token.clone();
 
-    // setup registration for signal processing configuration
-    let signal_config = read.next().await;
+    let mut processing_config = ProcessingConfig::default();
+    let mut initial_windowing = WindowingConfig::default();
 
-    // we have the ProcessingConfig struct
-    // check if we received a message (two layers of unwrapping needed)
-    let processing_config: ProcessingConfig = match signal_config {
+    // Give the frontend a short window to send configs before we start
+    // Use a timeout so we don't block forever if only one config arrives
+    let config_timeout = tokio::time::Duration::from_millis(500);
+    let deadline = tokio::time::Instant::now() + config_timeout;
 
-        Some(Ok(config_json)) => {
-
-            // here, we parse the json into a signal config struct using serde_json
-            let config_text = config_json.to_text().unwrap();
+    loop {
+        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+        if remaining.is_zero() {
+            info!("Config window elapsed, starting with current configs.");
+            break;
+        }
 
-            match serde_json::from_str(config_text) {
-                Ok(config) => config,
-                Err(e) => {
-                    error!("Error parsing signal configuration JSON: {}", e);
+        match tokio::time::timeout(remaining, read.next()).await {
+            Ok(Some(Ok(msg))) if msg.is_text() => {
+                let text = msg.to_text().unwrap();
+                if text == "clientClosing" {
+                    info!("Client closing during config phase.");
                     return;
                 }
+                match parse_config_message(text) {
+                    ConfigMessage::Processing(cfg) => {
+                        info!("Received ProcessingConfig");
+                        processing_config = cfg;
+                    }
+                    ConfigMessage::Windowing(cfg) => {
+                        info!("Received WindowingConfig: chunk={}, overlap={}", cfg.chunk_size, cfg.overlap_size);
+                        initial_windowing = cfg;
+                    }
+                    ConfigMessage::Unknown => {
+                        info!("Non-config message during config phase: {}", text);
+                        break;
+                    }
+                }
             }
-
+            Ok(Some(Err(e))) => { error!("Error receiving config: {}", e); return; }
+            Ok(None) => { error!("Connection closed during config phase."); return; }
+            Err(_) => {
+                // Timeout elapsed
+                info!("Config timeout, starting with received configs.");
+                break;
+            }
+            Ok(_) => {}
         }
+    }
 
-        Some(Err(e)) => {
-            error!("Error receiving signal configuration: {}", e);
-            return;
-        }
+    info!("Starting broadcast with chunk={}, overlap={}", initial_windowing.chunk_size, initial_windowing.overlap_size);
 
-        None => {
-            error!("No signal configuration received from client. Closing connection.");
-            return;
-        }
-    };
+    let (windowing_tx, windowing_rx) = watch::channel(initial_windowing);
 
-    // spawns the broadcast task
     let mut broadcast = Some(tokio::spawn(async move {
-        // pass ProcessingConfig into broadcast so it reaches receive_eeg
-        start_broadcast(write_clone, cancel_clone, processing_config).await;
+        start_broadcast(write_clone, cancel_clone, processing_config, windowing_rx).await;
     }));
 
-
-    //listens for incoming messages 
     while let Some(msg) = read.next().await {
         match msg {
-            Ok(msg) if msg.is_text() => { //prep for closing, this currently will not be called, waiting for frontend
+            Ok(msg) if msg.is_text() => {
                 let text = msg.to_text().unwrap();
-                info!("Received request: {}", text);
                 if text == "clientClosing" {
-                    handle_prep_close(&mut broadcast,&cancel_token, &write.clone()).await;
+                    handle_prep_close(&mut broadcast, &cancel_token, &write).await;
+                    break;
+                }
+                match parse_config_message(text) {
+                    ConfigMessage::Windowing(cfg) => {
+                        info!("Windowing update: chunk={}, overlap={}", cfg.chunk_size, cfg.overlap_size);
+                        let _ = windowing_tx.send(cfg);
+                    }
+                    ConfigMessage::Processing(_) => {
+                        info!("Processing config update received");
+                    }
+                    ConfigMessage::Unknown => {
+                        error!("Unknown mid-stream message: {}", text);
+                    }
                 }
             }
-            Ok(Message::Close(frame)) => { //handles closing.
-                info!("Received a close request from the client");
-                // cancel_token.cancel(); // remove after frontend updates
-                let mut write = write.lock().await;
-                let _ = write.send(Message::Close(frame)).await;
+            Ok(Message::Close(frame)) => {
+                let mut w = write.lock().await;
+                let _ = w.send(Message::Close(frame)).await;
                 break;
             }
             Ok(_) => continue,
-            Err(e) => {
-                error!("Read error (client likely disconnected): {}", e);
-                break;
-            }
+            Err(e) => { error!("Read error: {}", e); break; }
         }
     }
     info!("Client disconnected.");
 }
 
+// Discriminate config type by which fields are present
+enum ConfigMessage {
+    Processing(ProcessingConfig),
+    Windowing(WindowingConfig),
+    Unknown,
+}
+
+fn parse_config_message(text: &str) -> ConfigMessage {
+    let Ok(value) = serde_json::from_str::<Value>(text) else {
+        return ConfigMessage::Unknown;
+    };
+    if value.get("chunk_size").is_some() {
+        match serde_json::from_value::<WindowingConfig>(value) {
+            Ok(cfg) => ConfigMessage::Windowing(cfg),
+            Err(e) => { error!("Failed to parse WindowingConfig: {}", e); ConfigMessage::Unknown }
+        }
+    } else if value.get("apply_bandpass").is_some() {
+        match serde_json::from_value::<ProcessingConfig>(value) {
+            Ok(cfg) => ConfigMessage::Processing(cfg),
+            Err(e) => { error!("Failed to parse ProcessingConfig: {}", e); ConfigMessage::Unknown }
+        }
+    } else {
+        ConfigMessage::Unknown
+    }
+}
+
 // handle_prep_close uses the cancel_token to stop the broadcast sender task, and sends a "prep close complete" message to the client
 async fn handle_prep_close(
     broadcast_task: &mut Option<tokio::task::JoinHandle<()>>,