// 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");
}
}