jan/rusty-telemetry
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add use-case session management, analysis algorithms, and REST API (v0.5.0)

Browse source 
- Add session layer with UseCaseSession, SessionRun, SessionManager
  for use-case driven telemetry analysis (shift points, track mapping)
- Add analysis module with shift point detection (gear RPM ranges,
  upshift events, recommended shift RPM) and track map reconstruction
  (spline-aligned interpolation for center line from boundary runs)
- Add 9 REST API endpoints for session CRUD, run management, and
  analysis triggers under /api/sessions/*
- Add RecordingManager with always-on ring buffers (5 min, 18k frames)
  replacing raw binary file dumps
- Wire SessionManager in main.rs with Arc<Mutex> shared state
- Add TelemetryFrame Serialize derives for JSON API output
- Fix stop_recording race: hold active lock across flush I/O
- Fix flush: separate ring buffer lock from JSON serialization
- Fix flush failure propagation: errors prevent Stopped status
- Add data_truncated flag on SessionRun for ring buffer wrap detection
- Deduplicate gear_display into shared free function
- Deduplicate session_display into closure
- Mutex poisoning recovery on all session API handlers
This commit is contained in:
Jan-Peter von Hunnius 2026-06-05 10:40:17 +02:00
parent 56d3169033
commit 47a0896134
10 changed files with 2333 additions and 91 deletions
Download patch file Download diff file Expand all files Collapse all files

290
src/analysis.rs Normal file
View file

@ -0,0 +1,290 @@
//! Telemetry analysis algorithms for sim racing data.
//!
//! Provides shift point analysis, track map reconstruction, and a dispatch
//! function that routes a [`UseCase`] to the appropriate analysis pipeline.
use serde::Serialize;
use std::collections::HashMap;
use crate::recording::UseCase;
use crate::telemetry_frame::{gear_display, TelemetryFrame};
// ---------------------------------------------------------------------------
// Shift Point Analysis
// ---------------------------------------------------------------------------
#[derive(Debug, Clone, Serialize)]
pub struct GearRpmRange {
pub gear: i32,
pub gear_display: String,
pub min_rpm: f32,
pub max_rpm: f32,
pub avg_rpm: f32,
pub sample_count: usize,
}
#[derive(Debug, Clone, Serialize)]
pub struct ShiftEvent {
pub from_gear: i32,
pub to_gear: i32,
pub rpm_before_shift: f32,
pub rpm_after_shift: f32,
pub rpm_drop: f32,
pub speed_kmh: f32,
}
#[derive(Debug, Clone, Serialize)]
pub struct ShiftPointAnalysis {
pub gear_ranges: Vec<GearRpmRange>,
pub shift_events: Vec<ShiftEvent>,
pub recommended_shift_rpm: Option<f32>,
pub total_frames_analyzed: usize,
}
pub fn analyze_shift_points(frames: &[TelemetryFrame]) -> ShiftPointAnalysis {
let valid: Vec<&TelemetryFrame> = frames
.iter()
.filter(|f| f.gear >= 2 && f.engine_rpm > 0.0)
.collect();
let mut buckets: HashMap<i32, Vec<f32>> = HashMap::new();
for f in &valid {
buckets.entry(f.gear).or_default().push(f.engine_rpm);
}
let mut gear_ranges: Vec<GearRpmRange> = buckets
.into_iter()
.map(|(gear, rpms)| {
let min_rpm = rpms.iter().cloned().fold(f32::INFINITY, f32::min);
let max_rpm = rpms.iter().cloned().fold(f32::NEG_INFINITY, f32::max);
let avg_rpm = rpms.iter().sum::<f32>() / rpms.len() as f32;
GearRpmRange {
gear,
gear_display: gear_display(gear),
min_rpm,
max_rpm,
avg_rpm,
sample_count: rpms.len(),
}
})
.collect();
gear_ranges.sort_by_key(|g| g.gear);
let valid_indices: Vec<usize> = frames
.iter()
.enumerate()
.filter(|(_, f)| f.gear >= 2 && f.engine_rpm > 0.0)
.map(|(i, _)| i)
.collect();
let mut shift_events: Vec<ShiftEvent> = Vec::new();
for window in valid_indices.windows(2) {
let prev = &frames[window[0]];
let curr = &frames[window[1]];
if prev.gear + 1 == curr.gear {
shift_events.push(ShiftEvent {
from_gear: prev.gear,
to_gear: curr.gear,
rpm_before_shift: prev.engine_rpm,
rpm_after_shift: curr.engine_rpm,
rpm_drop: prev.engine_rpm - curr.engine_rpm,
speed_kmh: prev.speed_kmh,
});
}
}
let recommended_shift_rpm = {
let upshift_rpms: Vec<f32> = shift_events
.iter()
.filter(|e| e.from_gear >= 2 && e.to_gear >= 2)
.map(|e| e.rpm_before_shift)
.collect();
if upshift_rpms.is_empty() {
None
} else {
Some(upshift_rpms.iter().sum::<f32>() / upshift_rpms.len() as f32)
}
};
ShiftPointAnalysis {
gear_ranges,
shift_events,
recommended_shift_rpm,
total_frames_analyzed: valid.len(),
}
}
// ---------------------------------------------------------------------------
// Track Map Analysis
// ---------------------------------------------------------------------------
#[derive(Debug, Clone, Serialize)]
pub struct TrackPoint {
pub x: f32,
pub y: f32,
pub z: f32,
pub spline: f32,
}
#[derive(Debug, Clone, Serialize)]
pub struct TrackBoundary {
pub label: String,
pub point_count: usize,
pub points: Vec<TrackPoint>,
}
#[derive(Debug, Clone, Serialize)]
pub struct TrackMapAnalysis {
pub boundaries: Vec<TrackBoundary>,
pub center_line: Option<TrackBoundary>,
pub total_points: usize,
}
pub fn analyze_track_map(
runs: &[Vec<TelemetryFrame>],
labels: &[String],
) -> TrackMapAnalysis {
let mut boundaries: Vec<TrackBoundary> = Vec::new();
for (run, label) in runs.iter().zip(labels.iter()) {
let mut points: Vec<TrackPoint> = run
.iter()
.filter(|f| f.spline_position != 0.0)
.map(|f| TrackPoint {
x: f.world_x,
y: f.world_y,
z: f.world_z,
spline: f.spline_position,
})
.collect();
points.sort_by(|a, b| a.spline.partial_cmp(&b.spline).unwrap_or(std::cmp::Ordering::Equal));
let point_count = points.len();
boundaries.push(TrackBoundary {
label: label.clone(),
point_count,
points,
});
}
let center_line = compute_center_line(&boundaries);
let total_points: usize = boundaries.iter().map(|b| b.point_count).sum::<usize>()
+ center_line.as_ref().map_or(0, |c| c.point_count);
TrackMapAnalysis {
boundaries,
center_line,
total_points,
}
}
fn compute_center_line(boundaries: &[TrackBoundary]) -> Option<TrackBoundary> {
if boundaries.len() != 2 {
return None;
}
let left_idx = boundaries
.iter()
.position(|b| b.label == "left")?;
let right_idx = boundaries
.iter()
.position(|b| b.label == "right")?;
if left_idx == right_idx {
return None;
}
let left = &boundaries[left_idx];
let right = &boundaries[right_idx];
if left.points.is_empty() || right.points.is_empty() {
return None;
}
let num_samples = left.point_count.max(right.point_count).min(2000);
let center_points: Vec<TrackPoint> = (0..num_samples)
.filter_map(|i| {
let spline = i as f32 / (num_samples - 1).max(1) as f32;
let lp = interpolate_boundary(left, spline)?;
let rp = interpolate_boundary(right, spline)?;
Some(TrackPoint {
x: (lp.x + rp.x) / 2.0,
y: (lp.y + rp.y) / 2.0,
z: (lp.z + rp.z) / 2.0,
spline,
})
})
.collect();
if center_points.is_empty() {
return None;
}
Some(TrackBoundary {
label: "center_line".to_owned(),
point_count: center_points.len(),
points: center_points,
})
}
/// Linearly interpolate a TrackPoint at the given spline position from a boundary.
fn interpolate_boundary(boundary: &TrackBoundary, target_spline: f32) -> Option<TrackPoint> {
let points = &boundary.points;
if points.is_empty() {
return None;
}
if target_spline <= points.first()?.spline {
return Some(points.first()?.clone());
}
if target_spline >= points.last()?.spline {
return Some(points.last()?.clone());
}
let idx = points.partition_point(|p| p.spline < target_spline);
if idx == 0 || idx >= points.len() {
return points.first().cloned();
}
let a = &points[idx - 1];
let b = &points[idx];
let t = if (b.spline - a.spline).abs() < f32::EPSILON {
0.0
} else {
(target_spline - a.spline) / (b.spline - a.spline)
};
Some(TrackPoint {
x: a.x + t * (b.x - a.x),
y: a.y + t * (b.y - a.y),
z: a.z + t * (b.z - a.z),
spline: target_spline,
})
}
// ---------------------------------------------------------------------------
// Dispatch
// ---------------------------------------------------------------------------
pub fn analyze(
use_case: &UseCase,
frames_by_run: &[Vec<TelemetryFrame>],
run_labels: &[String],
) -> serde_json::Value {
match use_case {
UseCase::ShiftPoints => {
if let Some(first_run) = frames_by_run.first() {
serde_json::to_value(analyze_shift_points(first_run))
.unwrap_or_else(|_| serde_json::json!({"error": "serialization failed"}))
} else {
serde_json::json!({"error": "no frames provided"})
}
}
UseCase::TrackMap => serde_json::to_value(analyze_track_map(frames_by_run, run_labels))
.unwrap_or_else(|_| serde_json::json!({"error": "serialization failed"})),
_ => serde_json::json!({
"status": "not_implemented",
"use_case": format!("{:?}", use_case),
}),
}
}

526
src/api.rs Normal file
View file

@ -0,0 +1,526 @@
//! REST API server for recording management and live telemetry access.
//!
//! Runs as a parallel tokio task alongside the UDP listeners.
use axum::{
Json, Router,
extract::{Path, State},
http::StatusCode,
response::{IntoResponse, Response},
routing::{get, post},
};
use serde::{Deserialize, Serialize};
use std::sync::{Arc, Mutex, MutexGuard};
use tower_http::cors::CorsLayer;
use crate::analysis;
use crate::feed_state::FeedState;
use crate::recording::{Game, RecordingManager, RecordingStatus, StartRecordingRequest, UseCase};
use crate::session::SessionManager;
use crate::ui::FeedDisplayInfo;
pub struct AppState {
pub recording_manager: Arc<RecordingManager>,
pub feed_displays: Vec<(FeedDisplayInfo, Arc<Mutex<FeedState>>)>,
pub session_manager: Arc<std::sync::Mutex<SessionManager>>,
}
#[derive(Serialize)]
struct ApiError {
error: String,
}
fn ok_json<T: Serialize>(data: T) -> Response {
(StatusCode::OK, Json(data)).into_response()
}
fn err(status: StatusCode, msg: &str) -> Response {
(
status,
Json(ApiError {
error: msg.to_string(),
}),
)
.into_response()
}
fn lock_sessions(state: &Arc<AppState>) -> MutexGuard<'_, SessionManager> {
state.session_manager.lock().unwrap_or_else(|e| {
eprintln!("[WARN] SessionManager mutex poisoned, recovering: {}", e);
e.into_inner()
})
}
#[derive(Serialize)]
struct LiveFeedSnapshot {
label: String,
port: u16,
health: String,
frequency_hz: f64,
total_packets: u64,
data_rate_kb_s: f64,
speed_kmh: Option<f32>,
gear: Option<String>,
engine_rpm: Option<f32>,
throttle: Option<f32>,
brake: Option<f32>,
lap_time: Option<String>,
spline_position: Option<f32>,
}
#[derive(Serialize)]
struct LiveTelemetryResponse {
feeds: Vec<LiveFeedSnapshot>,
active_recording: Option<String>,
}
#[derive(Deserialize)]
struct CreateSessionRequest {
use_case: UseCase,
label: Option<String>,
}
#[derive(Deserialize)]
struct StartRunRequest {
feed_name: String,
label: Option<String>,
}
async fn create_session(
State(state): State<Arc<AppState>>,
Json(req): Json<CreateSessionRequest>,
) -> Response {
let mut mgr = lock_sessions(&state);
match mgr.create_session(req.use_case, req.label) {
Ok(session) => (StatusCode::CREATED, Json(session)).into_response(),
Err(msg) => err(StatusCode::BAD_REQUEST, &msg),
}
}
async fn list_sessions(State(state): State<Arc<AppState>>) -> Response {
let mut mgr = lock_sessions(&state);
ok_json(mgr.list_sessions())
}
async fn get_session(
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let mgr = lock_sessions(&state);
match mgr.get_session(&id) {
Some(session) => ok_json(session),
None => err(StatusCode::NOT_FOUND, "Session not found"),
}
}
async fn delete_session(
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let mut mgr = lock_sessions(&state);
match mgr.delete_session(&id) {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(msg) => err(StatusCode::NOT_FOUND, &msg),
}
}
async fn start_run(
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
Json(req): Json<StartRunRequest>,
) -> Response {
let mut mgr = lock_sessions(&state);
match mgr.start_run(&id, &req.feed_name, req.label) {
Ok(run) => (StatusCode::CREATED, Json(run)).into_response(),
Err(msg) => err(StatusCode::BAD_REQUEST, &msg),
}
}
async fn stop_run(
State(state): State<Arc<AppState>>,
Path((id, run_id)): Path<(String, String)>,
) -> Response {
let mut mgr = lock_sessions(&state);
match mgr.stop_run(&id, &run_id) {
Ok(run) => ok_json(run),
Err(msg) => err(StatusCode::BAD_REQUEST, &msg),
}
}
async fn complete_session(
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let (use_case, frames_by_run, run_labels) = {
let mut mgr = lock_sessions(&state);
match mgr.complete_session(&id) {
Ok(_session) => {}
Err(msg) => return err(StatusCode::BAD_REQUEST, &msg),
}
match mgr.collect_run_frames(&id) {
Some((uc, frames, labels)) => (uc, frames, labels),
None => return err(StatusCode::NOT_FOUND, "Session not found after completion"),
}
};
let result = analysis::analyze(&use_case, &frames_by_run, &run_labels);
{
let mut mgr = lock_sessions(&state);
let _ = mgr.set_analysis(&id, result);
}
let mgr = lock_sessions(&state);
match mgr.get_session(&id) {
Some(session) => ok_json(session),
None => err(StatusCode::INTERNAL_SERVER_ERROR, "Session lost after analysis"),
}
}
async fn get_analysis(
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let mgr = lock_sessions(&state);
match mgr.get_session(&id) {
Some(session) => {
match session.analysis {
Some(ref analysis) => ok_json(analysis),
None => err(StatusCode::NOT_FOUND, "Analysis not available yet. Complete the session first."),
}
}
None => err(StatusCode::NOT_FOUND, "Session not found"),
}
}
async fn run_analysis(
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let (use_case, frames_by_run, run_labels) = {
let mgr = lock_sessions(&state);
match mgr.get_session(&id) {
Some(session) => {
if session.status != crate::session::SessionStatus::Completed {
return err(StatusCode::BAD_REQUEST, "Session must be completed before analysis");
}
}
None => return err(StatusCode::NOT_FOUND, "Session not found"),
}
match mgr.collect_run_frames(&id) {
Some(data) => data,
None => return err(StatusCode::NOT_FOUND, "Session not found"),
}
};
let result = analysis::analyze(&use_case, &frames_by_run, &run_labels);
{
let mut mgr = lock_sessions(&state);
let _ = mgr.set_analysis(&id, result.clone());
}
ok_json(result)
}
async fn list_recordings(State(state): State<Arc<AppState>>) -> Response {
match state.recording_manager.list_recordings().await {
Ok(recordings) => ok_json(recordings),
Err(e) => err(
StatusCode::INTERNAL_SERVER_ERROR,
&format!("Failed to list recordings: {}", e),
),
}
}
async fn get_recording(
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
match state.recording_manager.get_recording(&id).await {
Ok(manifest) => ok_json(manifest),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
err(StatusCode::NOT_FOUND, "Recording not found")
}
Err(e) if e.kind() == std::io::ErrorKind::InvalidInput => {
err(StatusCode::BAD_REQUEST, &e.to_string())
}
Err(e) => err(
StatusCode::INTERNAL_SERVER_ERROR,
&format!("Failed to get recording: {}", e),
),
}
}
async fn start_recording(
State(state): State<Arc<AppState>>,
Json(req): Json<StartRecordingRequest>,
) -> Response {
match state.recording_manager.start_recording(req).await {
Ok(manifest) => (StatusCode::CREATED, Json(manifest)).into_response(),
Err(e) => err(
StatusCode::INTERNAL_SERVER_ERROR,
&format!("Failed to start recording: {}", e),
),
}
}
async fn stop_recording(State(state): State<Arc<AppState>>) -> Response {
match state.recording_manager.stop_recording().await {
Ok(Some(manifest)) => ok_json(manifest),
Ok(None) => err(StatusCode::BAD_REQUEST, "No active recording to stop"),
Err(e) => err(
StatusCode::INTERNAL_SERVER_ERROR,
&format!("Failed to stop recording: {}", e),
),
}
}
async fn stop_specific_recording(
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let manifest = match state.recording_manager.get_recording(&id).await {
Ok(m) => m,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
return err(StatusCode::NOT_FOUND, "Recording not found")
}
Err(e) if e.kind() == std::io::ErrorKind::InvalidInput => {
return err(StatusCode::BAD_REQUEST, &e.to_string())
}
Err(e) => {
return err(
StatusCode::INTERNAL_SERVER_ERROR,
&format!("Failed to get recording: {}", e),
)
}
};
if manifest.status != RecordingStatus::Recording {
return err(StatusCode::CONFLICT, "Recording is not active");
}
match state
.recording_manager
.stop_recording_internal(&id)
.await
{
Ok(stopped) => {
state
.recording_manager
.clear_active_if_matches(&id)
.await;
ok_json(stopped)
}
Err(e) => err(
StatusCode::INTERNAL_SERVER_ERROR,
&format!("Failed to stop recording: {}", e),
),
}
}
async fn delete_recording(
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
match state.recording_manager.delete_recording(&id).await {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(e) if e.kind() == std::io::ErrorKind::InvalidInput => {
err(StatusCode::CONFLICT, &e.to_string())
}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
err(StatusCode::NOT_FOUND, "Recording not found")
}
Err(e) => err(
StatusCode::INTERNAL_SERVER_ERROR,
&format!("Failed to delete recording: {}", e),
),
}
}
async fn get_feed_data(
State(state): State<Arc<AppState>>,
Path((id, feed_name)): Path<(String, String)>,
) -> Response {
let active_id = state.recording_manager.active_recording_id().await;
if active_id.as_deref() == Some(id.as_str()) {
match state.recording_manager.read_frames(&feed_name) {
Some(frames) => return ok_json(frames),
None => {
return err(
StatusCode::NOT_FOUND,
&format!("Feed '{}' not found in active recording", feed_name),
)
}
}
}
let manifest = match state.recording_manager.get_recording(&id).await {
Ok(m) => m,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
return err(StatusCode::NOT_FOUND, "Recording not found")
}
Err(e) if e.kind() == std::io::ErrorKind::InvalidInput => {
return err(StatusCode::BAD_REQUEST, &e.to_string())
}
Err(e) => {
return err(
StatusCode::INTERNAL_SERVER_ERROR,
&format!("Failed to get recording: {}", e),
)
}
};
let feed_stats = match manifest.feeds.iter().find(|f| f.name == feed_name) {
Some(f) => f,
None => {
return err(
StatusCode::NOT_FOUND,
&format!("Feed '{}' not found in recording", feed_name),
)
}
};
let file_path = manifest.dir_path(state.recording_manager.data_dir()).join(&feed_stats.file);
if !file_path.exists() {
return err(StatusCode::NOT_FOUND, "Feed data file not found on disk (recording may still be in progress)");
}
match tokio::fs::read_to_string(&file_path).await {
Ok(content) => {
let frames: Vec<serde_json::Value> = content
.lines()
.filter(|l| !l.is_empty())
.filter_map(|l| serde_json::from_str(l).ok())
.collect();
ok_json(frames)
}
Err(e) => err(
StatusCode::INTERNAL_SERVER_ERROR,
&format!("Failed to read feed data: {}", e),
),
}
}
async fn get_live(State(state): State<Arc<AppState>>) -> Response {
let mut feeds = Vec::new();
for (info, state_arc) in &state.feed_displays {
let snapshot = state_arc.lock().ok().map(|mut s| {
s.tick();
s.clone()
});
let snap = match snapshot {
Some(s) => s,
None => continue,
};
let (speed_kmh, gear, engine_rpm, throttle, brake, lap_time, spline_position) =
if let Some(ref frame) = snap.last_frame {
(
Some(frame.speed_kmh),
Some(frame.gear_display()),
Some(frame.engine_rpm),
Some(frame.throttle),
Some(frame.brake),
Some(
crate::telemetry_frame::TelemetryFrame::format_lap_time(frame.lap_time_ms),
),
Some(frame.spline_position),
)
} else {
(None, None, None, None, None, None, None)
};
feeds.push(LiveFeedSnapshot {
label: info.label.to_string(),
port: info.port,
health: format!("{:?}", snap.health).to_lowercase(),
frequency_hz: snap.frequency_hz,
total_packets: snap.total_packets,
data_rate_kb_s: snap.data_rate_kb_s,
speed_kmh,
gear,
engine_rpm,
throttle,
brake,
lap_time,
spline_position,
});
}
let active_recording = state.recording_manager.active_recording_id().await;
ok_json(LiveTelemetryResponse {
feeds,
active_recording,
})
}
async fn list_use_cases() -> Response {
ok_json(UseCase::all())
}
async fn list_games() -> Response {
ok_json(Game::valid_choices())
}
pub fn build_router(state: Arc<AppState>) -> Router {
Router::new()
.route("/api/sessions", get(list_sessions).post(create_session))
.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}/complete", post(complete_session))
.route("/api/sessions/{id}/analysis", get(get_analysis))
.route("/api/sessions/{id}/analyze", post(run_analysis))
.route("/api/recordings", get(list_recordings).post(start_recording))
.route("/api/recordings/stop", post(stop_recording))
.route(
"/api/recordings/{id}",
get(get_recording).delete(delete_recording),
)
.route("/api/recordings/{id}/stop", post(stop_specific_recording))
.route(
"/api/recordings/{id}/data/{feed_name}",
get(get_feed_data),
)
.route(
"/api/recordings/{id}/frames/{feed_name}",
get(get_feed_data),
)
.route("/api/live", get(get_live))
.route("/api/use-cases", get(list_use_cases))
.route("/api/games", get(list_games))
.layer(CorsLayer::permissive())
.with_state(state)
}
pub async fn run_server(state: Arc<AppState>, addr: String, mut shutdown_rx: tokio::sync::watch::Receiver<bool>) {
let router = build_router(state);
let listener = match tokio::net::TcpListener::bind(&addr).await {
Ok(l) => l,
Err(e) => {
eprintln!("[ERROR] Failed to bind API server on {}: {}", addr, e);
return;
}
};
eprintln!("[INFO] REST API server listening on {}", addr);
let shutdown_signal = async move {
while shutdown_rx.changed().await.is_ok() {
if *shutdown_rx.borrow() {
break;
}
}
};
if let Err(e) = axum::serve(listener, router)
.with_graceful_shutdown(shutdown_signal)
.await
{
eprintln!("[ERROR] API server error: {}", e);
}
eprintln!("[INFO] REST API server shut down.");
}

166
src/main.rs
View file

@ -1,10 +1,14 @@
mod api;
mod binary_utils;
mod feed_state;
mod ui;
mod telemetry_frame;
mod parser_ksudp;
mod parser_telemetry_tool;
mod parser_pcars;
mod parser_telemetry_tool;
mod recording;
mod telemetry_frame;
mod ui;
mod analysis;
mod session;
use feed_state::*;
use ui::*;
@ -21,10 +25,10 @@ use crossterm::{
};
use ratatui::backend::CrosstermBackend;
use ratatui::Terminal;
use tokio::io::BufWriter;
use tokio::io::AsyncWriteExt;
use crate::feed_state::HandshakePhase;
use crate::recording::FEED_DEFS;
use crate::session::SessionManager;
use crate::telemetry_frame::TelemetryFrame;
const VERSION: &str = env!("CARGO_PKG_VERSION");
@ -35,6 +39,20 @@ const HANDSHAKE_RESPONSE_SIZE: usize = 408;
const HANDSHAKE_REQUEST: [u8; 12] = [1, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0];
const SUBSCRIBE_UPDATE: [u8; 12] = [1, 0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0];
const API_PORT: u16 = 8080;
fn api_port() -> u16 {
std::env::var("RUSTY_TELEMETRY_API_PORT")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(API_PORT)
}
fn api_bind_addr() -> String {
let host = std::env::var("RUSTY_TELEMETRY_API_HOST")
.unwrap_or_else(|_| "127.0.0.1".to_string());
format!("{}:{}", host, api_port())
}
fn decode_utf16le(bytes: &[u8]) -> String {
let chars: Vec<u16> = bytes
@ -74,20 +92,20 @@ struct FeedConfig {
static FEEDS: &[FeedConfig] = &[
FeedConfig {
port: 9996,
name: "ksudp",
port: FEED_DEFS[0].0,
name: FEED_DEFS[0].1,
label: "KSUDP",
needs_handshake: true,
},
FeedConfig {
port: 10101,
name: "telemetry_tool",
port: FEED_DEFS[1].0,
name: FEED_DEFS[1].1,
label: "Telemetry Tool",
needs_handshake: false,
},
FeedConfig {
port: 5606,
name: "pcars",
port: FEED_DEFS[2].0,
name: FEED_DEFS[2].1,
label: "PCARS",
needs_handshake: false,
},
@ -143,7 +161,7 @@ impl CachedStrings {
async fn run_feed(
feed: FeedConfig,
socket: tokio::net::UdpSocket,
session_dir: String,
recording_manager: Arc<recording::RecordingManager>,
state: Arc<std::sync::Mutex<FeedState>>,
mut shutdown_rx: tokio::sync::watch::Receiver<bool>,
) {
@ -264,29 +282,6 @@ async fn run_feed(
None
};
let file_path = format!(
"{}/feed_{}_{}.bin",
session_dir, feed.port, feed.name
);
let file = match tokio::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&file_path)
.await
{
Ok(f) => f,
Err(e) => {
eprintln!(
"[ERROR] Cannot open dump file for :{} ({}): {}",
feed.port, feed.name, e
);
if let Ok(mut s) = state.lock() {
s.set_error(&format!("Cannot open dump file: {}", e));
}
return;
}
};
let mut file = BufWriter::new(file);
let mut buf = vec![0u8; BUFFER_SIZE];
let mut string_cache = CachedStrings::new();
@ -340,23 +335,13 @@ async fn run_feed(
// Update state in a single brief lock acquisition.
if let Ok(mut s) = state.lock() {
s.record_packet(len);
if let Some(frame) = parsed {
s.update_frame(frame);
if let Some(ref frame) = parsed {
s.update_frame(frame.clone());
}
}
let header = (len as u32).to_be_bytes();
if let Err(e) = file.write_all(&header).await {
eprintln!(
"[ERROR] Write error on :{} ({}): {}",
feed.port, feed.name, e
);
}
if let Err(e) = file.write_all(&buf[..len]).await {
eprintln!(
"[ERROR] Write error on :{} ({}): {}",
feed.port, feed.name, e
);
if let Some(ref frame) = parsed {
recording_manager.push_frame(feed.name, frame.clone());
}
}
Err(e) => {
@ -388,20 +373,8 @@ async fn run_feed(
}
}
if let Err(e) = file.flush().await {
eprintln!(
"[ERROR] Flush error on :{} ({}): {}",
feed.port, feed.name, e
);
}
if let Err(e) = file.get_ref().sync_all().await {
eprintln!(
"[ERROR] Sync error on :{} ({}): {}",
feed.port, feed.name, e
);
}
eprintln!(
"[INFO] Feed :{} ({}) flushed and closed",
"[INFO] Feed :{} ({}) stopped",
feed.port, feed.name
);
}
@ -411,14 +384,43 @@ async fn main() {
install_panic_hook();
let listen = listen_addr();
let session_id = chrono::Local::now().format("%Y%m%d_%H%M%S").to_string();
let session_dir = format!("./data/{}", session_id);
let data_dir = std::path::PathBuf::from("./data");
if let Err(e) = tokio::fs::create_dir_all(&session_dir).await {
eprintln!("[ERROR] Cannot create data dir {}: {}", session_dir, e);
if let Err(e) = tokio::fs::create_dir_all(&data_dir).await {
eprintln!("[ERROR] Cannot create data dir: {}", e);
std::process::exit(1);
}
let recording_manager = Arc::new(recording::RecordingManager::new(&data_dir));
let session_manager = Arc::new(std::sync::Mutex::new(
SessionManager::new(recording_manager.clone())
));
let auto_start = recording::StartRecordingRequest {
name: Some(
chrono::Local::now()
.format("Session %Y-%m-%d %H:%M")
.to_string(),
),
game: Some(recording::Game::Unknown),
track: None,
car: None,
use_case: Some(recording::UseCase::General),
notes: None,
};
match recording_manager.start_recording(auto_start).await {
Ok(m) => {
eprintln!(
"[INFO] Auto-started recording: {} ({})",
m.name, m.id
);
}
Err(e) => {
eprintln!("[ERROR] Failed to start recording: {}", e);
std::process::exit(1);
}
}
let mut terminal = match setup_terminal() {
Ok(t) => t,
Err(e) => {
@ -483,14 +485,23 @@ async fn main() {
if let Some(socket) = socket_opt {
let feed_clone = feed.clone();
let dir = session_dir.clone();
let rm = recording_manager.clone();
let sh = shutdown_rx.clone();
feed_tasks.push(tokio::spawn(async move {
run_feed(feed_clone, socket, dir, state, sh).await;
run_feed(feed_clone, socket, rm, state, sh).await;
}));
}
}
let api_state = Arc::new(api::AppState {
recording_manager: recording_manager.clone(),
feed_displays: feed_displays.clone(),
session_manager: session_manager.clone(),
});
let api_addr = api_bind_addr();
eprintln!("[INFO] REST API server will listen on {}", api_addr);
tokio::spawn(api::run_server(api_state, api_addr, shutdown_rx.clone()));
let (event_tx, mut event_rx) = tokio::sync::mpsc::channel::<Event>(100);
let event_running = Arc::new(AtomicBool::new(true));
let event_running_clone = event_running.clone();
@ -508,6 +519,12 @@ async fn main() {
let mut render_interval = tokio::time::interval(Duration::from_millis(RENDER_INTERVAL_MS));
let running = Arc::new(AtomicBool::new(true));
let recording_rx_display = recording_manager.subscribe();
let get_session_display = || {
recording_rx_display.borrow().clone()
.map(|id| format!("./data/{}/", id))
.unwrap_or_else(|| "No active recording".to_string())
};
while running.load(Ordering::Relaxed) {
tokio::select! {
@ -528,8 +545,9 @@ async fn main() {
}
}
Event::Resize(_, _) => {
let session_display = get_session_display();
let _ = terminal.draw(|f| {
render(f, &feed_displays, &session_dir, VERSION);
render(f, &feed_displays, &session_display, VERSION);
});
}
_ => {}
@ -537,8 +555,9 @@ async fn main() {
}
}
_ = render_interval.tick() => {
let session_display = get_session_display();
let _ = terminal.draw(|f| {
render(f, &feed_displays, &session_dir, VERSION);
render(f, &feed_displays, &session_display, VERSION);
});
}
}
@ -547,6 +566,11 @@ async fn main() {
event_running.store(false, Ordering::Relaxed);
let _ = shutdown_tx.send(true);
// Flush in-memory ring buffers to disk before exit.
if let Err(e) = recording_manager.stop_recording().await {
eprintln!("[WARN] Failed to flush recording on shutdown: {}", e);
}
match tokio::time::timeout(
Duration::from_secs(3),
futures::future::join_all(feed_tasks),
@ -561,7 +585,7 @@ async fn main() {
}
}
Err(_) => {
eprintln!("[WARN] Timeout waiting for feeds to flush; forcing exit");
eprintln!("[WARN] Timeout waiting for feeds to stop; forcing exit");
}
}

508
src/recording.rs Normal file
View file

@ -0,0 +1,508 @@
//! Recording session management with always-on ring buffers.
//!
//! Each feed has a rolling ring buffer holding the last 5 minutes of telemetry frames.
//! Explicit recordings can flush the ring buffer to disk as JSONL files.
//! Use-case sessions extract frames from the ring buffer for analysis.
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, VecDeque};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::fs;
use tokio::sync::{Mutex, watch};
use crate::telemetry_frame::TelemetryFrame;
/// Feed port/name definitions. Must be kept in sync with FEEDS in main.rs.
pub const FEED_DEFS: &[(u16, &str)] = &[
(9996, "ksudp"),
(10101, "telemetry_tool"),
(5606, "pcars"),
];
/// Validate that a recording ID is a well-formed UUID.
/// Prevents path traversal attacks via the API.
fn validate_recording_id(id: &str) -> std::io::Result<()> {
if uuid::Uuid::parse_str(id).is_err() {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"Invalid recording ID format",
));
}
Ok(())
}
/// Generate the JSONL filename for a feed.
pub fn feed_filename(port: u16, name: &str) -> String {
format!("feed_{}_{}.jsonl", port, name)
}
/// Ring buffer duration in seconds.
const RING_BUFFER_SECS: u64 = 300; // 5 minutes
/// Maximum sample rate across all feeds (Hz).
const MAX_FEED_RATE_HZ: u64 = 60;
/// Ring buffer capacity = duration × max rate.
pub const RING_BUFFER_CAPACITY: usize = (RING_BUFFER_SECS * MAX_FEED_RATE_HZ) as usize;
/// In-memory ring buffer of parsed telemetry frames for a single feed.
/// Always active — captures the last 5 minutes of data regardless of recording state.
pub struct RingBuffer {
frames: VecDeque<TelemetryFrame>,
total_pushed: u64,
dropped_count: u64,
}
impl RingBuffer {
pub fn new() -> Self {
Self {
frames: VecDeque::with_capacity(RING_BUFFER_CAPACITY),
total_pushed: 0,
dropped_count: 0,
}
}
/// Push a frame. Oldest frame is evicted when at capacity (true ring behavior).
pub fn push(&mut self, frame: TelemetryFrame) {
if self.frames.len() >= RING_BUFFER_CAPACITY {
self.frames.pop_front();
self.dropped_count += 1;
}
self.frames.push_back(frame);
self.total_pushed += 1;
}
/// Returns all frames in chronological order.
pub fn frames(&self) -> &VecDeque<TelemetryFrame> {
&self.frames
}
/// Number of frames currently stored.
pub fn len(&self) -> usize {
self.frames.len()
}
/// Whether the buffer is empty.
pub fn is_empty(&self) -> bool {
self.frames.is_empty()
}
/// Total frames ever pushed (monotonically increasing).
pub fn total_pushed(&self) -> u64 {
self.total_pushed
}
/// Number of frames evicted due to ring overflow.
pub fn dropped_count(&self) -> u64 {
self.dropped_count
}
/// Returns a marker representing the current position (for extract_since).
pub fn marker(&self) -> u64 {
self.total_pushed
}
/// Extract frames pushed since the given marker.
/// Returns all frames if the marker predates the buffer contents.
/// Returns empty vec if marker is at or past the current position.
pub fn extract_since(&self, marker: u64) -> Vec<TelemetryFrame> {
if marker >= self.total_pushed {
return Vec::new();
}
let buffer_start = self.total_pushed.saturating_sub(self.frames.len() as u64);
let start_offset = if marker < buffer_start {
0
} else {
(marker - buffer_start) as usize
};
self.frames.iter().skip(start_offset).cloned().collect()
}
/// Snapshot of all current frames.
pub fn snapshot(&self) -> Vec<TelemetryFrame> {
self.frames.iter().cloned().collect()
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum UseCase {
ShiftPoints,
RacingLine,
BrakingAnalysis,
LapTimes,
SectorAnalysis,
TrackMap,
General,
}
impl UseCase {
pub fn all() -> Vec<Self> {
vec![
UseCase::ShiftPoints,
UseCase::RacingLine,
UseCase::BrakingAnalysis,
UseCase::LapTimes,
UseCase::SectorAnalysis,
UseCase::TrackMap,
UseCase::General,
]
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum Game {
AssettoCorsa,
AssettoCorsaCompetizione,
ProjectCars1,
ProjectCars2,
Unknown,
}
impl Game {
/// Returns the list of valid game choices for API consumers (excludes Unknown).
pub fn valid_choices() -> Vec<Self> {
vec![
Game::AssettoCorsa,
Game::AssettoCorsaCompetizione,
Game::ProjectCars1,
Game::ProjectCars2,
]
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FeedStats {
pub port: u16,
pub name: String,
pub bytes: u64,
pub file: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum RecordingStatus {
Recording,
Stopped,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RecordingManifest {
pub id: String,
pub name: String,
pub game: Game,
pub track: String,
pub car: String,
pub use_case: UseCase,
pub status: RecordingStatus,
pub created_at: String,
pub stopped_at: Option<String>,
pub feeds: Vec<FeedStats>,
pub notes: String,
}
impl RecordingManifest {
pub async fn save(&self, data_dir: &Path) -> std::io::Result<()> {
let dir = data_dir.join(&self.id);
fs::create_dir_all(&dir).await?;
let json = serde_json::to_string_pretty(self)?;
fs::write(dir.join("manifest.json"), json).await?;
Ok(())
}
pub async fn load(dir: &Path) -> std::io::Result<Self> {
let json = fs::read_to_string(dir.join("manifest.json")).await?;
let manifest: RecordingManifest = serde_json::from_str(&json)?;
Ok(manifest)
}
pub fn dir_path(&self, data_dir: &Path) -> PathBuf {
data_dir.join(&self.id)
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StartRecordingRequest {
pub name: Option<String>,
pub game: Option<Game>,
pub track: Option<String>,
pub car: Option<String>,
pub use_case: Option<UseCase>,
pub notes: Option<String>,
}
/// Manages always-on ring buffers and optional disk-based recordings.
pub struct RecordingManager {
data_dir: PathBuf,
active: Mutex<Option<String>>,
recording_change_tx: watch::Sender<Option<String>>,
/// Per-feed ring buffers — always active, created at startup.
ring_buffers: HashMap<String, Arc<std::sync::Mutex<RingBuffer>>>,
}
impl RecordingManager {
pub fn new(data_dir: &Path) -> Self {
let (tx, _) = watch::channel(None);
let mut ring_buffers = HashMap::new();
for &(_port, name) in FEED_DEFS {
ring_buffers.insert(name.to_string(), Arc::new(std::sync::Mutex::new(RingBuffer::new())));
}
Self {
data_dir: data_dir.to_path_buf(),
active: Mutex::new(None),
recording_change_tx: tx,
ring_buffers,
}
}
pub fn data_dir(&self) -> &Path {
&self.data_dir
}
pub fn subscribe(&self) -> watch::Receiver<Option<String>> {
self.recording_change_tx.subscribe()
}
/// Push a telemetry frame into the feed's ring buffer. Always succeeds.
pub fn push_frame(&self, feed_name: &str, frame: TelemetryFrame) {
if let Some(buf) = self.ring_buffers.get(feed_name) {
if let Ok(mut b) = buf.lock() {
b.push(frame);
}
}
}
/// Get the current ring buffer marker for a feed (for extract_since).
pub fn ring_marker(&self, feed_name: &str) -> u64 {
self.ring_buffers.get(feed_name)
.and_then(|b| b.lock().ok())
.map(|b| b.marker())
.unwrap_or(0)
}
/// Extract frames from a feed's ring buffer since the given marker.
pub fn extract_since(&self, feed_name: &str, marker: u64) -> Vec<TelemetryFrame> {
self.ring_buffers.get(feed_name)
.and_then(|b| b.lock().ok())
.map(|b| b.extract_since(marker))
.unwrap_or_default()
}
/// Read all current frames from a feed's ring buffer.
pub fn read_frames(&self, feed_name: &str) -> Option<Vec<TelemetryFrame>> {
self.ring_buffers.get(feed_name)
.and_then(|b| b.lock().ok())
.map(|b| b.snapshot())
}
/// Get ring buffer stats for a feed.
pub fn ring_stats(&self, feed_name: &str) -> Option<(usize, u64, u64)> {
self.ring_buffers.get(feed_name)
.and_then(|b| b.lock().ok())
.map(|b| (b.len(), b.total_pushed(), b.dropped_count()))
}
pub async fn start_recording(
&self,
req: StartRecordingRequest,
) -> std::io::Result<RecordingManifest> {
// NOTE: Hold the active lock across I/O to prevent concurrent start_recording calls
// from creating orphaned recordings. For a single-user local tool, the contention
// cost is negligible compared to the correctness benefit.
let mut active = self.active.lock().await;
if let Some(ref id) = *active {
self.stop_recording_internal(id).await?;
}
let id = uuid::Uuid::new_v4().to_string();
let now = chrono::Utc::now().to_rfc3339();
let feed_name = |port: u16, name: &str| -> FeedStats {
FeedStats {
port,
name: name.to_string(),
bytes: 0,
file: feed_filename(port, name),
}
};
let manifest = RecordingManifest {
id: id.clone(),
name: req.name.unwrap_or_else(|| {
chrono::Local::now()
.format("Recording %Y-%m-%d %H:%M")
.to_string()
}),
game: req.game.unwrap_or(Game::Unknown),
track: req.track.unwrap_or_default(),
car: req.car.unwrap_or_default(),
use_case: req.use_case.unwrap_or(UseCase::General),
status: RecordingStatus::Recording,
created_at: now,
stopped_at: None,
feeds: FEED_DEFS.iter().map(|&(port, name)| feed_name(port, name)).collect(),
notes: req.notes.unwrap_or_default(),
};
let dir = manifest.dir_path(&self.data_dir);
fs::create_dir_all(&dir).await?;
manifest.save(&self.data_dir).await?;
*active = Some(id.clone());
let _ = self.recording_change_tx.send(Some(id));
Ok(manifest)
}
pub async fn stop_recording(&self) -> std::io::Result<Option<RecordingManifest>> {
let mut active = self.active.lock().await;
let id = match active.clone() {
Some(id) => id,
None => return Ok(None),
};
let result = self.stop_recording_internal(&id).await;
if result.is_ok() {
active.take();
let _ = self.recording_change_tx.send(None);
}
result.map(Some)
}
pub async fn stop_recording_internal(&self, id: &str) -> std::io::Result<RecordingManifest> {
validate_recording_id(id)?;
let dir = self.data_dir.join(id);
if !dir.exists() {
return Err(std::io::Error::new(
std::io::ErrorKind::NotFound,
"Recording not found",
));
}
let mut manifest = RecordingManifest::load(&dir).await?;
// Flush ring buffers to disk as JSONL files.
// Step 1: Clone frame data while locked (fast memcpy).
let frame_snapshots: Vec<(PathBuf, Vec<TelemetryFrame>)> = {
let mut snapshots = Vec::new();
for &(port, name) in FEED_DEFS {
if let Some(buf_arc) = self.ring_buffers.get(name) {
if let Ok(buf) = buf_arc.lock() {
if !buf.is_empty() {
let file_path = dir.join(feed_filename(port, name));
let frames: Vec<TelemetryFrame> = buf.frames().iter().cloned().collect();
snapshots.push((file_path, frames));
}
}
}
}
snapshots
};
// Step 2: Serialize to JSONL outside the lock.
let mut flush_errors = Vec::new();
for (path, frames) in &frame_snapshots {
let mut content = String::with_capacity(frames.len() * 400);
for frame in frames {
if let Ok(line) = serde_json::to_string(frame) {
content.push_str(&line);
content.push('\n');
}
}
if let Err(e) = fs::write(path, content).await {
eprintln!("[WARN] Failed to flush frames to {}: {}", path.display(), e);
flush_errors.push(format!("{}: {}", path.display(), e));
}
}
if !flush_errors.is_empty() {
return Err(std::io::Error::new(
std::io::ErrorKind::Other,
format!("Flush errors: {}", flush_errors.join(", ")),
));
}
// Update feed stats from actual file sizes AFTER flush.
for feed in &mut manifest.feeds {
let feed_path = dir.join(&feed.file);
if feed_path.exists() {
if let Ok(metadata) = fs::metadata(&feed_path).await {
feed.bytes = metadata.len();
}
}
}
manifest.status = RecordingStatus::Stopped;
manifest.stopped_at = Some(chrono::Utc::now().to_rfc3339());
manifest.save(&self.data_dir).await?;
Ok(manifest)
}
pub async fn list_recordings(&self) -> std::io::Result<Vec<RecordingManifest>> {
let mut recordings = Vec::new();
let mut entries = fs::read_dir(&self.data_dir).await?;
while let Some(entry) = entries.next_entry().await? {
let path = entry.path();
if path.is_dir() && path.join("manifest.json").exists() {
match RecordingManifest::load(&path).await {
Ok(m) => recordings.push(m),
Err(e) => {
eprintln!(
"[WARN] Failed to load manifest from {}: {}",
path.display(),
e
);
}
}
}
}
recordings.sort_by(|a, b| b.created_at.cmp(&a.created_at));
Ok(recordings)
}
pub async fn get_recording(&self, id: &str) -> std::io::Result<RecordingManifest> {
validate_recording_id(id)?;
let dir = self.data_dir.join(id);
if !dir.exists() {
return Err(std::io::Error::new(
std::io::ErrorKind::NotFound,
"Recording not found",
));
}
RecordingManifest::load(&dir).await
}
pub async fn delete_recording(&self, id: &str) -> std::io::Result<()> {
validate_recording_id(id)?;
{
let active = self.active.lock().await;
if active.as_deref() == Some(id) {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"Cannot delete an active recording. Stop it first.",
));
}
}
let dir = self.data_dir.join(id);
if !dir.exists() {
return Err(std::io::Error::new(
std::io::ErrorKind::NotFound,
"Recording not found",
));
}
fs::remove_dir_all(&dir).await
}
pub async fn active_recording_id(&self) -> Option<String> {
self.active.lock().await.clone()
}
pub async fn clear_active_if_matches(&self, id: &str) {
let mut active = self.active.lock().await;
if let Some(ref active_id) = *active {
if active_id == id {
*active = None;
let _ = self.recording_change_tx.send(None);
}
}
}
}

273
src/session.rs Normal file
View file

@ -0,0 +1,273 @@
//! Use-case driven session management for telemetry analysis.
//!
//! Users create sessions tied to specific analysis goals (shift point analysis,
//! 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.
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use crate::recording::{RecordingManager, UseCase};
use crate::telemetry_frame::TelemetryFrame;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum SessionStatus {
Active,
Completed,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum RunStatus {
Recording,
Stopped,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SessionRun {
pub id: String,
pub label: String,
pub feed_name: String,
pub status: RunStatus,
pub start_marker: u64,
pub end_marker: Option<u64>,
pub frame_count: usize,
pub data_truncated: bool,
#[serde(skip)]
pub frames: Vec<TelemetryFrame>,
}
impl SessionRun {
fn summary_clone(&self) -> Self {
Self {
id: self.id.clone(),
label: self.label.clone(),
feed_name: self.feed_name.clone(),
status: self.status.clone(),
start_marker: self.start_marker,
end_marker: self.end_marker,
frame_count: self.frame_count,
data_truncated: self.data_truncated,
frames: Vec::new(),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct UseCaseSession {
pub id: String,
pub use_case: UseCase,
pub label: String,
pub status: SessionStatus,
pub created_at: String,
pub completed_at: Option<String>,
pub runs: Vec<SessionRun>,
pub analysis: Option<serde_json::Value>,
}
impl UseCaseSession {
fn summary_clone(&self) -> Self {
Self {
id: self.id.clone(),
use_case: self.use_case.clone(),
label: self.label.clone(),
status: self.status.clone(),
created_at: self.created_at.clone(),
completed_at: self.completed_at.clone(),
runs: self.runs.iter().map(|r| r.summary_clone()).collect(),
analysis: self.analysis.clone(),
}
}
}
pub struct SessionManager {
sessions: HashMap<String, UseCaseSession>,
recording_manager: Arc<RecordingManager>,
}
impl SessionManager {
pub fn new(recording_manager: Arc<RecordingManager>) -> Self {
Self {
sessions: HashMap::new(),
recording_manager,
}
}
pub fn create_session(
&mut self,
use_case: UseCase,
label: Option<String>,
) -> Result<UseCaseSession, String> {
let id = uuid::Uuid::new_v4().to_string();
let now = chrono::Utc::now().to_rfc3339();
let session = UseCaseSession {
id: id.clone(),
label: label.unwrap_or_else(|| format!("{:?} Session", use_case)),
use_case,
status: SessionStatus::Active,
created_at: now,
completed_at: None,
runs: Vec::new(),
analysis: None,
};
self.sessions.insert(id, session.clone());
Ok(session)
}
pub fn list_sessions(&mut self) -> Vec<UseCaseSession> {
let mut sessions: Vec<UseCaseSession> = self
.sessions
.values()
.map(|s| s.summary_clone())
.collect();
sessions.sort_by(|a, b| b.created_at.cmp(&a.created_at));
sessions
}
pub fn get_session(&self, id: &str) -> Option<UseCaseSession> {
self.sessions.get(id).map(|s| s.summary_clone())
}
pub fn start_run(
&mut self,
session_id: &str,
feed_name: &str,
label: Option<String>,
) -> Result<SessionRun, String> {
let session = self
.sessions
.get_mut(session_id)
.ok_or_else(|| format!("Session '{}' not found", session_id))?;
if session.status != SessionStatus::Active {
return Err(format!("Session '{}' is not active", session_id));
}
let marker = self.recording_manager.ring_marker(feed_name);
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(),
status: RunStatus::Recording,
start_marker: marker,
end_marker: None,
frame_count: 0,
data_truncated: false,
frames: Vec::new(),
};
let summary = run.summary_clone();
session.runs.push(run);
Ok(summary)
}
pub fn stop_run(
&mut self,
session_id: &str,
run_id: &str,
) -> Result<SessionRun, String> {
let session = self
.sessions
.get_mut(session_id)
.ok_or_else(|| format!("Session '{}' not found", session_id))?;
let run = session
.runs
.iter_mut()
.find(|r| r.id == run_id)
.ok_or_else(|| format!("Run '{}' not found in session '{}'", run_id, session_id))?;
if run.status != RunStatus::Recording {
return Err(format!("Run '{}' is not recording", run_id));
}
let end_marker = self.recording_manager.ring_marker(&run.feed_name);
let frames = self
.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;
run.end_marker = Some(end_marker);
run.frame_count = frames.len();
run.frames = frames;
run.status = RunStatus::Stopped;
run.data_truncated = data_truncated;
Ok(run.summary_clone())
}
pub fn complete_session(&mut self, id: &str) -> Result<UseCaseSession, String> {
let session = self
.sessions
.get_mut(id)
.ok_or_else(|| format!("Session '{}' not found", id))?;
if session.status != SessionStatus::Active {
return Err(format!("Session '{}' is not active", id));
}
for run in &mut session.runs {
if run.status == RunStatus::Recording {
let end_marker = self.recording_manager.ring_marker(&run.feed_name);
let frames = self
.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;
run.end_marker = Some(end_marker);
run.frame_count = frames.len();
run.frames = frames;
run.status = RunStatus::Stopped;
run.data_truncated = data_truncated;
}
}
session.status = SessionStatus::Completed;
session.completed_at = Some(chrono::Utc::now().to_rfc3339());
Ok(session.summary_clone())
}
pub fn set_analysis(
&mut self,
session_id: &str,
result: serde_json::Value,
) -> Result<(), String> {
let session = self
.sessions
.get_mut(session_id)
.ok_or_else(|| format!("Session '{}' not found", session_id))?;
session.analysis = Some(result);
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<TelemetryFrame>>, Vec<String>)> {
let session = self.sessions.get(session_id)?;
let mut frames_by_run = Vec::new();
let mut run_labels = Vec::new();
for run in &session.runs {
frames_by_run.push(run.frames.clone());
run_labels.push(run.label.clone());
}
Some((session.use_case.clone(), frames_by_run, run_labels))
}
pub fn delete_session(&mut self, id: &str) -> Result<(), String> {
self.sessions
.remove(id)
.ok_or_else(|| format!("Session '{}' not found", id))?;
Ok(())
}
}

21
src/telemetry_frame.rs
View file

@ -3,8 +3,10 @@
//! Both the KSUDP (port 9996) and Telemetry Tool (port 10101) parsers produce
//! `TelemetryFrame` instances from their raw binary data.
use serde::Serialize;
/// Identifies which feed source produced this frame.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
pub enum FeedSource {
Ksudp,
TelemetryTool,
@ -18,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)]
#[derive(Debug, Clone, Serialize)]
pub struct TelemetryFrame {
// ── Speed ──────────────────────────────────────────────────────
pub speed_kmh: f32,
@ -233,10 +235,15 @@ impl TelemetryFrame {
/// Gear display: 0=Reverse, 1=Neutral, 2+=Gear N-1
pub fn gear_display(&self) -> String {
match self.gear {
0 => "R".to_string(),
1 => "N".to_string(),
n => format!("{}", n - 1),
}
gear_display(self.gear)
}
}
/// AC gear encoding: 0=Reverse, 1=Neutral, 2+=Gear N-1.
pub fn gear_display(gear: i32) -> String {
match gear {
0 => "R".to_string(),
1 => "N".to_string(),
n => format!("{}", n - 1),
}
}

2
src/ui.rs
View file

@ -208,7 +208,7 @@ pub fn render(
let footer_text = Line::from(vec![
Span::styled(
format!(" Session: {}/ ", session_dir),
format!(" Recording: {}/ ", session_dir),
Style::default().fg(Color::White),
),
Span::raw(" "),