jan/rusty-telemetry
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: KSUDP parser fix, warnings system, shift point rewrite, code quality (v0.5.1)

Browse source 
Bug Fixes:
- Fixed KSUDP binary parser offsets from packed (323B) to natural C
  alignment (328B) — 3 bytes padding after char identifier, 2 bytes
  after 6 bools
- Fixed setup_terminal() leaving terminal in raw mode on early exit

New Features:
- Thread-safe warning buffer (warnings.rs) with TUI bar + stderr fallback
- Rewrote shift point analysis per AC forum methodology: per-gear
  acceleration curves, crossover detection, same-gear merging
- UseCase::guidance() — step-by-step recording instructions
- Recording metadata enrichment in analysis responses
- recording_id field on sessions

Code Quality:
- Extracted finalize_run() helper (session.rs)
- Extracted execute_and_store_analysis() helper (api.rs)
- Extracted lerp_search<T>() generic helper (analysis.rs)

Breaking API Changes:
- GET /api/use-cases: string[] -> [{id, name, guidance}[]]
- Shift point analysis: gear_ranges/shift_events -> gear_curves/crossover_points

Updated README.md with new API response examples and TUI warning bar.
This commit is contained in:
Jan-Peter von Hunnius 2026-06-05 21:12:57 +02:00
parent 090ab7e559
commit e3aa511ac7
9 changed files with 696 additions and 312 deletions
Download patch file Download diff file Expand all files Collapse all files

50
README.md
View file

@ -119,6 +119,7 @@ The terminal UI shows one panel per active feed:
│ ╚ PCARS Port :5606 ╝ │
├──────────────────────────────────────────────────────────┤
│ Recording: 550e8400-… Ctrl+C Quit │
│ ⚠ Handshake response timed out (5s) — 12s ago │
└──────────────────────────────────────────────────────────┘
```
@ -239,18 +240,52 @@ All fields are optional.
}
```
### Use Cases Response
```json
[
{ "id": "shift_points", "name": "ShiftPoints", "guidance": "1. Drive to a long straight..." },
{ "id": "track_map", "name": "TrackMap", "guidance": "1. Drive a complete lap..." }
]
```
Each object contains:
- `id` — use case identifier (use in session creation)
- `name` — human-readable name
- `guidance` — step-by-step instructions for the use case
### Shift Point Analysis Response
```json
{
"gear_ranges": [
{ "gear": 2, "gear_display": "1", "min_rpm": 3200, "max_rpm": 8500, "avg_rpm": 6100, "sample_count": 1200 },
{ "gear": 3, "gear_display": "2", "min_rpm": 5800, "max_rpm": 8400, "avg_rpm": 7200, "sample_count": 900 }
"gear_curves": [
{
"gear": 2,
"gear_display": "1",
"run_label": "Run 1, Run 2 (2 runs)",
"data_points": [
{ "speed_kmh": 45.2, "rpm": 3200, "acceleration_g": 0.85 },
{ "speed_kmh": 55.8, "rpm": 4100, "acceleration_g": 0.82 }
],
"min_rpm": 3200,
"max_rpm": 8500,
"min_speed_kmh": 45.0,
"max_speed_kmh": 120.0,
"sample_count": 1200
}
],
"shift_events": [
{ "from_gear": 2, "to_gear": 3, "rpm_before_shift": 8400, "rpm_after_shift": 6100, "rpm_drop": 2300, "speed_kmh": 95.0 }
"crossover_points": [
{
"from_gear": 2,
"to_gear": 3,
"from_gear_display": "1",
"to_gear_display": "2",
"crossover_speed_kmh": 95.0,
"crossover_rpm": 7500
}
],
"recommended_shift_rpm": 8400.0,
"recommended_shift_rpm": 7500.0,
"runs_analyzed": 3,
"total_frames_analyzed": 2100
}
```
@ -295,7 +330,8 @@ src/
├── parser_ksudp.rs AC KSUDP binary parser (RTCarInfo, 6 tests)
├── parser_telemetry_tool.rs AC Telemetry Tool binary parser (357-byte driver struct, 7 tests)
├── parser_pcars.rs PCARS 1 & 2 binary parser with auto-detection (5 tests)
└── ui.rs Ratatui TUI rendering
├── ui.rs Ratatui TUI rendering
└── warnings.rs Thread-safe warning buffer with TUI bar + stderr fallback
```
## License

317
src/analysis.rs
View file

@ -14,106 +14,264 @@ use crate::telemetry_frame::{gear_display, TelemetryFrame};
// ---------------------------------------------------------------------------
#[derive(Debug, Clone, Serialize)]
pub struct GearRpmRange {
pub struct AccelerationPoint {
pub speed_kmh: f32,
pub rpm: f32,
pub acceleration_g: f32,
}
#[derive(Debug, Clone, Serialize)]
pub struct GearAccelerationCurve {
pub gear: i32,
pub gear_display: String,
pub run_label: String,
pub data_points: Vec<AccelerationPoint>,
pub min_rpm: f32,
pub max_rpm: f32,
pub avg_rpm: f32,
pub min_speed_kmh: f32,
pub max_speed_kmh: f32,
pub sample_count: usize,
}
#[derive(Debug, Clone, Serialize)]
pub struct ShiftEvent {
pub struct ShiftCrossoverPoint {
pub from_gear: i32,
pub to_gear: i32,
pub rpm_before_shift: f32,
pub rpm_after_shift: f32,
pub rpm_drop: f32,
pub speed_kmh: f32,
pub from_gear_display: String,
pub to_gear_display: String,
pub crossover_speed_kmh: f32,
pub crossover_rpm: f32,
}
#[derive(Debug, Clone, Serialize)]
pub struct ShiftPointAnalysis {
pub gear_ranges: Vec<GearRpmRange>,
pub shift_events: Vec<ShiftEvent>,
pub gear_curves: Vec<GearAccelerationCurve>,
pub crossover_points: Vec<ShiftCrossoverPoint>,
pub recommended_shift_rpm: Option<f32>,
pub runs_analyzed: usize,
pub total_frames_analyzed: usize,
}
pub fn analyze_shift_points(frames: &[TelemetryFrame]) -> ShiftPointAnalysis {
let valid: Vec<&TelemetryFrame> = frames
.iter()
.filter(|f| f.gear >= 2 && f.engine_rpm > 0.0)
.collect();
pub fn analyze_shift_points(
runs: &[Vec<TelemetryFrame>],
run_labels: &[String],
) -> ShiftPointAnalysis {
let mut gear_data: HashMap<i32, Vec<AccelerationPoint>> = HashMap::new();
let mut gear_labels: HashMap<i32, Vec<String>> = HashMap::new();
let mut total_frames_analyzed: usize = 0;
let mut runs_analyzed: usize = 0;
let mut buckets: HashMap<i32, Vec<f32>> = HashMap::new();
for f in &valid {
buckets.entry(f.gear).or_default().push(f.engine_rpm);
}
for (run_idx, run) in runs.iter().enumerate() {
let label = run_labels.get(run_idx).cloned().unwrap_or_default();
let mut gear_ranges: Vec<GearRpmRange> = buckets
.into_iter()
.map(|(gear, rpms)| {
let min_rpm = rpms.iter().cloned().fold(f32::INFINITY, f32::min);
let max_rpm = rpms.iter().cloned().fold(f32::NEG_INFINITY, f32::max);
let avg_rpm = rpms.iter().sum::<f32>() / rpms.len() as f32;
GearRpmRange {
gear,
gear_display: gear_display(gear),
min_rpm,
max_rpm,
avg_rpm,
sample_count: rpms.len(),
let valid: Vec<&TelemetryFrame> = run
.iter()
.filter(|f| f.throttle >= 0.9 && f.gear >= 2 && f.speed_kmh > 1.0 && f.engine_rpm > 500.0)
.collect();
if valid.len() < 10 {
continue;
}
let dominant_gear = {
let mut freq: HashMap<i32, usize> = HashMap::new();
for f in &valid {
*freq.entry(f.gear).or_default() += 1;
}
})
.collect();
gear_ranges.sort_by_key(|g| g.gear);
*freq.iter().max_by_key(|(_, c)| *c).map(|(g, _)| g).unwrap_or(&2)
};
let valid_indices: Vec<usize> = frames
.iter()
.enumerate()
.filter(|(_, f)| f.gear >= 2 && f.engine_rpm > 0.0)
.map(|(i, _)| i)
.collect();
let mut shift_events: Vec<ShiftEvent> = Vec::new();
for window in valid_indices.windows(2) {
let prev = &frames[window[0]];
let curr = &frames[window[1]];
if prev.gear + 1 == curr.gear {
shift_events.push(ShiftEvent {
from_gear: prev.gear,
to_gear: curr.gear,
rpm_before_shift: prev.engine_rpm,
rpm_after_shift: curr.engine_rpm,
rpm_drop: prev.engine_rpm - curr.engine_rpm,
speed_kmh: prev.speed_kmh,
for f in valid.into_iter().filter(|f| f.gear == dominant_gear) {
gear_data.entry(dominant_gear).or_default().push(AccelerationPoint {
speed_kmh: f.speed_kmh,
rpm: f.engine_rpm,
acceleration_g: f.acc_g_frontal,
});
}
gear_labels.entry(dominant_gear).or_default().push(label);
runs_analyzed += 1;
}
let mut gear_curves: Vec<GearAccelerationCurve> = Vec::new();
for (gear, mut points) in gear_data {
total_frames_analyzed += points.len();
points.sort_by(|a, b| a.speed_kmh.partial_cmp(&b.speed_kmh).unwrap_or(std::cmp::Ordering::Equal));
let window = (points.len() / 5).max(3).min(15);
let acc_values: Vec<f32> = points.iter().map(|p| p.acceleration_g).collect();
let smoothed = smooth(&acc_values, window);
for (i, s) in smoothed.into_iter().enumerate() {
points[i].acceleration_g = s;
}
let min_rpm = points.iter().map(|p| p.rpm).fold(f32::INFINITY, f32::min);
let max_rpm = points.iter().map(|p| p.rpm).fold(f32::NEG_INFINITY, f32::max);
let min_speed = points.iter().map(|p| p.speed_kmh).fold(f32::INFINITY, f32::min);
let max_speed = points.iter().map(|p| p.speed_kmh).fold(f32::NEG_INFINITY, f32::max);
let sample_count = points.len();
let labels = gear_labels.get(&gear).cloned().unwrap_or_default();
let run_label = if labels.len() == 1 {
labels[0].clone()
} else {
format!("{} ({} runs)", labels.join(", "), labels.len())
};
gear_curves.push(GearAccelerationCurve {
gear,
gear_display: gear_display(gear),
run_label,
data_points: points,
min_rpm,
max_rpm,
min_speed_kmh: min_speed,
max_speed_kmh: max_speed,
sample_count,
});
}
gear_curves.sort_by_key(|c| c.gear);
let crossover_points = find_crossovers(&gear_curves);
let recommended_shift_rpm = {
let upshift_rpms: Vec<f32> = shift_events
.iter()
.filter(|e| e.from_gear >= 2 && e.to_gear >= 2)
.map(|e| e.rpm_before_shift)
.collect();
if upshift_rpms.is_empty() {
let rpms: Vec<f32> = crossover_points.iter().map(|c| c.crossover_rpm).collect();
if rpms.is_empty() {
None
} else {
Some(upshift_rpms.iter().sum::<f32>() / upshift_rpms.len() as f32)
Some(rpms.iter().sum::<f32>() / rpms.len() as f32)
}
};
ShiftPointAnalysis {
gear_ranges,
shift_events,
gear_curves,
crossover_points,
recommended_shift_rpm,
total_frames_analyzed: valid.len(),
runs_analyzed,
total_frames_analyzed,
}
}
fn smooth(values: &[f32], window: usize) -> Vec<f32> {
let half = window / 2;
let n = values.len();
let mut result = Vec::with_capacity(n);
for i in 0..n {
let start = if i >= half { i - half } else { 0 };
let end = (i + half + 1).min(n);
let sum: f32 = values[start..end].iter().sum();
result.push(sum / (end - start) as f32);
}
result
}
fn find_crossovers(curves: &[GearAccelerationCurve]) -> Vec<ShiftCrossoverPoint> {
let mut crossovers = Vec::new();
for pair in curves.windows(2) {
let low = &pair[0];
let high = &pair[1];
if low.gear == high.gear {
continue;
}
let overlap_min = low.min_speed_kmh.max(high.min_speed_kmh);
let overlap_max = low.max_speed_kmh.min(high.max_speed_kmh);
if overlap_max - overlap_min < 1.0 {
continue;
}
let grid_size = 100;
let step = (overlap_max - overlap_min) / (grid_size as f32 - 1.0);
let mut found_crossover: Option<(f32, f32)> = None;
for i in 0..grid_size {
let target_speed = overlap_min + i as f32 * step;
let low_accel = match interpolate_curve(&low.data_points, target_speed) {
Some(a) => a,
None => continue,
};
let high_accel = match interpolate_curve(&high.data_points, target_speed) {
Some(a) => a,
None => continue,
};
if low_accel < high_accel {
let rpm = interpolate_rpm(&low.data_points, target_speed).unwrap_or(0.0);
found_crossover = Some((target_speed, rpm));
break;
}
}
if let Some((speed, rpm)) = found_crossover {
crossovers.push(ShiftCrossoverPoint {
from_gear: low.gear,
to_gear: high.gear,
from_gear_display: low.gear_display.clone(),
to_gear_display: high.gear_display.clone(),
crossover_speed_kmh: speed,
crossover_rpm: rpm,
});
}
}
crossovers
}
fn interpolate_field<F>(curve: &[AccelerationPoint], target_speed: f32, field: F) -> Option<f32>
where
F: Fn(&AccelerationPoint) -> f32,
{
let (t, a, b) = lerp_search(curve, |p| p.speed_kmh, target_speed)?;
Some(field(a) + t * (field(b) - field(a)))
}
fn interpolate_curve(curve: &[AccelerationPoint], target_speed: f32) -> Option<f32> {
interpolate_field(curve, target_speed, |p| p.acceleration_g)
}
fn interpolate_rpm(curve: &[AccelerationPoint], target_speed: f32) -> Option<f32> {
interpolate_field(curve, target_speed, |p| p.rpm)
}
fn lerp_search<T>(
items: &[T],
key: impl Fn(&T) -> f32,
target: f32,
) -> Option<(f32, &T, &T)> {
if items.is_empty() {
return None;
}
if items.len() == 1 {
return Some((0.0, &items[0], &items[0]));
}
let idx = items.partition_point(|p| key(p) < target);
if idx == 0 {
return Some((0.0, &items[0], &items[0]));
}
if idx >= items.len() {
return Some((0.0, &items[items.len() - 1], &items[items.len() - 1]));
}
let a = &items[idx - 1];
let b = &items[idx];
let t = if (key(b) - key(a)).abs() < f32::EPSILON {
0.0
} else {
(target - key(a)) / (key(b) - key(a))
};
Some((t, a, b))
}
// ---------------------------------------------------------------------------
// Track Map Analysis
// ---------------------------------------------------------------------------
@ -227,32 +385,9 @@ fn compute_center_line(boundaries: &[TrackBoundary]) -> Option<TrackBoundary> {
})
}
/// Linearly interpolate a TrackPoint at the given spline position from a boundary.
fn interpolate_boundary(boundary: &TrackBoundary, target_spline: f32) -> Option<TrackPoint> {
let points = &boundary.points;
if points.is_empty() {
return None;
}
if target_spline <= points.first()?.spline {
return Some(points.first()?.clone());
}
if target_spline >= points.last()?.spline {
return Some(points.last()?.clone());
}
let idx = points.partition_point(|p| p.spline < target_spline);
if idx == 0 || idx >= points.len() {
return points.first().cloned();
}
let a = &points[idx - 1];
let b = &points[idx];
let t = if (b.spline - a.spline).abs() < f32::EPSILON {
0.0
} else {
(target_spline - a.spline) / (b.spline - a.spline)
};
let (t, a, b) = lerp_search(points, |p| p.spline, target_spline)?;
Some(TrackPoint {
x: a.x + t * (b.x - a.x),
@ -273,11 +408,11 @@ pub fn analyze(
) -> serde_json::Value {
match use_case {
UseCase::ShiftPoints => {
if let Some(first_run) = frames_by_run.first() {
serde_json::to_value(analyze_shift_points(first_run))
.unwrap_or_else(|_| serde_json::json!({"error": "serialization failed"}))
} else {
if frames_by_run.is_empty() || frames_by_run.iter().all(|r| r.is_empty()) {
serde_json::json!({"error": "no frames provided"})
} else {
serde_json::to_value(analyze_shift_points(frames_by_run, run_labels))
.unwrap_or_else(|_| serde_json::json!({"error": "serialization failed"}))
}
}
UseCase::TrackMap => serde_json::to_value(analyze_track_map(frames_by_run, run_labels))

87
src/api.rs
View file

@ -17,6 +17,7 @@ use crate::analysis;
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 {
@ -90,8 +91,9 @@ 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) {
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),
}
@ -147,28 +149,62 @@ async fn stop_run(
}
}
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) = {
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)) => (uc, frames, labels),
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"),
}
};
let result = analysis::analyze(&use_case, &frames_by_run, &run_labels);
{
let mut mgr = lock_sessions(&state);
let _ = mgr.set_analysis(&id, result);
}
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) {
@ -197,7 +233,7 @@ async fn run_analysis(
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let (use_case, frames_by_run, run_labels) = {
let (use_case, frames_by_run, run_labels, recording_id) = {
let mgr = lock_sessions(&state);
match mgr.get_session(&id) {
Some(session) => {
@ -207,19 +243,14 @@ async fn run_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(data) => data,
Some((uc, frames, labels)) => (uc, frames, labels, rec_id),
None => return err(StatusCode::NOT_FOUND, "Session not found"),
}
};
let result = analysis::analyze(&use_case, &frames_by_run, &run_labels);
{
let mut mgr = lock_sessions(&state);
let _ = mgr.set_analysis(&id, result.clone());
}
let result = execute_and_store_analysis(&state, &id, use_case, frames_by_run, run_labels, recording_id).await;
ok_json(result)
}
@ -458,8 +489,26 @@ async fn get_live(State(state): State<Arc<AppState>>) -> Response {
})
}
#[derive(Serialize)]
struct UseCaseInfo {
id: String,
name: String,
guidance: String,
}
async fn list_use_cases() -> Response {
ok_json(UseCase::all())
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 {

170
src/main.rs
View file

@ -7,6 +7,7 @@ mod parser_telemetry_tool;
mod recording;
mod telemetry_frame;
mod ui;
mod warnings;
mod analysis;
mod session;
@ -30,6 +31,7 @@ use crate::feed_state::HandshakePhase;
use crate::recording::FEED_DEFS;
use crate::session::SessionManager;
use crate::telemetry_frame::TelemetryFrame;
use crate::warnings::Warnings;
const VERSION: &str = env!("CARGO_PKG_VERSION");
const BUFFER_SIZE: usize = 65535;
@ -63,19 +65,12 @@ fn decode_utf16le(bytes: &[u8]) -> String {
String::from_utf16_lossy(&chars)
}
/// IP address of the machine running the sim game.
/// Used as the KSUDP handshake target (AC binds :9996 on this host).
/// Default: 127.0.0.1 (local — game and listener on same machine).
/// Set to e.g. 192.168.1.2 when the game runs on a different machine.
fn game_host() -> String {
std::env::var("RUSTY_TELEMETRY_GAME_HOST")
.or_else(|_| std::env::var("RUSTY_TELEMETRY_BIND"))
.unwrap_or_else(|_| "127.0.0.1".to_string())
}
/// Local address to bind for incoming UDP feeds (Telemetry Tool, PCARS).
/// Default: 0.0.0.0 (accept from any interface — works for both local and remote).
/// Set to 127.0.0.1 to only accept local packets.
fn listen_addr() -> String {
std::env::var("RUSTY_TELEMETRY_LISTEN")
.or_else(|_| std::env::var("RUSTY_TELEMETRY_BIND"))
@ -118,8 +113,14 @@ fn setup_terminal() -> io::Result<Term> {
let mut stdout = io::stdout();
execute!(stdout, EnterAlternateScreen)?;
let backend = CrosstermBackend::new(stdout);
let terminal = Terminal::new(backend)?;
Ok(terminal)
match Terminal::new(backend) {
Ok(terminal) => Ok(terminal),
Err(e) => {
let _ = execute!(io::stdout(), LeaveAlternateScreen);
let _ = disable_raw_mode();
Err(e)
}
}
}
fn restore_terminal(terminal: &mut Term) -> io::Result<()> {
@ -163,16 +164,17 @@ async fn run_feed(
socket: tokio::net::UdpSocket,
recording_manager: Arc<recording::RecordingManager>,
state: Arc<std::sync::Mutex<FeedState>>,
warnings: Warnings,
mut shutdown_rx: tokio::sync::watch::Receiver<bool>,
) {
let handshake_target: Option<std::net::SocketAddr> = if feed.needs_handshake {
let target: std::net::SocketAddr = match format!("{}:{}", game_host(), feed.port).parse() {
Ok(addr) => addr,
Err(e) => {
eprintln!(
"[ERROR] {} :{} Invalid game host address '{}': {}",
warnings.error(&format!(
"{} :{} Invalid game host address '{}': {}",
feed.label, feed.port, game_host(), e
);
));
if let Ok(mut s) = state.lock() {
s.set_error(&format!("Invalid game host address: {}", e));
}
@ -180,22 +182,20 @@ async fn run_feed(
}
};
// Phase 1: Send HANDSHAKE
eprintln!(
"[INFO] {} :{} Phase 1: Sending HANDSHAKE to {}",
warnings.info(&format!(
"{} :{} Phase 1: Sending HANDSHAKE to {}",
feed.label, feed.port, target
);
));
if let Err(e) = socket.send_to(&HANDSHAKE_REQUEST, target).await {
eprintln!(
"[WARN] HANDSHAKE send failed for :{}: {}",
warnings.warn(&format!(
"HANDSHAKE send failed for :{}: {}",
feed.port, e
);
));
}
if let Ok(mut s) = state.lock() {
s.set_handshake_phase(HandshakePhase::WaitingForResponse);
}
// Phase 2: Wait for 408-byte HandshakerResponse with timeout
let mut hs_buf = [0u8; HANDSHAKE_RESPONSE_SIZE];
match tokio::time::timeout(
Duration::from_secs(5),
@ -215,67 +215,66 @@ async fn run_feed(
let track_name = decode_utf16le(&hs_buf[208..308]);
let track_config = decode_utf16le(&hs_buf[308..408]);
eprintln!(
"[INFO] {} :{} Phase 2: HandshakerResponse received",
warnings.info(&format!(
"{} :{} Phase 2: HandshakerResponse received",
feed.label, feed.port
);
eprintln!(
"[INFO] car_name=\"{}\" driver_name=\"{}\" id={} ver={}",
));
warnings.info(&format!(
" car_name=\"{}\" driver_name=\"{}\" id={} ver={}",
car_name, driver_name, identifier, version
);
eprintln!(
"[INFO] track_name=\"{}\" track_config=\"{}\"",
));
warnings.info(&format!(
" track_name=\"{}\" track_config=\"{}\"",
track_name, track_config
);
));
if identifier != 4242 {
eprintln!(
"[WARN] {} :{} Unexpected handshake identifier: {} (expected 4242)",
warnings.warn(&format!(
"{} :{} Unexpected handshake identifier: {} (expected 4242)",
feed.label, feed.port, identifier
);
));
}
}
Ok(Ok((len, _))) => {
eprintln!(
"[WARN] {} :{} Phase 2: Unexpected response size {} (expected {})",
warnings.warn(&format!(
"{} :{} Phase 2: Unexpected response size {} (expected {})",
feed.label, feed.port, len, HANDSHAKE_RESPONSE_SIZE
);
));
}
Ok(Err(e)) => {
eprintln!(
"[WARN] {} :{} Phase 2: Recv error: {}",
warnings.warn(&format!(
"{} :{} Phase 2: Recv error: {}",
feed.label, feed.port, e
);
));
}
Err(_) => {
eprintln!(
"[WARN] {} :{} Phase 2: Handshake response timed out (5s). AC may not be running.",
warnings.warn(&format!(
"{} :{} Phase 2: Handshake response timed out (5s). AC may not be running.",
feed.label, feed.port
);
));
}
}
// Phase 3: Send SUBSCRIBE_UPDATE
if let Ok(mut s) = state.lock() {
s.set_handshake_phase(HandshakePhase::Subscribing);
}
eprintln!(
"[INFO] {} :{} Phase 3: Sending SUBSCRIBE_UPDATE",
warnings.info(&format!(
"{} :{} Phase 3: Sending SUBSCRIBE_UPDATE",
feed.label, feed.port
);
));
if let Err(e) = socket.send_to(&SUBSCRIBE_UPDATE, target).await {
eprintln!(
"[WARN] SUBSCRIBE_UPDATE send failed for :{}: {}",
warnings.warn(&format!(
"SUBSCRIBE_UPDATE send failed for :{}: {}",
feed.port, e
);
));
}
if let Ok(mut s) = state.lock() {
s.set_handshake_phase(HandshakePhase::Subscribed);
}
eprintln!(
"[INFO] {} :{} Subscribed, entering streaming loop",
warnings.info(&format!(
"{} :{} Subscribed, entering streaming loop",
feed.label, feed.port
);
));
Some(target)
} else {
@ -294,24 +293,19 @@ async fn run_feed(
result = socket.recv_from(&mut buf) => {
match result {
Ok((len, _src)) => {
// Parse telemetry data outside the lock to minimize
// contention with the TUI render thread.
let parsed: Option<TelemetryFrame> = if feed.needs_handshake {
// KSUDP: parse RTCarInfo (skip 408-byte handshake responses)
if len != HANDSHAKE_RESPONSE_SIZE {
parser_ksudp::parse_rt_car_info(&buf[..len]).ok()
} else {
None
}
} else if feed.name == "pcars" {
// PCARS: auto-detect V1/V2 format
match parser_pcars::parse_pcars_packet(&buf[..len]) {
Ok(parser_pcars::PcarsPacket::V1Telemetry(f)) => Some(f),
Ok(parser_pcars::PcarsPacket::V2CarPhysics(f)) => Some(f),
_ => None,
}
} else {
// Telemetry Tool: parse first driver, cache rarely-changing strings
match parser_telemetry_tool::parse_first_driver_no_strings(&buf[..len]) {
Ok((mut frame, driver_id)) => {
if driver_id != string_cache.driver_id {
@ -332,7 +326,6 @@ async fn run_feed(
}
};
// Update state in a single brief lock acquisition.
if let Ok(mut s) = state.lock() {
s.record_packet(len);
if let Some(ref frame) = parsed {
@ -345,10 +338,10 @@ async fn run_feed(
}
}
Err(e) => {
eprintln!(
"[ERROR] Receive error on :{} ({}): {}",
warnings.error(&format!(
"Receive error on :{} ({}): {}",
feed.port, feed.name, e
);
));
}
}
}
@ -360,10 +353,10 @@ async fn run_feed(
} => {
if let Some(target) = handshake_target {
if let Err(e) = socket.send_to(&SUBSCRIBE_UPDATE, target).await {
eprintln!(
"[WARN] Periodic SUBSCRIBE_UPDATE send failed for :{}: {}",
warnings.warn(&format!(
"Periodic SUBSCRIBE_UPDATE send failed for :{}: {}",
feed.port, e
);
));
}
}
}
@ -373,10 +366,10 @@ async fn run_feed(
}
}
eprintln!(
"[INFO] Feed :{} ({}) stopped",
warnings.info(&format!(
"Feed :{} ({}) stopped",
feed.port, feed.name
);
));
}
#[tokio::main]
@ -385,9 +378,10 @@ async fn main() {
let listen = listen_addr();
let data_dir = std::path::PathBuf::from("./data");
let warnings = Warnings::new();
if let Err(e) = tokio::fs::create_dir_all(&data_dir).await {
eprintln!("[ERROR] Cannot create data dir: {}", e);
warnings.error(&format!("Cannot create data dir: {}", e));
std::process::exit(1);
}
@ -410,13 +404,13 @@ async fn main() {
};
match recording_manager.start_recording(auto_start).await {
Ok(m) => {
eprintln!(
"[INFO] Auto-started recording: {} ({})",
warnings.info(&format!(
"Auto-started recording: {} ({})",
m.name, m.id
);
));
}
Err(e) => {
eprintln!("[ERROR] Failed to start recording: {}", e);
warnings.error(&format!("Failed to start recording: {}", e));
std::process::exit(1);
}
}
@ -424,7 +418,7 @@ async fn main() {
let mut terminal = match setup_terminal() {
Ok(t) => t,
Err(e) => {
eprintln!("[ERROR] Failed to setup terminal: {}", e);
warnings.error(&format!("Failed to setup terminal: {}", e));
std::process::exit(1);
}
};
@ -447,10 +441,10 @@ async fn main() {
})
}
Err(e) => {
eprintln!(
"[WARN] Failed to bind handshake recv socket for :{} ({}): {}",
warnings.warn(&format!(
"Failed to bind handshake recv socket for :{} ({}): {}",
feed.port, feed.label, e
);
));
if let Ok(mut s) = state.lock() {
s.set_error(&format!("Bind failed: {}", e));
}
@ -465,10 +459,10 @@ async fn main() {
match tokio::net::UdpSocket::bind(&addr).await {
Ok(s) => (Some(s), FeedMode::BindListen),
Err(e) => {
eprintln!(
"[WARN] Failed to bind :{} ({}): {} -- feed is optional, continuing without it",
warnings.warn(&format!(
"Failed to bind :{} ({}): {} -- feed is optional, continuing without it",
feed.port, feed.label, e
);
));
(None, FeedMode::Disabled)
}
}
@ -486,9 +480,10 @@ async fn main() {
if let Some(socket) = socket_opt {
let feed_clone = feed.clone();
let rm = recording_manager.clone();
let w = warnings.clone();
let sh = shutdown_rx.clone();
feed_tasks.push(tokio::spawn(async move {
run_feed(feed_clone, socket, rm, state, sh).await;
run_feed(feed_clone, socket, rm, state, w, sh).await;
}));
}
}
@ -499,7 +494,7 @@ async fn main() {
session_manager: session_manager.clone(),
});
let api_addr = api_bind_addr();
eprintln!("[INFO] REST API server will listen on {}", api_addr);
warnings.info(&format!("REST API server will listen on {}", api_addr));
tokio::spawn(api::run_server(api_state, api_addr, shutdown_rx.clone()));
let (event_tx, mut event_rx) = tokio::sync::mpsc::channel::<Event>(100);
@ -547,7 +542,7 @@ async fn main() {
Event::Resize(_, _) => {
let session_display = get_session_display();
let _ = terminal.draw(|f| {
render(f, &feed_displays, &session_display, VERSION);
render(f, &feed_displays, &session_display, VERSION, &warnings);
});
}
_ => {}
@ -557,7 +552,7 @@ async fn main() {
_ = render_interval.tick() => {
let session_display = get_session_display();
let _ = terminal.draw(|f| {
render(f, &feed_displays, &session_display, VERSION);
render(f, &feed_displays, &session_display, VERSION, &warnings);
});
}
}
@ -566,9 +561,8 @@ async fn main() {
event_running.store(false, Ordering::Relaxed);
let _ = shutdown_tx.send(true);
// Flush in-memory ring buffers to disk before exit.
if let Err(e) = recording_manager.stop_recording().await {
eprintln!("[WARN] Failed to flush recording on shutdown: {}", e);
warnings.warn(&format!("Failed to flush recording on shutdown: {}", e));
}
match tokio::time::timeout(
@ -580,18 +574,18 @@ async fn main() {
Ok(results) => {
for r in results {
if let Err(e) = r {
eprintln!("[WARN] Task join error: {}", e);
warnings.warn(&format!("Task join error: {}", e));
}
}
}
Err(_) => {
eprintln!("[WARN] Timeout waiting for feeds to stop; forcing exit");
warnings.warn("Timeout waiting for feeds to stop; forcing exit");
}
}
if let Err(e) = restore_terminal(&mut terminal) {
eprintln!("[WARN] Failed to restore terminal: {}", e);
warnings.warn(&format!("Failed to restore terminal: {}", e));
}
eprintln!("[INFO] rusty-telemetry shut down cleanly.");
warnings.info("rusty-telemetry shut down cleanly.");
}

194
src/parser_ksudp.rs
View file

@ -2,13 +2,19 @@
//!
//! After the 3-phase handshake, AC sends continuous RTCarInfo structs at ~20 Hz.
//! This parser converts the raw binary data into a `TelemetryFrame`.
//!
//! IMPORTANT: The AC RTCarInfo C struct uses **natural alignment** (not packed).
//! This means padding bytes exist:
//! - 3 bytes after `char identifier` (offset 0) before `int size` (offset 4)
//! - 2 bytes after the 6 bools (offsets 20–25) before the next float (offset 28)
//! Total struct size: 328 bytes (not 323).
use crate::binary_utils::{read_bool, read_f32, read_f32_4, read_i32};
use crate::telemetry_frame::{FeedSource, TelemetryFrame};
/// Minimum expected packet size (identifier + size field + all known fields).
/// The `size` field in the packet itself reports the true struct size.
const MIN_PACKET_SIZE: usize = 323;
/// Minimum expected packet size for a naturally-aligned RTCarInfo struct.
/// The `size` field in the packet itself reports the true struct size (328).
const MIN_PACKET_SIZE: usize = 328;
/// Parse error type for KSUDP packets.
#[derive(Debug)]
@ -36,6 +42,47 @@ impl std::error::Error for KsudpParseError {}
/// Parse a raw KSUDP RTCarInfo packet into a `TelemetryFrame`.
///
/// Field offsets follow the naturally-aligned C struct layout:
/// - offset 0: char identifier
/// - offset 4: int size (3 bytes padding after char)
/// - offset 8: float speedKmh
/// - offset 12: float speedMph
/// - offset 16: float speedMs
/// - offset 20: bool isAbsEnabled .. isEngineLimiterOn (6 bools at 20-25)
/// - offset 28: float accG_vertical (2 bytes padding after bools)
/// - offset 32: float accG_horizontal
/// - offset 36: float accG_frontal
/// - offset 40: int lapTime
/// - offset 44: int lastLap
/// - offset 48: int bestLap
/// - offset 52: int lapCount
/// - offset 56: float gas
/// - offset 60: float brake
/// - offset 64: float fuel (not clutch — KSUDP has no clutch input)
/// - offset 68: float engineRPM
/// - offset 72: float steer
/// - offset 76: int gear
/// - offset 80: float cgHeight
/// - offset 84: float[4] wheelAngularSpeed
/// - offset 100: float[4] slipAngle
/// - offset 116: float[4] slipAngleContactPatch
/// - offset 132: float[4] slipRatio
/// - offset 148: float[4] tyreSlip
/// - offset 164: float[4] ndSlip
/// - offset 180: float[4] wheelLoad
/// - offset 196: float[4] wheelDy
/// - offset 212: float[4] wheelMz
/// - offset 228: float[4] tyreDirtyLevel
/// - offset 244: float[4] camberRAD
/// - offset 260: float[4] tyreRadius
/// - offset 276: float[4] tyreLoadedRadius
/// - offset 292: float[4] suspensionHeight
/// - offset 308: float carPositionNormalized
/// - offset 312: float carSlope
/// - offset 316: float worldX
/// - offset 320: float worldY
/// - offset 324: float worldZ
///
/// # Errors
/// Returns `KsudpParseError` if the packet is too short or has an invalid identifier.
pub fn parse_rt_car_info(data: &[u8]) -> Result<TelemetryFrame, KsudpParseError> {
@ -46,15 +93,13 @@ pub fn parse_rt_car_info(data: &[u8]) -> Result<TelemetryFrame, KsudpParseError>
});
}
// Validate identifier
let identifier = data[0];
if identifier != b'a' {
return Err(KsudpParseError::InvalidIdentifier { got: identifier });
}
let reported_size = read_i32(data, 1) as usize;
let reported_size = read_i32(data, 4) as usize;
if data.len() < reported_size && reported_size > 0 {
// Log warning but don't fail — the struct might have fields we don't read yet
eprintln!(
"[WARN] KSUDP: reported struct size {} but packet is {} bytes",
reported_size,
@ -66,73 +111,60 @@ pub fn parse_rt_car_info(data: &[u8]) -> Result<TelemetryFrame, KsudpParseError>
source: FeedSource::Ksudp,
raw_size: data.len(),
// Speed
speed_kmh: read_f32(data, 5),
speed_mph: read_f32(data, 9),
speed_ms: read_f32(data, 13),
speed_kmh: read_f32(data, 8),
speed_mph: read_f32(data, 12),
speed_ms: read_f32(data, 16),
// Flags
is_abs_enabled: read_bool(data, 17),
is_abs_in_action: read_bool(data, 18),
is_tc_in_action: read_bool(data, 19),
is_tc_enabled: read_bool(data, 20),
is_in_pit: read_bool(data, 21),
is_engine_limiter_on: read_bool(data, 22),
is_abs_enabled: read_bool(data, 20),
is_abs_in_action: read_bool(data, 21),
is_tc_in_action: read_bool(data, 22),
is_tc_enabled: read_bool(data, 23),
is_in_pit: read_bool(data, 24),
is_engine_limiter_on: read_bool(data, 25),
// G-forces
acc_g_vertical: read_f32(data, 23),
acc_g_horizontal: read_f32(data, 27),
acc_g_frontal: read_f32(data, 31),
acc_g_vertical: read_f32(data, 28),
acc_g_horizontal: read_f32(data, 32),
acc_g_frontal: read_f32(data, 36),
// Lap timing
lap_time_ms: read_i32(data, 35),
last_lap_ms: read_i32(data, 39),
best_lap_ms: read_i32(data, 43),
lap_count: read_i32(data, 47),
lap_time_ms: read_i32(data, 40),
last_lap_ms: read_i32(data, 44),
best_lap_ms: read_i32(data, 48),
lap_count: read_i32(data, 52),
// Driver inputs
throttle: read_f32(data, 51),
brake: read_f32(data, 55),
clutch: 0.0, // offset 59 is fuel (0.0–1.0), not clutch; KSUDP has no clutch input
engine_rpm: read_f32(data, 63),
steer: read_f32(data, 67),
throttle: read_f32(data, 56),
brake: read_f32(data, 60),
clutch: 0.0, // offset 64 is fuel level (0.0-1.0), not clutch; KSUDP has no clutch input
engine_rpm: read_f32(data, 68),
steer: read_f32(data, 72),
// Gear & chassis
gear: read_i32(data, 71),
cg_height: read_f32(data, 75),
gear: read_i32(data, 76),
cg_height: read_f32(data, 80),
// Per-wheel arrays
wheel_angular_speed: read_f32_4(data, 79),
slip_angle: read_f32_4(data, 95),
slip_angle_contact_patch: read_f32_4(data, 111),
slip_ratio: read_f32_4(data, 127),
tyre_slip: read_f32_4(data, 143),
nd_slip: read_f32_4(data, 159),
wheel_load: read_f32_4(data, 175),
wheel_dy: read_f32_4(data, 191),
wheel_mz: read_f32_4(data, 207),
tyre_dirty_level: read_f32_4(data, 223),
camber_rad: read_f32_4(data, 239),
tyre_radius: read_f32_4(data, 255),
tyre_loaded_radius: read_f32_4(data, 271),
suspension_height: read_f32_4(data, 287),
wheel_angular_speed: read_f32_4(data, 84),
slip_angle: read_f32_4(data, 100),
slip_angle_contact_patch: read_f32_4(data, 116),
slip_ratio: read_f32_4(data, 132),
tyre_slip: read_f32_4(data, 148),
nd_slip: read_f32_4(data, 164),
wheel_load: read_f32_4(data, 180),
wheel_dy: read_f32_4(data, 196),
wheel_mz: read_f32_4(data, 212),
tyre_dirty_level: read_f32_4(data, 228),
camber_rad: read_f32_4(data, 244),
tyre_radius: read_f32_4(data, 260),
tyre_loaded_radius: read_f32_4(data, 276),
suspension_height: read_f32_4(data, 292),
// Position
// NOTE: carPosition (offset 303) is the normalized spline position (0.0–1.0).
// Both car_position_normalized and spline_position map to the same AC field.
car_position_normalized: read_f32(data, 303),
car_slope: read_f32(data, 307),
world_x: read_f32(data, 311),
world_y: read_f32(data, 315),
world_z: read_f32(data, 319),
spline_position: read_f32(data, 303), // same as car_position_normalized (AC SDK: carPosition)
car_position_normalized: read_f32(data, 308),
car_slope: read_f32(data, 312),
world_x: read_f32(data, 316),
world_y: read_f32(data, 320),
world_z: read_f32(data, 324),
spline_position: read_f32(data, 308),
..Default::default()
};
// KSUDP-specific: mark Telemetry Tool fields as None (already default)
// Keep KSUDP-specific fields populated above.
Ok(frame)
}
@ -143,17 +175,12 @@ mod tests {
#[test]
fn test_parse_minimal_valid_packet() {
let mut packet = vec![0u8; MIN_PACKET_SIZE];
packet[0] = b'a'; // identifier
// size = MIN_PACKET_SIZE
packet[1..5].copy_from_slice(&(MIN_PACKET_SIZE as i32).to_le_bytes());
// speed_Kmh = 123.4
packet[5..9].copy_from_slice(&123.4f32.to_le_bytes());
// gear = 3 (meaning gear 2)
packet[71..75].copy_from_slice(&3i32.to_le_bytes());
// engineRPM = 7500.0
packet[63..67].copy_from_slice(&7500.0f32.to_le_bytes());
// lapTime = 78123 ms (1:18.123)
packet[35..39].copy_from_slice(&78123i32.to_le_bytes());
packet[0] = b'a';
packet[4..8].copy_from_slice(&(MIN_PACKET_SIZE as i32).to_le_bytes());
packet[8..12].copy_from_slice(&123.4f32.to_le_bytes());
packet[76..80].copy_from_slice(&3i32.to_le_bytes());
packet[68..72].copy_from_slice(&7500.0f32.to_le_bytes());
packet[40..44].copy_from_slice(&78123i32.to_le_bytes());
let frame = parse_rt_car_info(&packet).unwrap();
assert_eq!(frame.source, FeedSource::Ksudp);
@ -192,12 +219,12 @@ mod tests {
fn test_boolean_flags() {
let mut packet = vec![0u8; MIN_PACKET_SIZE];
packet[0] = b'a';
packet[17] = 1; // isAbsEnabled = true
packet[18] = 0; // isAbsInAction = false
packet[19] = 1; // isTcInAction = true
packet[20] = 1; // isTcEnabled = true
packet[21] = 0; // isInPit = false
packet[22] = 1; // isEngineLimiterOn = true
packet[20] = 1;
packet[21] = 0;
packet[22] = 1;
packet[23] = 1;
packet[24] = 0;
packet[25] = 1;
let frame = parse_rt_car_info(&packet).unwrap();
assert!(frame.is_abs_enabled);
@ -212,10 +239,9 @@ mod tests {
fn test_per_wheel_arrays() {
let mut packet = vec![0u8; MIN_PACKET_SIZE];
packet[0] = b'a';
// slipRatio at offset 127: [0.1, 0.2, 0.3, 0.4]
for i in 0..4 {
let val = (i as f32 + 1.0) * 0.1;
packet[127 + i * 4..127 + i * 4 + 4].copy_from_slice(&val.to_le_bytes());
packet[132 + i * 4..132 + i * 4 + 4].copy_from_slice(&val.to_le_bytes());
}
let frame = parse_rt_car_info(&packet).unwrap();
@ -229,9 +255,9 @@ mod tests {
fn test_world_coordinates() {
let mut packet = vec![0u8; MIN_PACKET_SIZE];
packet[0] = b'a';
packet[311..315].copy_from_slice(&100.5f32.to_le_bytes()); // x
packet[315..319].copy_from_slice(&2.3f32.to_le_bytes()); // y
packet[319..323].copy_from_slice(&(-50.7f32).to_le_bytes()); // z
packet[316..320].copy_from_slice(&100.5f32.to_le_bytes());
packet[320..324].copy_from_slice(&2.3f32.to_le_bytes());
packet[324..328].copy_from_slice(&(-50.7f32).to_le_bytes());
let frame = parse_rt_car_info(&packet).unwrap();
assert!((frame.world_x - 100.5).abs() < 0.01);

11
src/recording.rs
View file

@ -148,6 +148,17 @@ impl UseCase {
UseCase::General,
]
}
pub fn guidance(&self) -> &'static str {
match self {
UseCase::ShiftPoints => "Record one run per gear: accelerate at full throttle from as low an RPM as possible to the rev limiter (or until you run out of road). Each run should be in a single gear only. Use Hotlap mode on a long straight. Start recording a run just before going WOT, stop after reaching max RPM. Repeat for each gear.",
UseCase::TrackMap => "Record two runs on the same lap: one driving the left boundary, one driving the right boundary. Label each run accordingly.",
UseCase::BrakingAnalysis => "Record runs with heavy braking into corners from high speed.",
UseCase::LapTimes => "Record one or more complete laps.",
UseCase::SectorAnalysis => "Record one or more complete laps.",
UseCase::RacingLine => "Record multiple laps with different racing line approaches.",
UseCase::General => "Record freely - no specific procedure required.",
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]

36
src/session.rs
View file

@ -66,6 +66,9 @@ pub struct UseCaseSession {
pub completed_at: Option<String>,
pub runs: Vec<SessionRun>,
pub analysis: Option<serde_json::Value>,
#[serde(default)]
pub guidance: String,
pub recording_id: Option<String>,
}
impl UseCaseSession {
@ -79,6 +82,8 @@ impl UseCaseSession {
completed_at: self.completed_at.clone(),
runs: self.runs.iter().map(|r| r.summary_clone()).collect(),
analysis: self.analysis.clone(),
guidance: self.guidance.clone(),
recording_id: self.recording_id.clone(),
}
}
}
@ -100,9 +105,11 @@ impl SessionManager {
&mut self,
use_case: UseCase,
label: Option<String>,
recording_id: Option<String>,
) -> Result<UseCaseSession, String> {
let id = uuid::Uuid::new_v4().to_string();
let now = chrono::Utc::now().to_rfc3339();
let guidance = use_case.guidance().to_string();
let session = UseCaseSession {
id: id.clone(),
label: label.unwrap_or_else(|| format!("{:?} Session", use_case)),
@ -112,6 +119,8 @@ impl SessionManager {
completed_at: None,
runs: Vec::new(),
analysis: None,
guidance,
recording_id,
};
self.sessions.insert(id, session.clone());
Ok(session)
@ -184,9 +193,14 @@ impl SessionManager {
return Err(format!("Run '{}' is not recording", run_id));
}
let end_marker = self.recording_manager.ring_marker(&run.feed_name);
let frames = self
.recording_manager
Self::finalize_run(&self.recording_manager, run);
Ok(run.summary_clone())
}
fn finalize_run(recording_manager: &Arc<RecordingManager>, run: &mut SessionRun) {
let end_marker = recording_manager.ring_marker(&run.feed_name);
let frames = recording_manager
.extract_since(&run.feed_name, run.start_marker);
let expected = end_marker.saturating_sub(run.start_marker);
@ -197,8 +211,6 @@ impl SessionManager {
run.frames = frames;
run.status = RunStatus::Stopped;
run.data_truncated = data_truncated;
Ok(run.summary_clone())
}
pub fn complete_session(&mut self, id: &str) -> Result<UseCaseSession, String> {
@ -213,19 +225,7 @@ impl SessionManager {
for run in &mut session.runs {
if run.status == RunStatus::Recording {
let end_marker = self.recording_manager.ring_marker(&run.feed_name);
let frames = self
.recording_manager
.extract_since(&run.feed_name, run.start_marker);
let expected = end_marker.saturating_sub(run.start_marker);
let data_truncated = (frames.len() as u64) < expected && expected > 0;
run.end_marker = Some(end_marker);
run.frame_count = frames.len();
run.frames = frames;
run.status = RunStatus::Stopped;
run.data_truncated = data_truncated;
Self::finalize_run(&self.recording_manager, run);
}
}

54
src/ui.rs
View file

@ -1,3 +1,5 @@
use std::sync::{Arc, Mutex};
use ratatui::{
Frame,
layout::{Constraint, Direction, Layout, Rect},
@ -8,7 +10,7 @@ use ratatui::{
use crate::feed_state::{FeedHealth, FeedMode, FeedState, HandshakePhase};
use crate::telemetry_frame::TelemetryFrame;
use std::sync::{Arc, Mutex};
use crate::warnings::{WarningLevel, Warnings};
#[derive(Clone)]
pub struct FeedDisplayInfo {
@ -144,11 +146,52 @@ fn render_feed_panel(frame: &mut Frame, area: Rect, info: &FeedDisplayInfo, stat
frame.render_widget(paragraph, inner);
}
pub fn render_warning_bar(frame: &mut Frame, area: Rect, warnings: &Warnings) {
let latest = warnings.latest();
let recent = latest.filter(|e| e.timestamp.elapsed().as_secs() < 30);
let border_color = match &recent {
Some(e) => match e.level {
WarningLevel::Error => Color::Red,
WarningLevel::Warn => Color::Yellow,
WarningLevel::Info => Color::Cyan,
},
None => Color::DarkGray,
};
let block = Block::default()
.title(" Status ")
.borders(Borders::ALL)
.border_style(Style::default().fg(border_color));
let inner = block.inner(area);
frame.render_widget(block, area);
if let Some(e) = recent {
let prefix = match e.level {
WarningLevel::Error => "[ERR]",
WarningLevel::Warn => "[WARN]",
WarningLevel::Info => "[INFO]",
};
let prefix_style = match e.level {
WarningLevel::Error => Style::default().fg(Color::Red).add_modifier(Modifier::BOLD),
WarningLevel::Warn => Style::default().fg(Color::Yellow).add_modifier(Modifier::BOLD),
WarningLevel::Info => Style::default().fg(Color::Cyan),
};
let text = Line::from(vec![
Span::styled(format!(" {} ", prefix), prefix_style),
Span::raw(&e.message),
]);
let paragraph = Paragraph::new(text);
frame.render_widget(paragraph, inner);
}
}
pub fn render(
frame: &mut Frame,
feeds: &[(FeedDisplayInfo, Arc<Mutex<FeedState>>)],
session_dir: &str,
version: &str,
warnings: &Warnings,
) {
let total_feeds = feeds.len() as u16;
@ -158,6 +201,7 @@ pub fn render(
Constraint::Length(3),
Constraint::Min(total_feeds * 4 + 2),
Constraint::Length(3),
Constraint::Length(3),
])
.split(frame.area());
@ -189,8 +233,6 @@ pub fn render(
.split(outer[1]);
for (i, (info, state)) in feeds.iter().enumerate() {
// Clone inside the lock, then drop the guard before rendering
// to avoid blocking the UDP receive task during the entire draw.
let snapshot = state.lock().ok().map(|mut s| {
s.tick();
s.clone()
@ -200,11 +242,13 @@ pub fn render(
}
}
render_warning_bar(frame, outer[2], warnings);
let footer_block = Block::default()
.borders(Borders::ALL)
.border_style(Style::default().fg(Color::DarkGray));
let footer_inner = footer_block.inner(outer[2]);
frame.render_widget(footer_block, outer[2]);
let footer_inner = footer_block.inner(outer[3]);
frame.render_widget(footer_block, outer[3]);
let footer_text = Line::from(vec![
Span::styled(

89
src/warnings.rs Normal file
View file

@ -0,0 +1,89 @@
use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use std::time::Instant;
const MAX_WARNINGS: usize = 50;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WarningLevel {
Info,
Warn,
Error,
}
#[derive(Debug, Clone)]
pub struct WarningEntry {
pub message: String,
pub timestamp: Instant,
pub level: WarningLevel,
}
#[derive(Clone)]
pub struct Warnings {
entries: Arc<Mutex<VecDeque<WarningEntry>>>,
}
impl Warnings {
pub fn new() -> Self {
Self {
entries: Arc::new(Mutex::new(VecDeque::with_capacity(MAX_WARNINGS))),
}
}
fn push(&self, level: WarningLevel, msg: &str) {
match level {
WarningLevel::Error => eprintln!("[ERROR] {}", msg),
WarningLevel::Warn => eprintln!("[WARN] {}", msg),
WarningLevel::Info => {}
}
let mut guard = match self.entries.lock() {
Ok(g) => g,
Err(e) => e.into_inner(),
};
if guard.len() >= MAX_WARNINGS {
guard.pop_front();
}
guard.push_back(WarningEntry {
message: msg.to_string(),
timestamp: Instant::now(),
level,
});
}
pub fn warn(&self, msg: &str) {
self.push(WarningLevel::Warn, msg);
}
pub fn info(&self, msg: &str) {
self.push(WarningLevel::Info, msg);
}
pub fn error(&self, msg: &str) {
self.push(WarningLevel::Error, msg);
}
/// Returns warnings from the last `max_age_secs` seconds.
/// Currently unused but kept for potential future TUI scrolling.
#[allow(dead_code)]
pub fn recent(&self, max_age_secs: u64) -> Vec<WarningEntry> {
let guard = match self.entries.lock() {
Ok(g) => g,
Err(e) => e.into_inner(),
};
let cutoff = Instant::now() - std::time::Duration::from_secs(max_age_secs);
guard
.iter()
.filter(|e| e.timestamp >= cutoff)
.cloned()
.collect()
}
pub fn latest(&self) -> Option<WarningEntry> {
let guard = match self.entries.lock() {
Ok(g) => g,
Err(e) => e.into_inner(),
};
guard.back().cloned()
}
}