// rusty-telemetry — Sim racing telemetry analysis platform // Copyright (C) 2026 Jan-Peter v. Hunnius // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published by // the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. // // This program is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU Affero General Public License for more details. // // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . use crate::telemetry_frame::TelemetryFrame; use std::time::Instant; const RING_CAPACITY: usize = 50; const STALE_THRESHOLD_SECS: f64 = 5.0; const MIN_TIME_SPAN_SECS: f64 = 0.001; #[derive(Clone, Copy, Debug, PartialEq)] pub enum FeedHealth { Waiting, Alive, Stale, Error, } #[derive(Clone, Copy, Debug, PartialEq)] pub enum HandshakePhase { NotStarted, WaitingForResponse, Subscribing, Subscribed, } #[derive(Clone, Copy, Debug, PartialEq)] pub enum FeedMode { BindListen, Handshake { target_port: u16, recv_port: u16, }, Disabled, } #[derive(Clone)] pub struct FeedState { pub health: FeedHealth, pub handshake_phase: HandshakePhase, pub status_text: String, pub total_packets: u64, pub total_bytes: u64, pub frequency_hz: f64, pub data_rate_kb_s: f64, pub last_packet_ago_secs: Option, pub start_time: Option, pub last_packet_time: Option, pub packet_timestamps: [Instant; RING_CAPACITY], pub ring_index: usize, pub ring_count: usize, pub last_frame: Option, } impl FeedState { pub fn new() -> Self { Self { health: FeedHealth::Waiting, handshake_phase: HandshakePhase::NotStarted, status_text: String::from("Bound, waiting for data"), total_packets: 0, total_bytes: 0, frequency_hz: 0.0, data_rate_kb_s: 0.0, last_packet_ago_secs: None, start_time: None, last_packet_time: None, packet_timestamps: [Instant::now(); RING_CAPACITY], ring_index: 0, ring_count: 0, last_frame: None, } } pub fn update_frame(&mut self, frame: TelemetryFrame) { self.last_frame = Some(frame); } pub fn record_packet(&mut self, len: usize) { let now = Instant::now(); if self.start_time.is_none() { self.start_time = Some(now); } self.total_packets += 1; self.total_bytes += len as u64; self.last_packet_time = Some(now); self.last_packet_ago_secs = Some(0.0); if self.ring_count < RING_CAPACITY { self.packet_timestamps[self.ring_count] = now; self.ring_count += 1; self.ring_index = self.ring_count % RING_CAPACITY; } else { self.packet_timestamps[self.ring_index] = now; self.ring_index = (self.ring_index + 1) % RING_CAPACITY; } self.frequency_hz = self.calc_frequency(); self.data_rate_kb_s = self.calc_data_rate(); self.health = FeedHealth::Alive; if self.status_text != "Receiving" { self.status_text = String::from("Receiving"); } } fn calc_frequency(&self) -> f64 { if self.ring_count < 2 { return 0.0; } let count = self.ring_count.min(RING_CAPACITY); let oldest_idx = if self.ring_count >= RING_CAPACITY { self.ring_index } else { 0 }; let newest_idx = if self.ring_count >= RING_CAPACITY { if self.ring_index == 0 { RING_CAPACITY - 1 } else { self.ring_index - 1 } } else { self.ring_count - 1 }; let oldest = self.packet_timestamps[oldest_idx]; let newest = self.packet_timestamps[newest_idx]; let span = newest.duration_since(oldest).as_secs_f64(); if span < MIN_TIME_SPAN_SECS { return 0.0; } count as f64 / span } fn calc_data_rate(&self) -> f64 { match self.start_time { Some(start) => { let elapsed = start.elapsed().as_secs_f64(); if elapsed > 0.0 { (self.total_bytes as f64 / 1024.0) / elapsed } else { 0.0 } } None => 0.0, } } pub fn update_health(&mut self) { match self.last_packet_ago_secs { None => { if self.health != FeedHealth::Error { self.health = FeedHealth::Waiting; self.status_text = String::from("Waiting for data"); } } Some(ago) => { if ago < STALE_THRESHOLD_SECS { self.health = FeedHealth::Alive; self.status_text = String::from("Receiving"); } else { self.health = FeedHealth::Stale; self.status_text = format!("Stale ({:.1}s since last packet)", ago); } } } } pub fn tick(&mut self) { if let Some(last) = self.last_packet_time { self.last_packet_ago_secs = Some(last.elapsed().as_secs_f64()); } self.frequency_hz = self.calc_frequency(); self.data_rate_kb_s = self.calc_data_rate(); self.update_health(); } pub fn set_handshake_phase(&mut self, phase: HandshakePhase) { self.handshake_phase = phase; } pub fn set_error(&mut self, msg: &str) { self.health = FeedHealth::Error; self.status_text = format!("Error: {}", msg); } } impl Default for FeedState { fn default() -> Self { Self::new() } } #[cfg(test)] mod tests { use super::*; use std::thread; #[test] fn test_new_state_is_waiting() { let state = FeedState::new(); assert_eq!(state.health, FeedHealth::Waiting); assert_eq!(state.handshake_phase, HandshakePhase::NotStarted); assert_eq!(state.total_packets, 0); assert_eq!(state.total_bytes, 0); assert_eq!(state.frequency_hz, 0.0); assert!(state.last_packet_ago_secs.is_none()); assert!(state.last_frame.is_none()); } #[test] fn test_record_packet_increments_counters() { let mut state = FeedState::new(); state.record_packet(100); assert_eq!(state.total_packets, 1); assert_eq!(state.total_bytes, 100); assert_eq!(state.health, FeedHealth::Alive); assert_eq!(state.status_text, "Receiving"); state.record_packet(200); assert_eq!(state.total_packets, 2); assert_eq!(state.total_bytes, 300); } #[test] fn test_frequency_zero_with_fewer_than_two_packets() { let mut state = FeedState::new(); state.record_packet(100); assert_eq!(state.frequency_hz, 0.0); // With 2 packets at the same instant, span is < MIN_TIME_SPAN_SECS, so still 0 state.record_packet(100); // frequency could be 0 if the two packets are within MIN_TIME_SPAN_SECS // This is expected behavior } #[test] fn test_frequency_nonzero_after_spaced_packets() { let mut state = FeedState::new(); for _ in 0..10 { state.record_packet(100); thread::sleep(std::time::Duration::from_millis(10)); } // 10 packets over ~90ms → frequency should be > 0 assert!(state.frequency_hz > 0.0); // Rough sanity: 10 packets / 0.09s ≈ 111 Hz, but with overhead, just check > 0 assert!(state.frequency_hz < 10_000.0); // sanity upper bound } #[test] fn test_ring_buffer_fills_without_wraparound() { let mut state = FeedState::new(); for _ in 0..RING_CAPACITY { state.record_packet(50); } assert_eq!(state.ring_count, RING_CAPACITY); assert_eq!(state.ring_index, 0); // wraps to 0 after filling } #[test] fn test_ring_buffer_wraps_around() { let mut state = FeedState::new(); for _ in 0..RING_CAPACITY { state.record_packet(50); } assert_eq!(state.ring_index, 0); // Add more packets to trigger wraparound for _ in 0..10 { state.record_packet(50); } assert_eq!(state.ring_count, RING_CAPACITY); assert_eq!(state.ring_index, 10); assert_eq!(state.total_packets, (RING_CAPACITY + 10) as u64); } #[test] fn test_health_waiting_to_alive() { let mut state = FeedState::new(); assert_eq!(state.health, FeedHealth::Waiting); state.record_packet(100); assert_eq!(state.health, FeedHealth::Alive); } #[test] fn test_health_alive_to_stale() { let mut state = FeedState::new(); state.record_packet(100); assert_eq!(state.health, FeedHealth::Alive); // Simulate time passing by manually setting last_packet_ago_secs state.last_packet_ago_secs = Some(6.0); state.update_health(); assert_eq!(state.health, FeedHealth::Stale); assert!(state.status_text.starts_with("Stale")); } #[test] fn test_health_stale_to_alive() { let mut state = FeedState::new(); state.record_packet(100); state.last_packet_ago_secs = Some(6.0); state.update_health(); assert_eq!(state.health, FeedHealth::Stale); // New packet brings it back to alive state.record_packet(100); assert_eq!(state.health, FeedHealth::Alive); } #[test] fn test_health_error_is_sticky_via_update() { let mut state = FeedState::new(); state.set_error("test error"); assert_eq!(state.health, FeedHealth::Error); // update_health should not override Error when ago is None state.last_packet_ago_secs = None; state.update_health(); assert_eq!(state.health, FeedHealth::Error); } #[test] fn test_tick_updates_ago_secs() { let mut state = FeedState::new(); // Before any packet, tick should not set ago state.tick(); assert!(state.last_packet_ago_secs.is_none()); state.record_packet(100); // Immediately after, ago should be ~0 state.tick(); assert!(state.last_packet_ago_secs.is_some()); assert!(state.last_packet_ago_secs.unwrap() < 1.0); // After sleeping, ago should be > 0 thread::sleep(std::time::Duration::from_millis(100)); state.tick(); let ago = state.last_packet_ago_secs.unwrap(); assert!(ago >= 0.09, "expected ago >= 0.09, got {}", ago); } #[test] fn test_data_rate_calculation() { let mut state = FeedState::new(); state.record_packet(1024); thread::sleep(std::time::Duration::from_millis(50)); state.record_packet(1024); // 2048 bytes over ~50ms → roughly 40 KB/s state.data_rate_kb_s = state.calc_data_rate(); assert!(state.data_rate_kb_s > 0.0); } #[test] fn test_update_frame() { let mut state = FeedState::new(); assert!(state.last_frame.is_none()); let frame = TelemetryFrame::default(); state.update_frame(frame); assert!(state.last_frame.is_some()); } #[test] fn test_set_handshake_phase() { let mut state = FeedState::new(); assert_eq!(state.handshake_phase, HandshakePhase::NotStarted); state.set_handshake_phase(HandshakePhase::WaitingForResponse); assert_eq!(state.handshake_phase, HandshakePhase::WaitingForResponse); state.set_handshake_phase(HandshakePhase::Subscribed); assert_eq!(state.handshake_phase, HandshakePhase::Subscribed); } #[test] fn test_set_error() { let mut state = FeedState::new(); state.set_error("bind failed"); assert_eq!(state.health, FeedHealth::Error); assert_eq!(state.status_text, "Error: bind failed"); } }