From ed7dbcf1c9f287da805efeaca11a78099bb42eba Mon Sep 17 00:00:00 2001 From: "Jan-Peter v. Hunnius" Date: Wed, 24 Jun 2026 09:32:16 +0200 Subject: [PATCH] feat: persistent sessions, remove track_map use case, car characteristics DB - Remove obsolete TrackMap use case variant (superseded by TrackMapping) - Add session/run/analysis persistence: metadata to data/sessions/.json, frames to .frames.json (split for lazy loading) - Atomic writes (temp + rename), compact JSON, off-thread frame serialization - load_all loads metadata only at startup; ensure_frames_loaded lazy-loads frames before analysis - Normalize Recording runs to Stopped on reload (stale ring-buffer markers) - Cache car_name on UseCaseSession during run finalization - Add characteristics.rs: aggregate shift-point data per car across sessions - New API: GET /api/characteristics/cars, GET /api/characteristics/cars/{car} - Add Deserialize to TelemetryFrame and FeedSource for session round-trip --- src/analysis.rs | 2 +- src/api.rs | 23 +++- src/characteristics.rs | 211 ++++++++++++++++++++++++++++++++++++ src/main.rs | 10 +- src/recording.rs | 3 - src/session.rs | 239 +++++++++++++++++++++++++++++++++++++++-- src/telemetry_frame.rs | 6 +- 7 files changed, 477 insertions(+), 17 deletions(-) create mode 100644 src/characteristics.rs diff --git a/src/analysis.rs b/src/analysis.rs index c03c1f6..67642ae 100644 --- a/src/analysis.rs +++ b/src/analysis.rs @@ -735,7 +735,7 @@ pub fn analyze( .unwrap_or_else(|_| serde_json::json!({"error": "serialization failed"})) } } - UseCase::TrackMap | UseCase::TrackMapping => { + UseCase::TrackMapping => { serde_json::to_value(analyze_track_map(frames_by_run, run_labels)) .unwrap_or_else(|_| serde_json::json!({"error": "serialization failed"})) } diff --git a/src/api.rs b/src/api.rs index 808aeee..79d7869 100644 --- a/src/api.rs +++ b/src/api.rs @@ -14,6 +14,7 @@ use std::sync::{Arc, Mutex, MutexGuard}; use tower_http::cors::CorsLayer; use crate::analysis; +use crate::characteristics; use crate::feed_state::FeedState; use crate::recording::{Game, RecordingManager, RecordingStatus, StartRecordingRequest, UseCase}; use crate::session::SessionManager; @@ -267,7 +268,7 @@ async fn run_analysis( Path(id): Path, ) -> Response { let (use_case, frames_by_run, run_labels, recording_id) = { - let mgr = lock_sessions(&state); + let mut mgr = lock_sessions(&state); match mgr.get_session(&id) { Some(session) => { if session.status != crate::session::SessionStatus::Completed { @@ -681,6 +682,24 @@ async fn list_games() -> Response { ok_json(Game::valid_choices()) } +async fn list_car_profiles( + State(state): State>, +) -> Response { + let mgr = lock_sessions(&state); + ok_json(characteristics::car_profiles(&mgr)) +} + +async fn get_car_profile( + State(state): State>, + Path(car): Path, +) -> Response { + let mgr = lock_sessions(&state); + match characteristics::car_profile(&mgr, &car) { + Some(profile) => ok_json(profile), + None => err(StatusCode::NOT_FOUND, &format!("No data for car '{}'", car)), + } +} + pub fn build_router(state: Arc) -> Router { Router::new() .route("/api/sessions", get(list_sessions).post(create_session)) @@ -714,6 +733,8 @@ pub fn build_router(state: Arc) -> Router { .route("/api/tracks/build", post(build_track)) .route("/api/use-cases", get(list_use_cases)) .route("/api/games", get(list_games)) + .route("/api/characteristics/cars", get(list_car_profiles)) + .route("/api/characteristics/cars/{car}", get(get_car_profile)) .layer(CorsLayer::permissive()) .with_state(state) } diff --git a/src/characteristics.rs b/src/characteristics.rs new file mode 100644 index 0000000..4a88716 --- /dev/null +++ b/src/characteristics.rs @@ -0,0 +1,211 @@ +//! Car & track characteristics database — server-side aggregation across +//! persisted sessions. +//! +//! Aggregates analysis results (currently shift-point data) grouped by car +//! name to build a characteristics profile. This is a read-only query layer +//! over the in-memory session store. + +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; + +use crate::recording::UseCase; +use crate::session::{SessionManager, SessionStatus}; + +/// Aggregated gear range for a single car, merged across all sessions. +#[derive(Debug, Clone, Serialize)] +pub struct CarGearProfile { + pub gear: i32, + pub gear_display: String, + pub min_rpm: f32, + pub max_rpm: f32, + pub min_speed_kmh: f32, + pub max_speed_kmh: f32, + pub session_count: usize, +} + +/// Aggregated shift-point characteristics for a single car. +#[derive(Debug, Clone, Serialize)] +pub struct CarProfile { + pub car_name: String, + pub session_count: usize, + pub recommended_shift_rpm_avg: Option, + pub recommended_shift_rpm_min: Option, + pub recommended_shift_rpm_max: Option, + pub gear_profiles: Vec, + pub last_session_id: String, + pub last_session_date: String, +} + +/// Typed view of a gear curve from serialized `ShiftPointAnalysis`. +/// Mirrors the fields of `analysis::GearAccelerationCurve` so that renaming +/// a field in the serializer produces a compile-time error here. +#[derive(Deserialize)] +struct GearCurveView { + gear: i32, + min_rpm: f64, + max_rpm: f64, + min_speed_kmh: f64, + max_speed_kmh: f64, +} + +/// Typed view of serialized `ShiftPointAnalysis`. +#[derive(Deserialize)] +struct ShiftPointAnalysisView { + recommended_shift_rpm: Option, + gear_curves: Vec, +} + +/// Per-session shift-point data extracted from analysis JSON. +struct SessionShiftData { + car_name: String, + recommended_shift_rpm: Option, + gear_data: HashMap, + session_id: String, + created_at: String, +} + +/// Collect shift-point data from a completed session's analysis JSON. +fn extract_shift_data( + session: &crate::session::UseCaseSession, +) -> Option { + if session.use_case != UseCase::ShiftPoints { + return None; + } + if session.status != SessionStatus::Completed { + return None; + } + let analysis = session.analysis.as_ref()?; + + let view = serde_json::from_value::(analysis.clone()).ok()?; + + let mut gear_data: HashMap = HashMap::new(); + for curve in view.gear_curves { + gear_data.insert( + curve.gear, + ( + curve.min_rpm as f32, + curve.max_rpm as f32, + curve.min_speed_kmh as f32, + curve.max_speed_kmh as f32, + ), + ); + } + + let car_name = session + .car_name + .clone() + .unwrap_or_else(|| "Unknown".to_string()); + + Some(SessionShiftData { + car_name, + recommended_shift_rpm: view.recommended_shift_rpm.map(|f| f as f32), + gear_data, + session_id: session.id.clone(), + created_at: session.created_at.clone(), + }) +} + +fn gear_display(gear: i32) -> String { + crate::telemetry_frame::gear_display(gear) +} + +/// Aggregate all shift-point sessions into per-car profiles. +pub fn car_profiles(mgr: &SessionManager) -> Vec { + let sessions: Vec = mgr + .iter_sessions() + .filter_map(extract_shift_data) + .collect(); + + let mut by_car: HashMap> = HashMap::new(); + for data in sessions { + by_car.entry(data.car_name.clone()).or_default().push(data); + } + + let mut profiles: Vec = by_car + .into_iter() + .map(|(car_name, sessions)| build_profile(&car_name, &sessions)) + .collect(); + + profiles.sort_by(|a, b| a.car_name.cmp(&b.car_name)); + profiles +} + +/// Aggregate a single car profile by name. +pub fn car_profile(mgr: &SessionManager, car: &str) -> Option { + let sessions: Vec = mgr + .iter_sessions() + .filter_map(extract_shift_data) + .filter(|d| d.car_name.eq_ignore_ascii_case(car)) + .collect(); + + if sessions.is_empty() { + return None; + } + + Some(build_profile(car, &sessions)) +} + +fn build_profile(car_name: &str, sessions: &[SessionShiftData]) -> CarProfile { + let session_count = sessions.len(); + + // Recommended shift RPM aggregation. + let shift_rpms: Vec = sessions + .iter() + .filter_map(|d| d.recommended_shift_rpm) + .collect(); + + let (recommended_avg, recommended_min, recommended_max) = if shift_rpms.is_empty() { + (None, None, None) + } else { + let sum: f32 = shift_rpms.iter().sum(); + let avg = sum / shift_rpms.len() as f32; + let min = shift_rpms.iter().cloned().fold(f32::INFINITY, f32::min); + let max = shift_rpms.iter().cloned().fold(f32::NEG_INFINITY, f32::max); + (Some(avg), Some(min), Some(max)) + }; + + // Gear profile aggregation: merge min/max across sessions per gear. + let mut gear_agg: HashMap = HashMap::new(); + for session in sessions { + for (&gear, &(min_rpm, max_rpm, min_spd, max_spd)) in &session.gear_data { + let entry = gear_agg + .entry(gear) + .or_insert((f32::INFINITY, f32::NEG_INFINITY, f32::INFINITY, f32::NEG_INFINITY, 0)); + entry.0 = entry.0.min(min_rpm); + entry.1 = entry.1.max(max_rpm); + entry.2 = entry.2.min(min_spd); + entry.3 = entry.3.max(max_spd); + entry.4 += 1; + } + } + + let mut gear_profiles: Vec = gear_agg + .into_iter() + .map(|(gear, (min_rpm, max_rpm, min_spd, max_spd, count))| CarGearProfile { + gear, + gear_display: gear_display(gear), + min_rpm, + max_rpm, + min_speed_kmh: min_spd, + max_speed_kmh: max_spd, + session_count: count, + }) + .collect(); + gear_profiles.sort_by_key(|g| g.gear); + + let last = sessions + .iter() + .max_by(|a, b| a.created_at.cmp(&b.created_at)) + .expect("at least one session"); + + CarProfile { + car_name: car_name.to_string(), + session_count, + recommended_shift_rpm_avg: recommended_avg, + recommended_shift_rpm_min: recommended_min, + recommended_shift_rpm_max: recommended_max, + gear_profiles, + last_session_id: last.session_id.clone(), + last_session_date: last.created_at.clone(), + } +} diff --git a/src/main.rs b/src/main.rs index 08ae89f..a43139d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -10,6 +10,7 @@ mod telemetry_frame; mod ui; mod warnings; mod analysis; +mod characteristics; mod session; use feed_state::*; @@ -389,9 +390,16 @@ async fn main() { let recording_manager = Arc::new(recording::RecordingManager::new(&data_dir)); let track_model_store = Arc::new(crate::track_model::TrackModelStore::new(&data_dir)); let session_manager = Arc::new(std::sync::Mutex::new( - SessionManager::new(recording_manager.clone()) + SessionManager::new(recording_manager.clone(), data_dir.clone()) )); + { + let mut mgr = session_manager.lock().unwrap(); + mgr.load_all(); + let count = mgr.list_sessions().len(); + warnings.info(&format!("Loaded {} persisted session(s) from disk", count)); + } + let auto_start = recording::StartRecordingRequest { name: Some( chrono::Local::now() diff --git a/src/recording.rs b/src/recording.rs index 0376cdd..31ba44f 100644 --- a/src/recording.rs +++ b/src/recording.rs @@ -132,7 +132,6 @@ pub enum UseCase { BrakingAnalysis, LapTimes, SectorAnalysis, - TrackMap, TrackMapping, General, } @@ -145,7 +144,6 @@ impl UseCase { UseCase::BrakingAnalysis, UseCase::LapTimes, UseCase::SectorAnalysis, - UseCase::TrackMap, UseCase::TrackMapping, UseCase::General, ] @@ -153,7 +151,6 @@ impl UseCase { pub fn guidance(&self) -> &'static str { match self { UseCase::ShiftPoints => "Record one run per gear: accelerate at full throttle from as low an RPM as possible to the rev limiter (or until you run out of road). Each run should be in a single gear only. Use Hotlap mode on a long straight. Start recording a run just before going WOT, stop after reaching max RPM. Repeat for each gear.", - UseCase::TrackMap => "Record two runs on the same lap: one driving the left boundary, one driving the right boundary. Label each run accordingly.", UseCase::BrakingAnalysis => "Record runs with heavy braking into corners from high speed.", UseCase::LapTimes => "Record one or more complete laps.", UseCase::SectorAnalysis => "Record one or more complete laps.", diff --git a/src/session.rs b/src/session.rs index d6742d3..5d660d9 100644 --- a/src/session.rs +++ b/src/session.rs @@ -4,9 +4,15 @@ //! track mapping, braking analysis, etc.). Each session contains multiple "runs" //! that capture telemetry frames from the always-on ring buffer between start/stop //! markers. Frames are stored in-memory and can be retrieved for analysis. +//! +//! Persistence: session metadata is written to `data/sessions/.json` +//! (compact, without frames). Frame data is written separately to +//! `data/sessions/.frames.json` off-thread. On startup, only metadata is +//! loaded; frames are lazy-loaded on demand via [`SessionManager::ensure_frames_loaded`]. use serde::{Deserialize, Serialize}; use std::collections::HashMap; +use std::path::{Path, PathBuf}; use std::sync::Arc; use crate::recording::{RecordingManager, UseCase}; @@ -78,6 +84,8 @@ pub struct UseCaseSession { #[serde(default)] pub guidance: String, pub recording_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub car_name: Option, } impl UseCaseSession { @@ -93,25 +101,199 @@ impl UseCaseSession { analysis: self.analysis.clone(), guidance: self.guidance.clone(), recording_id: self.recording_id.clone(), + car_name: self.car_name.clone(), } } } +/// Serialized form of run frames for the `.frames.json` file. +/// One entry per run, in the same order as `UseCaseSession.runs`. +#[derive(Serialize, Deserialize)] +struct SessionFramesFile { + runs: Vec>, +} + pub struct SessionManager { sessions: HashMap, recording_manager: Arc, + data_dir: PathBuf, } impl SessionManager { pub fn new( recording_manager: Arc, + data_dir: PathBuf, ) -> Self { + let sessions_dir = data_dir.join("sessions"); + let _ = std::fs::create_dir_all(&sessions_dir); Self { sessions: HashMap::new(), recording_manager, + data_dir, } } + fn sessions_dir(&self) -> PathBuf { + self.data_dir.join("sessions") + } + + fn session_file_path(&self, id: &str) -> PathBuf { + self.sessions_dir().join(format!("{}.json", id)) + } + + fn frames_file_path(&self, id: &str) -> PathBuf { + self.sessions_dir().join(format!("{}.frames.json", id)) + } + + /// Atomically write bytes to a path via temp file + rename (same directory). + fn atomic_write(path: &Path, bytes: &[u8]) { + let tmp = path.with_extension("json.tmp"); + if let Err(e) = std::fs::write(&tmp, bytes) { + eprintln!("[WARN] Failed to write temp file {}: {}", tmp.display(), e); + return; + } + if let Err(e) = std::fs::rename(&tmp, path) { + eprintln!("[WARN] Failed to rename temp file {}: {}", tmp.display(), e); + let _ = std::fs::remove_file(&tmp); + } + } + + /// Persist session metadata (compact JSON, excludes frames) synchronously, + /// then write frame data off-thread to a separate file. + fn persist_session(&self, id: &str) { + let Some(session) = self.sessions.get(id) else { + return; + }; + let id = id.to_string(); + + // Metadata is small (no frames) — write synchronously with atomic rename. + let meta_path = self.session_file_path(&id); + match serde_json::to_vec(session) { + Ok(bytes) => Self::atomic_write(&meta_path, &bytes), + Err(e) => eprintln!("[WARN] Failed to serialize session metadata {}: {}", id, e), + } + + // Frames are large — clone under lock (memcpy) then serialize + write off-thread. + let runs_frames: Vec> = + session.runs.iter().map(|r| r.frames.clone()).collect(); + let frames_path = self.frames_file_path(&id); + std::thread::spawn(move || { + let frames_file = SessionFramesFile { runs: runs_frames }; + match serde_json::to_vec(&frames_file) { + Ok(bytes) => Self::atomic_write(&frames_path, &bytes), + Err(e) => eprintln!("[WARN] Failed to serialize frames for session {}: {}", id, e), + } + }); + } + + fn delete_session_file(&self, id: &str) { + for path in [self.session_file_path(id), self.frames_file_path(id)] { + if path.exists() { + if let Err(e) = std::fs::remove_file(&path) { + eprintln!("[WARN] Failed to delete file {}: {}", path.display(), e); + } + } + } + } + + /// Load all persisted session metadata from disk into memory. + /// + /// Only metadata files (`.json`) are loaded — not frame data + /// (`.frames.json`), which is lazy-loaded on demand. Any run found + /// in the `Recording` state (left over from an interrupted session) is + /// normalized to `Stopped` with `data_truncated = true`, since the ring + /// buffer it referenced no longer exists. + pub fn load_all(&mut self) { + let entries = match std::fs::read_dir(self.sessions_dir()) { + Ok(e) => e, + Err(_) => return, + }; + for entry in entries.flatten() { + let path = entry.path(); + let name = match path.file_name().and_then(|n| n.to_str()) { + Some(n) => n, + None => continue, + }; + // Skip frame files — loaded lazily. + if name.ends_with(".frames.json") { + continue; + } + if path.extension().and_then(|e| e.to_str()) != Some("json") { + continue; + } + match std::fs::read_to_string(&path) { + Ok(json) => match serde_json::from_str::(&json) { + Ok(mut session) => { + for run in &mut session.runs { + if run.status == RunStatus::Recording { + run.status = RunStatus::Stopped; + run.data_truncated = true; + } + } + self.sessions.insert(session.id.clone(), session); + } + Err(e) => { + eprintln!( + "[WARN] Failed to parse session file {}: {}", + path.display(), + e + ); + } + }, + Err(e) => { + eprintln!( + "[WARN] Failed to read session file {}: {}", + path.display(), + e + ); + } + } + } + } + + /// Lazy-load frame data for a session from its `.frames.json` file. + /// Called before operations that need frames (analysis, etc.). + /// No-op if frames are already populated or the file doesn't exist. + pub fn ensure_frames_loaded(&mut self, id: &str) { + let needs_load = self + .sessions + .get(id) + .map(|s| s.runs.iter().any(|r| r.frame_count > 0 && r.frames.is_empty())) + .unwrap_or(false); + if !needs_load { + return; + } + + let frames_path = self.frames_file_path(id); + let frames_file = match std::fs::read(&frames_path) + .ok() + .and_then(|bytes| serde_json::from_slice::(&bytes).ok()) + { + Some(f) => f, + None => { + eprintln!( + "[WARN] Failed to load frames for session {} from {}", + id, + frames_path.display() + ); + return; + } + }; + + if let Some(session) = self.sessions.get_mut(id) { + for (run, frames) in session.runs.iter_mut().zip(frames_file.runs) { + if run.frames.is_empty() { + run.frames = frames; + } + } + } + } + + /// Read-only iterator over all sessions (for characteristics aggregation). + pub fn iter_sessions(&self) -> impl Iterator { + self.sessions.values() + } + pub fn create_session( &mut self, use_case: UseCase, @@ -132,8 +314,10 @@ impl SessionManager { analysis: None, guidance, recording_id, + car_name: None, }; - self.sessions.insert(id, session.clone()); + self.sessions.insert(id.clone(), session.clone()); + self.persist_session(&id); Ok(session) } @@ -190,6 +374,7 @@ impl SessionManager { }; let summary = run.summary_clone(); session.runs.push(run); + self.persist_session(session_id); Ok(summary) } @@ -215,7 +400,14 @@ impl SessionManager { Self::finalize_run(&self.recording_manager, run); - Ok(run.summary_clone()) + let summary = run.summary_clone(); + let car_name = extract_car_name(&session.runs); + if session.car_name.is_none() { + session.car_name = car_name; + } + + self.persist_session(session_id); + Ok(summary) } fn finalize_run(recording_manager: &Arc, run: &mut SessionRun) { @@ -269,10 +461,18 @@ impl SessionManager { } } + // Cache car name from finalized frames. + let car_name = extract_car_name(&session.runs); + if session.car_name.is_none() { + session.car_name = car_name; + } + session.status = SessionStatus::Completed; session.completed_at = Some(chrono::Utc::now().to_rfc3339()); - Ok(session.summary_clone()) + let summary = session.summary_clone(); + self.persist_session(id); + Ok(summary) } pub fn delete_run(&mut self, session_id: &str, run_id: &str) -> Result<(), String> { @@ -297,6 +497,7 @@ impl SessionManager { session.analysis = None; } + self.persist_session(session_id); Ok(()) } @@ -314,7 +515,9 @@ impl SessionManager { session.completed_at = None; session.analysis = None; - Ok(session.summary_clone()) + let summary = session.summary_clone(); + self.persist_session(session_id); + Ok(summary) } pub fn set_analysis( @@ -327,15 +530,20 @@ impl SessionManager { .get_mut(session_id) .ok_or_else(|| format!("Session '{}' not found", session_id))?; session.analysis = Some(result); + self.persist_session(session_id); Ok(()) } /// Collect frames and labels from all runs for analysis. /// - /// Clones frames to allow the caller to release the SessionManager lock - /// before running CPU-bound analysis. Frames must remain in runs for - /// potential re-analysis via `POST /api/sessions/{id}/analyze`. - pub fn collect_run_frames(&self, session_id: &str) -> Option<(UseCase, Vec>, Vec)> { + /// Lazy-loads frame data from disk if needed, then clones frames to allow + /// the caller to release the SessionManager lock before running CPU-bound + /// analysis. Frames must remain in runs for potential re-analysis. + pub fn collect_run_frames( + &mut self, + session_id: &str, + ) -> Option<(UseCase, Vec>, Vec)> { + self.ensure_frames_loaded(session_id); let session = self.sessions.get(session_id)?; let mut frames_by_run = Vec::new(); let mut run_labels = Vec::new(); @@ -350,10 +558,25 @@ impl SessionManager { self.sessions .remove(id) .ok_or_else(|| format!("Session '{}' not found", id))?; + self.delete_session_file(id); Ok(()) } } +/// Extract the car name from the first frame (across all runs) that has one. +fn extract_car_name(runs: &[SessionRun]) -> Option { + for run in runs { + for frame in &run.frames { + if let Some(ref name) = frame.car_name { + if !name.is_empty() { + return Some(name.clone()); + } + } + } + } + None +} + /// Merge frames from multiple feeds into a single sorted, deduplicated sequence. /// /// Frames are sorted by `(lap_count, lap_time_ms)`. Near-duplicate frames diff --git a/src/telemetry_frame.rs b/src/telemetry_frame.rs index d3c7d2c..a391193 100644 --- a/src/telemetry_frame.rs +++ b/src/telemetry_frame.rs @@ -3,10 +3,10 @@ //! Both the KSUDP (port 9996) and Telemetry Tool (port 10101) parsers produce //! `TelemetryFrame` instances from their raw binary data. -use serde::Serialize; +use serde::{Deserialize, Serialize}; /// Identifies which feed source produced this frame. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub enum FeedSource { Ksudp, TelemetryTool, @@ -20,7 +20,7 @@ pub const NUM_WHEELS: usize = 4; /// /// Fields present in both feeds are stored directly; fields unique to one feed /// are wrapped in `Option`. -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct TelemetryFrame { // ── Speed ────────────────────────────────────────────────────── pub speed_kmh: f32,