clankers_data/
inspect.rs

1use std::collections::HashMap;
2use std::fs;
3use std::path::Path;
4
5use chrono::{DateTime, Utc};
6use mcap::MessageStream;
7
8use clankers_core::{RobotError, RobotResult, Timestamp};
9
10/// Summary of an MCAP file's contents.
11#[derive(Debug, Clone, Default)]
12pub struct InspectReport {
13    pub path: String,
14    pub topics: Vec<TopicInfo>,
15    pub start_time: Option<Timestamp>,
16    pub end_time: Option<Timestamp>,
17}
18
19#[derive(Debug, Clone)]
20pub struct TopicInfo {
21    pub name: String,
22    pub message_type: String,
23    pub message_count: u64,
24    pub schema: Option<String>,
25}
26
27/// Read-only handle to an MCAP log file.
28pub struct McapLog {
29    path: std::path::PathBuf,
30    data: Vec<u8>,
31    report: InspectReport,
32}
33
34impl McapLog {
35    pub fn open(path: impl AsRef<Path>) -> RobotResult<Self> {
36        let path = path.as_ref().to_path_buf();
37        let data = fs::read(&path)
38            .map_err(|e| RobotError::Data(format!("read {}: {e}", path.display())))?;
39        let report = inspect_bytes(&path, &data)?;
40        Ok(Self { path, data, report })
41    }
42
43    pub fn report(&self) -> &InspectReport {
44        &self.report
45    }
46
47    pub fn path(&self) -> &Path {
48        &self.path
49    }
50
51    pub fn messages(&self) -> RobotResult<Vec<McapRecord>> {
52        read_all_messages(&self.data)
53    }
54
55    pub fn topic_messages(&self, topic: &str) -> RobotResult<Vec<McapRecord>> {
56        Ok(self
57            .messages()?
58            .into_iter()
59            .filter(|m| m.topic == topic)
60            .collect())
61    }
62}
63
64#[derive(Debug, Clone)]
65pub struct McapRecord {
66    pub topic: String,
67    pub log_time: Timestamp,
68    pub publish_time: Timestamp,
69    pub data: Vec<u8>,
70    pub schema_name: Option<String>,
71}
72
73pub fn inspect_file(path: &Path) -> RobotResult<InspectReport> {
74    let data =
75        fs::read(path).map_err(|e| RobotError::Data(format!("read {}: {e}", path.display())))?;
76    inspect_bytes(path, &data)
77}
78
79fn inspect_bytes(path: &Path, data: &[u8]) -> RobotResult<InspectReport> {
80    if let Ok(Some(summary)) = mcap::Summary::read(data) {
81        let mut topics = Vec::new();
82        let counts = summary
83            .stats
84            .as_ref()
85            .map(|s| s.channel_message_counts.clone())
86            .unwrap_or_default();
87
88        for (channel_id, channel) in &summary.channels {
89            let message_count = counts.get(channel_id).copied().unwrap_or(0);
90            let schema = channel
91                .schema
92                .as_ref()
93                .map(|s| s.name.clone())
94                .unwrap_or_else(|| channel.message_encoding.clone());
95            topics.push(TopicInfo {
96                name: channel.topic.clone(),
97                message_type: channel.message_encoding.clone(),
98                message_count,
99                schema: Some(schema),
100            });
101        }
102        topics.sort_by(|a, b| a.name.cmp(&b.name));
103
104        let (start_time, end_time) = summary
105            .stats
106            .as_ref()
107            .map(|s| {
108                (
109                    Some(Timestamp::from_nanos(s.message_start_time)),
110                    Some(Timestamp::from_nanos(s.message_end_time)),
111                )
112            })
113            .unwrap_or((None, None));
114
115        if !topics.is_empty() {
116            return Ok(InspectReport {
117                path: path.display().to_string(),
118                topics,
119                start_time,
120                end_time,
121            });
122        }
123    }
124
125    // Fallback: scan all messages
126    let mut topic_counts: HashMap<String, u64> = HashMap::new();
127    let mut topic_types: HashMap<String, String> = HashMap::new();
128    let mut topic_schemas: HashMap<String, String> = HashMap::new();
129    let mut start: Option<u64> = None;
130    let mut end: Option<u64> = None;
131
132    for msg in MessageStream::new(data).map_err(|e| RobotError::Data(e.to_string()))? {
133        let msg = msg.map_err(|e| RobotError::Data(e.to_string()))?;
134        let topic = msg.channel.topic.clone();
135        *topic_counts.entry(topic.clone()).or_insert(0) += 1;
136        topic_types
137            .entry(topic.clone())
138            .or_insert_with(|| msg.channel.message_encoding.clone());
139        if let Some(schema) = &msg.channel.schema {
140            topic_schemas
141                .entry(topic)
142                .or_insert_with(|| schema.name.clone());
143        }
144        start = Some(start.map_or(msg.log_time, |s| s.min(msg.log_time)));
145        end = Some(end.map_or(msg.log_time, |e| e.max(msg.log_time)));
146    }
147
148    let topics: Vec<TopicInfo> = topic_counts
149        .into_iter()
150        .map(|(name, message_count)| TopicInfo {
151            message_type: topic_types.get(&name).cloned().unwrap_or_default(),
152            schema: topic_schemas.get(&name).cloned(),
153            name,
154            message_count,
155        })
156        .collect();
157
158    Ok(InspectReport {
159        path: path.display().to_string(),
160        topics,
161        start_time: start.map(Timestamp::from_nanos),
162        end_time: end.map(Timestamp::from_nanos),
163    })
164}
165
166pub fn read_all_messages(data: &[u8]) -> RobotResult<Vec<McapRecord>> {
167    let mut messages = Vec::new();
168    for msg in MessageStream::new(data).map_err(|e| RobotError::Data(e.to_string()))? {
169        let msg = msg.map_err(|e| RobotError::Data(e.to_string()))?;
170        messages.push(McapRecord {
171            topic: msg.channel.topic.clone(),
172            log_time: Timestamp::from_nanos(msg.log_time),
173            publish_time: Timestamp::from_nanos(msg.publish_time),
174            data: msg.data.to_vec(),
175            schema_name: msg.channel.schema.as_ref().map(|s| s.name.clone()),
176        });
177    }
178    messages.sort_by_key(|m| m.log_time.as_nanos());
179    Ok(messages)
180}
181
182pub fn format_datetime(ts: &Timestamp) -> String {
183    ts.to_datetime()
184        .map(|dt: DateTime<Utc>| dt.to_rfc3339_opts(chrono::SecondsFormat::Millis, true))
185        .unwrap_or_else(|| format!("{}ns", ts.as_nanos()))
186}
187
188pub fn format_inspect_report(report: &InspectReport) -> String {
189    let mut out = format!("File: {}\n\nTopics:\n", report.path);
190    for t in &report.topics {
191        let schema = t.schema.as_deref().unwrap_or(t.message_type.as_str());
192        out.push_str(&format!(
193            "  {:<24} {:<24} {:>8} messages\n",
194            t.name, schema, t.message_count
195        ));
196    }
197    out.push_str("\nTime range:\n");
198    match (&report.start_time, &report.end_time) {
199        (Some(s), Some(e)) => {
200            out.push_str(&format!("  start: {}\n", format_datetime(s)));
201            out.push_str(&format!("  end:   {}\n", format_datetime(e)));
202        }
203        _ => out.push_str("  (no messages)\n"),
204    }
205    out
206}
clankers_data/inspect.rs

clankers_data/
inspect.rs
