pitwall 0.1.0

Modern, type-safe Rust library for iRacing telemetry data
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

//! Replay connection for IBT files

use futures::{Stream, StreamExt};
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::watch;
use tokio_stream::wrappers::WatchStream;
use tokio_util::sync::CancellationToken;
use tracing::{debug, info, warn};

use crate::driver::Driver;
use crate::provider::Provider;
use crate::providers::replay::ReplayProvider;
use crate::stream::ThrottleExt;
use crate::types::{FramePacket, UpdateRate};
use crate::{FrameAdapter, Result, SessionInfo, VariableSchema};

/// Replay connection from IBT file
pub struct ReplayConnection {
    /// Frame watch receiver
    frames: watch::Receiver<Option<Arc<FramePacket>>>,

    /// Session watch receiver
    sessions: watch::Receiver<Option<Arc<SessionInfo>>>,

    /// Variable schema
    schema: Arc<VariableSchema>,

    /// Source frequency
    source_hz: f64,

    /// Cancellation token for stopping tasks
    cancel: CancellationToken,
}

impl ReplayConnection {
    /// Open an IBT file for replay.
    ///
    /// Waits for the first frame to be available before returning to ensure
    /// the connection is fully initialized and ready for subscriptions.
    pub async fn open<P: AsRef<Path>>(path: P) -> Result<Self> {
        let path = path.as_ref();
        info!("Opening IBT file: {}", path.display());

        // Create provider and extract metadata
        let provider = ReplayProvider::new(path)?;
        let schema = provider.schema();
        let source_hz = provider.tick_rate();

        // Spawn driver tasks
        let channels = Driver::spawn(provider);

        // Wait for first frame to be available
        let mut frame_rx = channels.frames.clone();
        let timeout = std::time::Duration::from_secs(5);
        let wait_result = tokio::time::timeout(timeout, async {
            loop {
                frame_rx.changed().await.ok();
                if frame_rx.borrow().is_some() {
                    break;
                }
            }
        })
        .await;

        if wait_result.is_err() {
            warn!("Timeout waiting for first frame from replay file");
        }

        info!("Replay connection opened ({}Hz)", source_hz);

        Ok(Self {
            frames: channels.frames,
            sessions: channels.sessions,
            schema,
            source_hz,
            cancel: channels.cancel,
        })
    }

    /// Subscribe to telemetry frames
    pub fn subscribe<T>(&self, rate: UpdateRate) -> impl Stream<Item = T> + 'static
    where
        T: FrameAdapter + Send + 'static,
    {
        // Validate schema once at subscription time
        let validation = T::validate_schema(&self.schema).expect("Schema validation failed");

        // Create base frame stream from watch channel
        let frames = WatchStream::new(self.frames.clone()).filter_map(|opt| async move { opt });

        // Apply rate control and adaptation
        let effective_rate = rate.normalize(self.source_hz);

        match effective_rate {
            UpdateRate::Native => {
                // Direct adaptation, no throttling
                frames.map(move |packet| T::adapt(&packet, &validation)).boxed()
            }
            UpdateRate::Max(hz) => {
                // Throttle then adapt
                let interval = Duration::from_secs_f64(1.0 / hz as f64);
                frames.throttle(interval).map(move |packet| T::adapt(&packet, &validation)).boxed()
            }
        }
    }

    /// Get session updates as a stream
    pub fn session_updates(&self) -> impl Stream<Item = Arc<SessionInfo>> + 'static {
        // Simply watch the session channel - Driver handles all the complexity!
        WatchStream::new(self.sessions.clone()).filter_map(|opt| async move { opt })
    }

    /// Get current session info (if available)
    pub fn current_session(&self) -> Option<Arc<SessionInfo>> {
        self.sessions.borrow().clone()
    }

    /// Get the source telemetry frequency
    pub fn source_hz(&self) -> f64 {
        self.source_hz
    }

    /// Get the variable schema
    pub fn schema(&self) -> &VariableSchema {
        &self.schema
    }
}

impl Drop for ReplayConnection {
    fn drop(&mut self) {
        debug!("Dropping replay connection");
        // Cancel tasks on drop for clean shutdown
        self.cancel.cancel();
    }
}