refactor: TUI multi-feed monitor with binary parsers
- Add ratatui/crossterm TUI replacing stdout data dump - Add binary parsers for KSUDP RTCarInfo (port 9996) and Telemetry Tool (port 10101) - Add shared binary_utils module for safe packet reading - Add ring-buffer frequency calculation and feed health state machine - Add parse_first_driver() to avoid unnecessary multi-driver allocation - Add 27 unit tests (14 feed_state + 6 ksudp + 7 telemetry_tool) - Move parsing outside mutex lock to minimize TUI contention - Replace expect() on env var with graceful error handling - Remove unused serde dependency - Delete obsolete stats.rs and telemetry.rs
This commit is contained in:
parent
1731495197
commit
7a1c37fc86
11 changed files with 2308 additions and 307 deletions
389
src/feed_state.rs
Normal file
389
src/feed_state.rs
Normal file
|
|
@ -0,0 +1,389 @@
|
|||
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<f64>,
|
||||
pub start_time: Option<Instant>,
|
||||
pub last_packet_time: Option<Instant>,
|
||||
pub packet_timestamps: Vec<Instant>,
|
||||
pub ring_index: usize,
|
||||
pub ring_count: usize,
|
||||
pub last_frame: Option<TelemetryFrame>,
|
||||
}
|
||||
|
||||
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: vec![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 {
|
||||
if self.ring_count < self.packet_timestamps.len() {
|
||||
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");
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue