rusty-telemetry/src/api.rs
Jan-Peter v. Hunnius ed7dbcf1c9 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/<id>.json,
  frames to <id>.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
2026-06-24 09:32:16 +02:00

768 lines
24 KiB
Rust

//! 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::{delete, get, post},
};
use serde::{Deserialize, Serialize};
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;
use crate::telemetry_frame::TelemetryFrame;
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>>,
pub track_model_store: Arc<crate::track_model::TrackModelStore>,
}
#[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>,
#[serde(default)]
merge_feeds: Vec<String>,
}
#[derive(Deserialize)]
struct BuildTrackRequest {
session_id: String,
game: Option<String>,
track_name: Option<String>,
}
async fn create_session(
State(state): State<Arc<AppState>>,
Json(req): Json<CreateSessionRequest>,
) -> Response {
let recording_id = state.recording_manager.active_recording_id().await;
let mut mgr = lock_sessions(&state);
match mgr.create_session(req.use_case, req.label, recording_id) {
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, req.merge_feeds) {
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 delete_run(
State(state): State<Arc<AppState>>,
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<Arc<AppState>>,
Path(id): Path<String>,
) -> 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<AppState>,
result: &mut serde_json::Value,
recording_id: Option<&str>,
) {
let rec_id = match recording_id {
Some(id) => id,
None => return,
};
if let Ok(manifest) = state.recording_manager.get_recording(rec_id).await {
if let Some(obj) = result.as_object_mut() {
obj.insert("car".to_string(), serde_json::json!(manifest.car));
obj.insert("game".to_string(), serde_json::json!(manifest.game));
obj.insert("track".to_string(), serde_json::json!(manifest.track));
}
}
}
async fn execute_and_store_analysis(
state: &Arc<AppState>,
session_id: &str,
use_case: UseCase,
frames_by_run: Vec<Vec<TelemetryFrame>>,
run_labels: Vec<String>,
recording_id: Option<String>,
) -> serde_json::Value {
let mut result = analysis::analyze(&use_case, &frames_by_run, &run_labels);
enrich_with_recording_meta(state, &mut result, recording_id.as_deref()).await;
let mut mgr = lock_sessions(state);
let _ = mgr.set_analysis(session_id, result.clone());
result
}
async fn complete_session(
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let (use_case, frames_by_run, run_labels, recording_id) = {
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)) => {
let rec_id = mgr.get_session(&id).and_then(|s| s.recording_id.clone());
(uc, frames, labels, rec_id)
}
None => return err(StatusCode::NOT_FOUND, "Session not found after completion"),
}
};
execute_and_store_analysis(&state, &id, use_case, frames_by_run, run_labels, recording_id).await;
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, recording_id) = {
let mut 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"),
}
let rec_id = mgr.get_session(&id).unwrap().recording_id.clone();
match mgr.collect_run_frames(&id) {
Some((uc, frames, labels)) => (uc, frames, labels, rec_id),
None => return err(StatusCode::NOT_FOUND, "Session not found"),
}
};
let result = execute_and_store_analysis(&state, &id, use_case, frames_by_run, run_labels, recording_id).await;
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,
})
}
#[derive(Serialize)]
struct UseCaseInfo {
id: String,
name: String,
guidance: String,
}
async fn list_tracks(State(state): State<Arc<AppState>>) -> Response {
match state.track_model_store.list().await {
Ok(summaries) => ok_json(summaries),
Err(e) => err(
StatusCode::INTERNAL_SERVER_ERROR,
&format!("Failed to list tracks: {}", e),
),
}
}
async fn get_track(
State(state): State<Arc<AppState>>,
Path((game, track_name)): Path<(String, String)>,
) -> Response {
match state.track_model_store.load(&game, &track_name).await {
Ok(model) => ok_json(model),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
err(StatusCode::NOT_FOUND, "Track model 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 load track model: {}", e),
),
}
}
async fn delete_track(
State(state): State<Arc<AppState>>,
Path((game, track_name)): Path<(String, String)>,
) -> Response {
match state.track_model_store.delete(&game, &track_name).await {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
err(StatusCode::NOT_FOUND, "Track model 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 delete track model: {}", e),
),
}
}
async fn get_track_corners(
State(state): State<Arc<AppState>>,
Path((game, track_name)): Path<(String, String)>,
) -> Response {
match state.track_model_store.load(&game, &track_name).await {
Ok(model) => ok_json(&model.corners),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
err(StatusCode::NOT_FOUND, "Track model 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 load track model: {}", e),
),
}
}
async fn build_track(
State(state): State<Arc<AppState>>,
Json(req): Json<BuildTrackRequest>,
) -> Response {
let rec_id = {
let mgr = lock_sessions(&state);
let session = match mgr.get_session(&req.session_id) {
Some(s) => s,
None => return err(StatusCode::NOT_FOUND, "Session not found"),
};
match session.recording_id.clone() {
Some(id) => id,
None => return err(StatusCode::BAD_REQUEST, "Session has no associated recording"),
}
};
let manifest = match state.recording_manager.get_recording(&rec_id).await {
Ok(m) => m,
Err(_) => return err(StatusCode::NOT_FOUND, "Recording not found"),
};
let game = req.game.unwrap_or_else(|| manifest.game.to_track_dir_name().to_string());
let track_name = req.track_name.unwrap_or(manifest.track);
let boundaries = {
let mgr = lock_sessions(&state);
let session = match mgr.get_session(&req.session_id) {
Some(s) => s,
None => return err(StatusCode::NOT_FOUND, "Session not found"),
};
if session.use_case != UseCase::TrackMapping {
return err(StatusCode::BAD_REQUEST, "Session is not a track_mapping session");
}
if session.status != crate::session::SessionStatus::Completed {
return err(StatusCode::BAD_REQUEST, "Session must be completed first");
}
let analysis = match &session.analysis {
Some(a) => a.clone(),
None => return err(StatusCode::BAD_REQUEST, "Session has no analysis results"),
};
let boundaries_val = match analysis.get("boundaries") {
Some(v) => v.clone(),
None => return err(StatusCode::BAD_REQUEST, "No boundaries in analysis"),
};
match serde_json::from_value::<Vec<crate::analysis::TrackBoundary>>(boundaries_val) {
Ok(b) => b,
Err(e) => return err(StatusCode::BAD_REQUEST, &format!("Failed to parse boundaries: {}", e)),
}
};
let source = crate::track_model::TrackModelSource::BoundaryRecording;
let track_model = match analysis::build_track_model(&boundaries, game.clone(), track_name.clone(), source) {
Ok(m) => m,
Err(e) => return err(StatusCode::BAD_REQUEST, &format!("Failed to build track model: {}", e)),
};
match state.track_model_store.save(&track_model).await {
Ok(()) => (StatusCode::CREATED, Json(track_model)).into_response(),
Err(e) => err(StatusCode::INTERNAL_SERVER_ERROR, &format!("Failed to save track model: {}", e)),
}
}
async fn list_use_cases() -> Response {
let use_cases: Vec<UseCaseInfo> = UseCase::all()
.into_iter()
.map(|uc| UseCaseInfo {
id: serde_json::to_value(&uc)
.ok()
.and_then(|v| v.as_str().map(String::from))
.unwrap_or_default(),
name: format!("{:?}", uc),
guidance: uc.guidance().to_string(),
})
.collect();
ok_json(use_cases)
}
async fn list_games() -> Response {
ok_json(Game::valid_choices())
}
async fn list_car_profiles(
State(state): State<Arc<AppState>>,
) -> Response {
let mgr = lock_sessions(&state);
ok_json(characteristics::car_profiles(&mgr))
}
async fn get_car_profile(
State(state): State<Arc<AppState>>,
Path(car): Path<String>,
) -> 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<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}/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))
.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/tracks", get(list_tracks))
.route("/api/tracks/{game}/{track_name}", get(get_track).delete(delete_track))
.route("/api/tracks/{game}/{track_name}/corners", get(get_track_corners))
.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)
}
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.");
}