From 9fae8ba9d86df8a0a00a07489b4d6430759b4773 Mon Sep 17 00:00:00 2001 From: "Jan-Peter v. Hunnius" Date: Sat, 6 Jun 2026 09:38:42 +0200 Subject: [PATCH] feat: add run deletion, session reopen, and feed merging (v0.6.0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - DELETE /api/sessions/{id}/runs/{run_id} removes stopped runs - POST /api/sessions/{id}/reopen transitions Completed → Active - Optional merge_feeds field on POST /api/sessions/{id}/runs combines frames from multiple feeds with priority-based deduplication - In-place merge_frames() avoids 2× peak memory - All new fields use #[serde(default)] for backward compatibility - Updated README.md with new endpoints and feed merging docs --- Cargo.lock | 2 +- Cargo.toml | 2 +- README.md | 48 ++++++++++++++++- src/api.rs | 30 ++++++++++- src/session.rs | 137 +++++++++++++++++++++++++++++++++++++++++++++++-- 5 files changed, 208 insertions(+), 11 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f832b4a..d317950 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -786,7 +786,7 @@ checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" [[package]] name = "rusty-telemetry" -version = "0.5.0" +version = "0.6.0" dependencies = [ "axum", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 62c801d..d0e3b1a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rusty-telemetry" -version = "0.5.0" +version = "0.6.0" edition = "2021" description = "Sim racing telemetry receiver and analyzer" diff --git a/README.md b/README.md index 984c6a8..3bb5850 100644 --- a/README.md +++ b/README.md @@ -100,7 +100,7 @@ The terminal UI shows one panel per active feed: ``` ┌──────────────────────────────────────────────────────────┐ -│ rusty-telemetry v0.5.0 -- Telemetry Listener │ +│ rusty-telemetry v0.6.0 -- Telemetry Listener │ ├──────────────────────────────────────────────────────────┤ │ ╔ * Receiving ╗ │ │ ║ 20.0 Hz 1200 packets 15.2 KB/s last: 0.05s ║ @@ -151,9 +151,32 @@ Use-case sessions allow targeted data capture and analysis: 2. **Start runs** — each run captures frames from the ring buffer between start and stop 3. **Complete the session** — auto-analyzes captured data and stores results 4. **Retrieve analysis** — shift point RPM ranges or track boundary coordinates +5. **Re-open if needed** — re-open a completed session to add more runs, then complete again +6. **Delete runs** — remove unwanted runs from a session (must be stopped first) Sessions are in-memory and independent of persistent recordings. +#### Feed Merging + +When multiple feeds are available during a recording (e.g., both KSUDP on port 9996 and Telemetry Tool on port 10101), runs can merge frames from multiple sources for more accurate analysis data. + +To start a merged run, specify `merge_feeds` in the start run request: + +```json +{ + "feed_name": "ksudp", + "merge_feeds": ["telemetry_tool"] +} +``` + +The merge algorithm: +1. Extracts frames independently from each feed using ring buffer markers +2. Sorts all frames by `(lap_count, lap_time_ms)` +3. Deduplicates near-simultaneous frames (within 15ms) by keeping the higher-priority feed +4. Feed priority: Telemetry Tool > KSUDP > PCARS + +This results in higher effective data density (combining 60 Hz TT + 20 Hz KSUDP) while preferring the richer Telemetry Tool data where available. + ## REST API A built-in HTTP API server runs on port 8080 for recording management and live telemetry access. @@ -178,6 +201,8 @@ A built-in HTTP API server runs on port 8080 for recording management and live t | DELETE | `/api/sessions/{id}` | Delete session | | POST | `/api/sessions/{id}/runs` | Start a run | | POST | `/api/sessions/{id}/runs/{run_id}/stop` | Stop a run | +| DELETE | `/api/sessions/{id}/runs/{run_id}` | Delete a stopped run from session | +| POST | `/api/sessions/{id}/reopen` | Re-open a completed session | | POST | `/api/sessions/{id}/complete` | Complete session + auto-analyze | | GET | `/api/sessions/{id}/analysis` | Get analysis results | | POST | `/api/sessions/{id}/analyze` | Re-run analysis on completed session | @@ -236,10 +261,29 @@ All fields are optional. ```json { "feed_name": "ksudp", - "label": "Acceleration runs" + "label": "Acceleration runs", + "merge_feeds": ["telemetry_tool"] } ``` +`merge_feeds` is optional. When specified, frames from the listed feeds are merged with the primary feed's frames into a single deduplicated dataset. This is useful for combining the higher-frequency Telemetry Tool data (~60 Hz) with KSUDP data (~20 Hz) for more accurate analysis. + +### Delete Run + +Deletes a stopped run from a session. If the session was completed and had analysis results, the analysis is cleared (since the underlying data changed). Re-run analysis after deletion if needed. + +Returns `204 No Content` on success. Cannot delete a run that is currently recording — stop it first. + +### Reopen Session + +Re-opens a completed session, changing its status back to active. This clears the `completed_at` timestamp and any analysis results, allowing new runs to be added. Complete the session again when done to re-run analysis. + +```json +{} +``` + +Returns the updated session on success. + ### Use Cases Response ```json diff --git a/src/api.rs b/src/api.rs index 0eac999..98ca245 100644 --- a/src/api.rs +++ b/src/api.rs @@ -7,7 +7,7 @@ use axum::{ extract::{Path, State}, http::StatusCode, response::{IntoResponse, Response}, - routing::{get, post}, + routing::{delete, get, post}, }; use serde::{Deserialize, Serialize}; use std::sync::{Arc, Mutex, MutexGuard}; @@ -85,6 +85,8 @@ struct CreateSessionRequest { struct StartRunRequest { feed_name: String, label: Option, + #[serde(default)] + merge_feeds: Vec, } async fn create_session( @@ -132,7 +134,7 @@ async fn start_run( Json(req): Json, ) -> Response { let mut mgr = lock_sessions(&state); - match mgr.start_run(&id, &req.feed_name, req.label) { + match mgr.start_run(&id, &req.feed_name, req.label, req.merge_feeds) { Ok(run) => (StatusCode::CREATED, Json(run)).into_response(), Err(msg) => err(StatusCode::BAD_REQUEST, &msg), } @@ -149,6 +151,28 @@ async fn stop_run( } } +async fn delete_run( + State(state): State>, + Path((id, run_id)): Path<(String, String)>, +) -> Response { + let mut mgr = lock_sessions(&state); + match mgr.delete_run(&id, &run_id) { + Ok(()) => StatusCode::NO_CONTENT.into_response(), + Err(msg) => err(StatusCode::BAD_REQUEST, &msg), + } +} + +async fn reopen_session( + State(state): State>, + Path(id): Path, +) -> Response { + let mut mgr = lock_sessions(&state); + match mgr.reopen_session(&id) { + Ok(session) => ok_json(session), + Err(msg) => err(StatusCode::BAD_REQUEST, &msg), + } +} + async fn enrich_with_recording_meta( state: &Arc, result: &mut serde_json::Value, @@ -521,6 +545,8 @@ pub fn build_router(state: Arc) -> Router { .route("/api/sessions/{id}", get(get_session).delete(delete_session)) .route("/api/sessions/{id}/runs", post(start_run)) .route("/api/sessions/{id}/runs/{run_id}/stop", post(stop_run)) + .route("/api/sessions/{id}/runs/{run_id}", delete(delete_run)) + .route("/api/sessions/{id}/reopen", post(reopen_session)) .route("/api/sessions/{id}/complete", post(complete_session)) .route("/api/sessions/{id}/analysis", get(get_analysis)) .route("/api/sessions/{id}/analyze", post(run_analysis)) diff --git a/src/session.rs b/src/session.rs index 395b4fe..7bf6e87 100644 --- a/src/session.rs +++ b/src/session.rs @@ -31,9 +31,15 @@ pub struct SessionRun { pub id: String, pub label: String, pub feed_name: String, + #[serde(default)] + pub merge_feeds: Vec, pub status: RunStatus, pub start_marker: u64, + #[serde(default)] + pub merge_start_markers: HashMap, pub end_marker: Option, + #[serde(default)] + pub merge_end_markers: HashMap>, pub frame_count: usize, pub data_truncated: bool, #[serde(skip)] @@ -46,9 +52,12 @@ impl SessionRun { id: self.id.clone(), label: self.label.clone(), feed_name: self.feed_name.clone(), + merge_feeds: self.merge_feeds.clone(), status: self.status.clone(), start_marker: self.start_marker, + merge_start_markers: self.merge_start_markers.clone(), end_marker: self.end_marker, + merge_end_markers: self.merge_end_markers.clone(), frame_count: self.frame_count, data_truncated: self.data_truncated, frames: Vec::new(), @@ -145,6 +154,7 @@ impl SessionManager { session_id: &str, feed_name: &str, label: Option, + merge_feeds: Vec, ) -> Result { let session = self .sessions @@ -156,14 +166,22 @@ impl SessionManager { } let marker = self.recording_manager.ring_marker(feed_name); + let mut merge_start_markers = HashMap::new(); + for mf in &merge_feeds { + merge_start_markers.insert(mf.clone(), self.recording_manager.ring_marker(mf)); + } + let run_id = uuid::Uuid::new_v4().to_string(); let run = SessionRun { id: run_id, label: label.unwrap_or_else(|| format!("Run {}", session.runs.len() + 1)), feed_name: feed_name.to_string(), + merge_feeds, status: RunStatus::Recording, start_marker: marker, + merge_start_markers, end_marker: None, + merge_end_markers: HashMap::new(), frame_count: 0, data_truncated: false, frames: Vec::new(), @@ -200,17 +218,37 @@ impl SessionManager { fn finalize_run(recording_manager: &Arc, run: &mut SessionRun) { let end_marker = recording_manager.ring_marker(&run.feed_name); - let frames = recording_manager + let primary_frames = recording_manager .extract_since(&run.feed_name, run.start_marker); let expected = end_marker.saturating_sub(run.start_marker); - let data_truncated = (frames.len() as u64) < expected && expected > 0; + let primary_truncated = (primary_frames.len() as u64) < expected && expected > 0; + + let mut all_frames = primary_frames; + let mut any_truncated = primary_truncated; + + for feed in &run.merge_feeds { + if let Some(&start_m) = run.merge_start_markers.get(feed) { + let end_m = recording_manager.ring_marker(feed); + let frames = recording_manager.extract_since(feed, start_m); + let exp = end_m.saturating_sub(start_m); + if (frames.len() as u64) < exp && exp > 0 { + any_truncated = true; + } + run.merge_end_markers.insert(feed.clone(), Some(end_m)); + all_frames.extend(frames); + } + } + + if !run.merge_feeds.is_empty() { + all_frames = merge_frames(all_frames); + } run.end_marker = Some(end_marker); - run.frame_count = frames.len(); - run.frames = frames; + run.frame_count = all_frames.len(); + run.frames = all_frames; run.status = RunStatus::Stopped; - run.data_truncated = data_truncated; + run.data_truncated = any_truncated; } pub fn complete_session(&mut self, id: &str) -> Result { @@ -235,6 +273,48 @@ impl SessionManager { Ok(session.summary_clone()) } + pub fn delete_run(&mut self, session_id: &str, run_id: &str) -> Result<(), String> { + let session = self + .sessions + .get_mut(session_id) + .ok_or_else(|| format!("Session '{}' not found", session_id))?; + + let run_idx = session + .runs + .iter() + .position(|r| r.id == run_id) + .ok_or_else(|| format!("Run '{}' not found in session '{}'", run_id, session_id))?; + + if session.runs[run_idx].status == RunStatus::Recording { + return Err("Cannot delete a recording run. Stop it first.".to_string()); + } + + session.runs.remove(run_idx); + + if session.status == SessionStatus::Completed && session.analysis.is_some() { + session.analysis = None; + } + + Ok(()) + } + + pub fn reopen_session(&mut self, session_id: &str) -> Result { + let session = self + .sessions + .get_mut(session_id) + .ok_or_else(|| format!("Session '{}' not found", session_id))?; + + if session.status != SessionStatus::Completed { + return Err(format!("Session '{}' is not completed", session_id)); + } + + session.status = SessionStatus::Active; + session.completed_at = None; + session.analysis = None; + + Ok(session.summary_clone()) + } + pub fn set_analysis( &mut self, session_id: &str, @@ -271,3 +351,50 @@ impl SessionManager { Ok(()) } } + +/// Merge frames from multiple feeds into a single sorted, deduplicated sequence. +/// +/// Frames are sorted by `(lap_count, lap_time_ms)`. Near-duplicate frames +/// (within 15ms on the same lap) are resolved by keeping the frame from the +/// higher-priority feed: Telemetry Tool > KSUDP > PCARS. +fn merge_frames(mut frames: Vec) -> Vec { + if frames.len() <= 1 { + return frames; + } + + frames.sort_by(|a, b| { + a.lap_count + .cmp(&b.lap_count) + .then_with(|| a.lap_time_ms.cmp(&b.lap_time_ms)) + }); + + const TOLERANCE_MS: i32 = 15; + + let mut write = 0; + for read in 1..frames.len() { + if frames[write].lap_count == frames[read].lap_count + && (frames[read].lap_time_ms - frames[write].lap_time_ms).abs() <= TOLERANCE_MS + { + // Near-duplicate: keep higher-priority frame at write position + if feed_priority(frames[read].source) > feed_priority(frames[write].source) { + frames[write] = std::mem::replace(&mut frames[read], TelemetryFrame::default()); + } + } else { + write += 1; + frames.swap(write, read); + } + } + + frames.truncate(write + 1); + frames +} + +/// Feed priority for merge deduplication: TelemetryTool > KSUDP > PCARS. +fn feed_priority(source: crate::telemetry_frame::FeedSource) -> u8 { + use crate::telemetry_frame::FeedSource; + match source { + FeedSource::TelemetryTool => 3, + FeedSource::Ksudp => 2, + FeedSource::Pcars => 1, + } +}