diff --git a/src/api.rs b/src/api.rs index d4343f4..500ff91 100644 --- a/src/api.rs +++ b/src/api.rs @@ -41,6 +41,7 @@ pub struct AppState { pub feed_displays: Vec<(FeedDisplayInfo, Arc>)>, pub session_manager: Arc>, pub track_model_store: Arc, + pub game_info: Arc>>, } #[derive(Serialize)] @@ -121,7 +122,19 @@ async fn create_session( 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(), + Ok(session) => { + let (car, track) = state.game_info.lock() + .map(|gi| { + let (c, t) = gi.as_ref().map_or((None, None), |i| { + (i.car_name.clone(), i.track_name.clone()) + }); + (c, t) + }) + .unwrap_or((None, None)); + mgr.fill_session_metadata(&session.id, car, track); + let updated = mgr.get_session(&session.id); + (StatusCode::CREATED, Json(updated.unwrap_or(session))).into_response() + } Err(msg) => err(StatusCode::BAD_REQUEST, &msg), } } @@ -334,8 +347,20 @@ async fn get_recording( async fn start_recording( State(state): State>, - Json(req): Json, + Json(mut req): Json, ) -> Response { + // Auto-fill game/track/car from handshake metadata if not specified. + if let Ok(Some(info)) = state.game_info.lock().as_deref() { + if req.game.is_none() { + req.game = Some(Game::AssettoCorsa); + } + if req.track.is_none() { + req.track = info.track_name.clone(); + } + if req.car.is_none() { + req.car = info.car_name.clone(); + } + } match state.recording_manager.start_recording(req).await { Ok(manifest) => (StatusCode::CREATED, Json(manifest)).into_response(), Err(e) => err( @@ -356,6 +381,11 @@ async fn stop_recording(State(state): State>) -> Response { } } +async fn get_game_info(State(state): State>) -> Response { + let info = state.game_info.lock().map(|gi| gi.clone()).unwrap_or(None); + ok_json(info.unwrap_or_default()) +} + async fn stop_specific_recording( State(state): State>, Path(id): Path, @@ -746,6 +776,7 @@ pub fn build_router(state: Arc) -> Router { get(get_feed_data), ) .route("/api/live", get(get_live)) + .route("/api/game-info", get(get_game_info)) .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)) diff --git a/src/main.rs b/src/main.rs index 9ee04e2..31da406 100644 --- a/src/main.rs +++ b/src/main.rs @@ -28,6 +28,7 @@ mod analysis; mod characteristics; mod session; +use serde::Serialize; use feed_state::*; use ui::*; @@ -53,13 +54,50 @@ use crate::warnings::Warnings; const VERSION: &str = env!("CARGO_PKG_VERSION"); const BUFFER_SIZE: usize = 65535; const RENDER_INTERVAL_MS: u64 = 250; -const HANDSHAKE_INTERVAL_SECS: u64 = 10; +const HANDSHAKE_INTERVAL_SECS: u64 = 5; const HANDSHAKE_RESPONSE_SIZE: usize = 408; +const STALE_TIMEOUT_SECS: u64 = 5; 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; +/// Metadata extracted from the KSUDP handshake response. +/// Updated whenever AC reconnects (track/car change) and shared +/// with the API for auto-filling recording and session metadata. +#[derive(Debug, Clone, Serialize, Default)] +pub struct GameInfo { + #[serde(skip_serializing_if = "Option::is_none")] + pub car_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub track_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub driver_name: Option, +} + +/// Parse a 408-byte KSUDP HandshakerResponse into GameInfo. +/// Layout: car_name (0-100), driver_name (100-200), identifier (200-204), +/// version (204-208), track_name (208-308), track_config (308-408). +fn parse_handshake_response(data: &[u8]) -> Option { + if data.len() < HANDSHAKE_RESPONSE_SIZE { + return None; + } + let car_name = decode_utf16le(&data[0..100]); + let driver_name = decode_utf16le(&data[100..200]); + let track_name = decode_utf16le(&data[208..308]); + + let clean = |s: String| -> Option { + let trimmed = s.trim_end_matches('\0').trim().to_string(); + if trimmed.is_empty() { None } else { Some(trimmed) } + }; + + Some(GameInfo { + car_name: clean(car_name), + driver_name: clean(driver_name), + track_name: clean(track_name), + }) +} + fn api_port() -> u16 { std::env::var("RUSTY_TELEMETRY_API_PORT") .ok() @@ -183,6 +221,7 @@ async fn run_feed( state: Arc>, warnings: Warnings, mut shutdown_rx: tokio::sync::watch::Receiver, + game_info: Arc>>, ) { let handshake_target: Option = if feed.needs_handshake { let target: std::net::SocketAddr = match format!("{}:{}", game_host(), feed.port).parse() { @@ -221,34 +260,25 @@ async fn run_feed( .await { Ok(Ok((len, _src))) if len == HANDSHAKE_RESPONSE_SIZE => { - let car_name = decode_utf16le(&hs_buf[0..100]); - let driver_name = decode_utf16le(&hs_buf[100..200]); - let identifier = i32::from_le_bytes([ - hs_buf[200], hs_buf[201], hs_buf[202], hs_buf[203], - ]); - let version = i32::from_le_bytes([ - hs_buf[204], hs_buf[205], hs_buf[206], hs_buf[207], - ]); - let track_name = decode_utf16le(&hs_buf[208..308]); - let track_config = decode_utf16le(&hs_buf[308..408]); - - warnings.info(&format!( - "{} :{} Phase 2: HandshakerResponse received", - feed.label, feed.port - )); - warnings.info(&format!( - " car_name=\"{}\" driver_name=\"{}\" id={} ver={}", - car_name, driver_name, identifier, version - )); - warnings.info(&format!( - " track_name=\"{}\" track_config=\"{}\"", - track_name, track_config - )); - - if identifier != 4242 { + let info = parse_handshake_response(&hs_buf[..len]); + if let Some(ref info) = info { + warnings.info(&format!( + "{} :{} Phase 2: HandshakerResponse received", + feed.label, feed.port + )); + warnings.info(&format!( + " car=\"{}\" driver=\"{}\" track=\"{}\"", + info.car_name.as_deref().unwrap_or("?"), + info.driver_name.as_deref().unwrap_or("?"), + info.track_name.as_deref().unwrap_or("?"), + )); + if let Ok(mut gi) = game_info.lock() { + *gi = Some(info.clone()); + } + } else { warnings.warn(&format!( - "{} :{} Unexpected handshake identifier: {} (expected 4242)", - feed.label, feed.port, identifier + "{} :{} Phase 2: Failed to parse handshake response", + feed.label, feed.port )); } } @@ -300,9 +330,13 @@ async fn run_feed( let mut buf = vec![0u8; BUFFER_SIZE]; let mut string_cache = CachedStrings::new(); + let mut last_recv_time: Option = None; + let stale_timeout = Duration::from_secs(STALE_TIMEOUT_SECS); let mut hs_interval: Option = handshake_target.map(|_| { - tokio::time::interval(Duration::from_secs(HANDSHAKE_INTERVAL_SECS)) + let mut iv = tokio::time::interval(Duration::from_secs(HANDSHAKE_INTERVAL_SECS)); + iv.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + iv }); loop { @@ -310,12 +344,32 @@ async fn run_feed( result = socket.recv_from(&mut buf) => { match result { Ok((len, _src)) => { - let parsed: Option = if feed.needs_handshake { - if len != HANDSHAKE_RESPONSE_SIZE { - parser_ksudp::parse_rt_car_info(&buf[..len]).ok() - } else { - None + // Handle handshake response packets (AC reconnected with new track/car). + if feed.needs_handshake && len == HANDSHAKE_RESPONSE_SIZE { + if let Some(info) = parse_handshake_response(&buf[..len]) { + warnings.info(&format!( + "{} :{} Re-handshake: car=\"{}\" track=\"{}\"", + feed.label, feed.port, + info.car_name.as_deref().unwrap_or("?"), + info.track_name.as_deref().unwrap_or("?"), + )); + if let Ok(mut gi) = game_info.lock() { + *gi = Some(info); + } + if let Some(target) = handshake_target { + let _ = socket.send_to(&SUBSCRIBE_UPDATE, target).await; + } + if let Ok(mut s) = state.lock() { + s.set_handshake_phase(HandshakePhase::Subscribed); + } } + continue; + } + + last_recv_time = Some(std::time::Instant::now()); + + let parsed: Option = if feed.needs_handshake { + parser_ksudp::parse_rt_car_info(&buf[..len]).ok() } else if feed.name == "pcars" { match parser_pcars::parse_pcars_packet(&buf[..len]) { Ok(parser_pcars::PcarsPacket::V1Telemetry(f)) => Some(f), @@ -369,11 +423,24 @@ async fn run_feed( } } => { if let Some(target) = handshake_target { - if let Err(e) = socket.send_to(&SUBSCRIBE_UPDATE, target).await { - warnings.warn(&format!( - "Periodic SUBSCRIBE_UPDATE send failed for :{}: {}", - feed.port, e - )); + let is_stale = last_recv_time.is_none_or(|t| t.elapsed() > stale_timeout); + if is_stale { + if let Ok(mut s) = state.lock() { + s.set_handshake_phase(HandshakePhase::WaitingForResponse); + } + if let Err(e) = socket.send_to(&HANDSHAKE_REQUEST, target).await { + warnings.warn(&format!( + "HANDSHAKE_REQUEST send failed for :{}: {}", + feed.port, e + )); + } + } else { + if let Err(e) = socket.send_to(&SUBSCRIBE_UPDATE, target).await { + warnings.warn(&format!( + "SUBSCRIBE_UPDATE send failed for :{}: {}", + feed.port, e + )); + } } } } @@ -452,6 +519,9 @@ async fn main() { let mut feed_tasks: Vec> = Vec::new(); let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false); + let game_info: Arc>> = + Arc::new(std::sync::Mutex::new(None)); + for feed in FEEDS { let state = Arc::new(std::sync::Mutex::new(FeedState::new())); @@ -507,8 +577,9 @@ async fn main() { let rm = recording_manager.clone(); let w = warnings.clone(); let sh = shutdown_rx.clone(); + let gi = game_info.clone(); feed_tasks.push(tokio::spawn(async move { - run_feed(feed_clone, socket, rm, state, w, sh).await; + run_feed(feed_clone, socket, rm, state, w, sh, gi).await; })); } } @@ -518,6 +589,7 @@ async fn main() { feed_displays: feed_displays.clone(), session_manager: session_manager.clone(), track_model_store, + game_info: game_info.clone(), }); let api_addr = api_bind_addr(); warnings.info(&format!("REST API server will listen on {}", api_addr)); diff --git a/src/session.rs b/src/session.rs index 465cd3a..f4998ec 100644 --- a/src/session.rs +++ b/src/session.rs @@ -101,6 +101,8 @@ pub struct UseCaseSession { pub recording_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub car_name: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub track_name: Option, } impl UseCaseSession { @@ -117,6 +119,7 @@ impl UseCaseSession { guidance: self.guidance.clone(), recording_id: self.recording_id.clone(), car_name: self.car_name.clone(), + track_name: self.track_name.clone(), } } } @@ -335,12 +338,25 @@ impl SessionManager { guidance, recording_id, car_name: None, + track_name: None, }; self.sessions.insert(id.clone(), session.clone()); self.persist_session(&id); Ok(session) } + /// Update car_name and track_name on a session if they are not yet set. + pub fn fill_session_metadata(&mut self, id: &str, car: Option, track: Option) { + if let Some(session) = self.sessions.get_mut(id) { + if session.car_name.is_none() { + session.car_name = car.filter(|c| !c.is_empty()); + } + if session.track_name.is_none() { + session.track_name = track.filter(|t| !t.is_empty()); + } + } + } + pub fn list_sessions(&mut self) -> Vec { let mut sessions: Vec = self .sessions