EEG Export and Import Update

backend/api-server/Cargo.toml

@@ -33,4 +33,7 @@ pyo3 = { version = "0.18.0", features = ["auto-initialize"] }
 env_logger = "0.11"
 
 # Shared logic crate
-shared-logic = { path = "../shared-logic" }
+shared-logic = { path = "../shared-logic" }
+
+# CSV serialization/deserialization
+csv = "1.4"

backend/api-server/src/main.rs

@@ -5,6 +5,7 @@ use axum::{
     routing::{get, post},
     Json,
     Router,
+    body::Bytes,
 };
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Value};
@@ -17,11 +18,14 @@ use pyo3::Python;
 use pyo3::types::{PyList, PyModule, PyTuple};
 use pyo3::PyResult;
 use pyo3::{IntoPy, ToPyObject};
+use chrono::{DateTime, Utc};
+use axum::http::{HeaderMap, HeaderValue, header};
+use axum::response::IntoResponse;
 use rand_core::OsRng;
 
 // shared logic library
-use shared_logic::db::{initialize_connection, DbClient};
-use shared_logic::models::{User, NewUser, UpdateUser, Session, FrontendState};
+use shared_logic::db::{DbClient, initialize_connection, export_eeg_data_as_csv, get_earliest_eeg_timestamp};
+use shared_logic::models::{NewUser, Session, FrontendState};
 
 // Argon2 imports
 use argon2::{
@@ -36,6 +40,21 @@ struct AppState {
     db_client: DbClient,
 }
 
+// define request struct for exporting EEG data
+#[derive(Deserialize)]
+struct ExportEEGRequest {
+    filename: String,
+    options: ExportOptions
+}
+
+#[derive(Deserialize)]
+struct ExportOptions {
+    format: String,
+    includeHeader: bool,
+    start_time: Option<DateTime<Utc>>,
+    end_time: Option<DateTime<Utc>>,
+}
+
 
 #[derive(Debug, Clone, Deserialize)]
 pub struct LoginRequest {
@@ -55,6 +74,40 @@ struct DeleteUserRequest {
     id: i32,
 }
 
+/// Helper function for eeg data to get the start and end timestamps for a given session
+/// 
+/// Returns the start and end timestamps on success.
+pub async fn get_eeg_time_range(
+    client: &DbClient,
+    session_id: i32,
+    options: &ExportOptions,
+) -> Result<(DateTime<Utc>, DateTime<Utc>), (StatusCode, String)> {
+    // check for time range, else use defaults
+    // for end time, we default to the current time
+    // for start time, we default to the earliest timestamp for the session
+    let end_time = match options.end_time {
+        Some(t) => t,
+        None => Utc::now(),
+    };
+
+    let start_time = match options.start_time {
+        Some(t) => t,
+        None => {
+            // we call the helper function above to get the earliest timestamp
+            match get_earliest_eeg_timestamp(client, session_id).await {
+                Ok(Some(t)) => t,
+                Ok(None) => return Err((StatusCode::NOT_FOUND, format!("No EEG data found for session {}", session_id))),
+                Err(e) => return Err((StatusCode::INTERNAL_SERVER_ERROR, format!("Failed to get earliest EEG timestamp: {}", e))),
+            }
+        }
+    };
+
+    if start_time > end_time {
+        return Err((StatusCode::BAD_REQUEST, "start_time cannot be after end_time".to_string()));
+    }
+
+    Ok((start_time, end_time))
+}
 
 // creates new user when POST /users is called
 async fn create_user(
@@ -247,6 +300,65 @@ async fn get_frontend_state(
     }
 }
 
+// Handler for POST /api/sessions/{session_id}/eeg_data/export
+async fn export_eeg_data(
+    State(app_state): State<AppState>,
+    Path(session_id): Path<i32>,
+    Json(request): Json<ExportEEGRequest>,
+) -> Result<impl IntoResponse, (StatusCode, String)> {
+    info!("Received request to export EEG data for session {}", session_id);
+
+    // right now the only export format supported is CSV, so we just check for that
+    if request.options.format.to_lowercase() != "csv" {
+        return Err((StatusCode::BAD_REQUEST, format!("Unsupported export format: {}", request.options.format)));
+    }
+
+    let (start_time, end_time) =
+        get_eeg_time_range(&app_state.db_client, session_id, &request.options).await?;
+
+    let header_included = request.options.includeHeader;
+
+    // finally call the export function in db.rs
+    let return_csv = match export_eeg_data_as_csv(&app_state.db_client, session_id, start_time, end_time, header_included).await {
+        Ok(csv_data) => csv_data,
+        Err(e) => {
+            error!("Failed to export EEG data: {}", e);
+            return Err((StatusCode::INTERNAL_SERVER_ERROR, format!("Failed to export EEG data: {}", e)));
+        }
+    };
+
+    // small safety: avoid quotes breaking header
+    let filename = request.filename.replace('"', "");
+
+    let mut headers = HeaderMap::new();
+    headers.insert(header::CONTENT_TYPE, HeaderValue::from_static("text/csv; charset=utf-8"));
+
+    let content_disp = format!("attachment; filename=\"{}\"", filename);
+    headers.insert(
+        header::CONTENT_DISPOSITION,
+        HeaderValue::from_str(&content_disp).map_err(|e| {
+            (StatusCode::BAD_REQUEST, format!("Invalid filename for header: {}", e))
+        })?,
+    );
+
+    // return CSV directly as the body
+    Ok((headers, return_csv))
+
+}
+
+// Handler for POST /api/sessions/{session_id}/eeg_data/import
+async fn import_eeg_data(
+    State(app_state): State<AppState>,
+    Path(session_id): Path<i32>,
+    // we expect the CSV data to be sent as raw text in the body of the request
+    body: Bytes,
+) -> Result<Json<Value>, (StatusCode, String)> {
+    shared_logic::db::import_eeg_data_from_csv(&app_state.db_client, session_id, &body)
+        .await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("Failed to import EEG data: {}", e)))?;
+
+    Ok(Json(json!({"status": "success"})))
+}
 
 
 
@@ -353,6 +465,9 @@ async fn main() {
         .route("/api/sessions/:session_id/frontend-state", post(set_frontend_state))
         .route("/api/sessions/:session_id/frontend-state", get(get_frontend_state))
 
+        .route("/api/sessions/:session_id/eeg_data/export", post(export_eeg_data))
+        .route("/api/sessions/:session_id/eeg_data/import", post(import_eeg_data))
+
         // Share application state with all handlers
         .with_state(app_state);
 

backend/migrations/20260301000000_unique_eeg_session_time.sql

@@ -0,0 +1,4 @@
+-- we enforce uniqueness on imports through imposing a unique constraint on the 
+-- combination of session_id and time, preventing duplicate entries for the same session and timestamp
+ALTER TABLE eeg_data
+ADD CONSTRAINT uq_eeg_session_time UNIQUE (session_id, time);

backend/migrations/20263101120000_sessions_on_eeg_data.sql

@@ -0,0 +1,8 @@
+-- note that this assumes that eeg_data has no rows (since as of this migration there should be no real data yet)
+-- to do so just run TRUNCATE TABLE eeg_data before applying this migration
+ALTER TABLE eeg_data
+ADD COLUMN session_id INTEGER NOT NULL,
+ADD CONSTRAINT fk_session FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE;
+
+-- we can create an index on session_id and time, since the bulk of our queries will be filtering based on these
+CREATE INDEX eeg_data_session_time_idx ON eeg_data (session_id, time DESC); -- using DESC since i'm expecting recent data to be more relevant

backend/migrations/20263103300000_add_auth_column.sql

@@ -0,0 +1,2 @@
+ALTER TABLE users
+ADD COLUMN password_hash VARCHAR(257) NOT NULL DEFAULT '';

backend/shared-logic/Cargo.toml

@@ -48,4 +48,7 @@ rand_core = "0.6"
 
 # working with python 
 pyo3 = { version = "0.18.0", features = ["auto-initialize"] }
-numpy = "0.18"
+numpy = "0.18"
+
+# CSV serialization/deserialization
+csv = "1.4"

backend/shared-logic/src/bc.rs

@@ -24,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
+    session_id: i32, // takes in session id to tag incoming data with the correct session
     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
@@ -55,7 +56,7 @@ pub async fn start_broadcast(
 
     // Subscribe for database Receiver 
     tokio::spawn(async move { 
-        db_receiver( rx_db).await;
+        db_receiver( rx_db, session_id).await;
     });
 
     //waits for sender to complete. 
@@ -111,7 +112,10 @@ pub async fn ws_receiver(write: &Arc<Mutex<SplitSink<WebSocketStream<TcpStream>,
 
 //db_broadcast_receiver takes EEGDataPacket  struct from the broadcast sender and inserts it into the database
 // it inserts as a batch of 100.
-pub async fn db_receiver(mut rx_db: Receiver<Arc<EEGDataPacket>>){
+pub async fn db_receiver(
+    mut rx_db: Receiver<Arc<EEGDataPacket>>,
+    session_id: i32,
+){
     let db_client = get_db_client();
 
     let mut packet_count = 0; // for debug purposes
@@ -131,7 +135,7 @@ pub async fn db_receiver(mut rx_db: Receiver<Arc<EEGDataPacket>>){
                 // Insert the packet directly
                 tokio::spawn(async move {
                     let now = Instant::now(); // for debug purposes
-                    if let Err(e) = insert_batch_eeg(&db_client_clone, &eeg_packet).await {
+                    if let Err(e) = insert_batch_eeg(&db_client_clone, session_id, &eeg_packet).await {
                         error!("Packet insert failed: {:?}", e);
                     }
                     info!("Packet insert took {:?}", now.elapsed()); // for debug purposes