eventcv_core/io/
prophesee.rs

1//! Prophesee `.dat` CD (change-detection) event reader. The file is an ASCII comment
2//! header (lines starting with `%`), then two bytes giving the event type and event size,
3//! then little-endian 8-byte records: a 32-bit microsecond timestamp and a 32-bit packed
4//! word `x = bits 0..13`, `y = bits 14..27`, `polarity = bit 28`.
5//!
6//! Implemented from the published `.dat` layout (used by Prophesee/Metavision and Tonic).
7//! It has not yet been checked against a real `.dat` file — the unit tests cover the byte
8//! layout exactly. EVT3 `.raw` recordings are handled by the sibling `prophesee_raw` module.
9
10use std::fs::File;
11use std::io::{BufRead, BufReader, BufWriter, ErrorKind, Write};
12use std::path::Path;
13
14use super::{EventSource, IoError, LoadOptions, RawEvent};
15use crate::{EventStream, EventStreamBuilder};
16
17/// Prophesee timestamps tick in microseconds.
18const TIMESTAMP_SCALE_MS: f64 = 0.001;
19
20/// Only the 8-byte CD event layout is implemented.
21const CD_EVENT_SIZE: u8 = 8;
22
23/// Reads the ASCII comment header (lines starting with `%`), leaving `reader` positioned at
24/// the event-type/size bytes.
25fn read_header<R: BufRead>(reader: &mut R) -> Result<Vec<String>, IoError> {
26    let mut lines = Vec::new();
27    loop {
28        let is_comment = match reader.fill_buf()? {
29            [] => break,
30            buf => buf[0] == b'%',
31        };
32        if !is_comment {
33            break;
34        }
35        let mut line = Vec::new();
36        reader.read_until(b'\n', &mut line)?;
37        lines.push(String::from_utf8_lossy(&line).trim_end().to_owned());
38    }
39    Ok(lines)
40}
41
42/// Pulls the sensor size from `% Width`/`% Height` header lines, or an explicit override.
43fn resolve_sensor(
44    header: &[String],
45    sensor: Option<(usize, usize)>,
46) -> Result<(usize, usize), IoError> {
47    if let Some(size) = sensor {
48        return Ok(size);
49    }
50    let (mut width, mut height) = (None, None);
51    for line in header {
52        let body = line.trim_start_matches('%').trim().to_ascii_lowercase();
53        let value = |rest: &str| rest.split_whitespace().next().and_then(|v| v.parse().ok());
54        if let Some(rest) = body.strip_prefix("width") {
55            width = value(rest);
56        } else if let Some(rest) = body.strip_prefix("height") {
57            height = value(rest);
58        }
59    }
60    match (width, height) {
61        (Some(width), Some(height)) => Ok((width, height)),
62        _ => Err(IoError::Unsupported(
63            "could not determine sensor size from the Prophesee .dat header; pass sensor_size"
64                .to_owned(),
65        )),
66    }
67}
68
69struct DatSource<R: BufRead> {
70    reader: R,
71    width: usize,
72    height: usize,
73}
74
75impl<R: BufRead> EventSource for DatSource<R> {
76    fn sensor_size(&self) -> (usize, usize) {
77        (self.width, self.height)
78    }
79
80    fn timestamp_scale_ms(&self) -> f64 {
81        TIMESTAMP_SCALE_MS
82    }
83
84    fn next_event(&mut self) -> Result<Option<RawEvent>, IoError> {
85        let mut record = [0u8; 8];
86        match self.reader.read_exact(&mut record) {
87            Ok(()) => {}
88            Err(error) if error.kind() == ErrorKind::UnexpectedEof => return Ok(None),
89            Err(error) => return Err(IoError::Io(error)),
90        }
91        let timestamp = u32::from_le_bytes([record[0], record[1], record[2], record[3]]);
92        let data = u32::from_le_bytes([record[4], record[5], record[6], record[7]]);
93        Ok(Some(RawEvent {
94            x: (data & 0x3FFF) as u16,
95            y: ((data >> 14) & 0x3FFF) as u16,
96            t: i64::from(timestamp),
97            p: (data >> 28) & 1 != 0,
98        }))
99    }
100}
101
102fn read_dat_from<R: BufRead>(mut reader: R, options: &LoadOptions) -> Result<EventStream, IoError> {
103    let header = read_header(&mut reader)?;
104    let (width, height) = resolve_sensor(&header, options.sensor_size)?;
105
106    let mut type_size = [0u8; 2];
107    match reader.read_exact(&mut type_size) {
108        Ok(()) => {
109            let event_size = type_size[1];
110            if event_size != CD_EVENT_SIZE {
111                return Err(IoError::Unsupported(format!(
112                    "unsupported Prophesee .dat event size {event_size} (only 8-byte CD events are implemented)"
113                )));
114            }
115        }
116        // Header with no events: a valid, empty recording.
117        Err(error) if error.kind() == ErrorKind::UnexpectedEof => {
118            return Ok(EventStreamBuilder::new(width, height, TIMESTAMP_SCALE_MS).build());
119        }
120        Err(error) => return Err(IoError::Io(error)),
121    }
122
123    super::read_capped(
124        DatSource {
125            reader,
126            width,
127            height,
128        },
129        options.max_events,
130    )
131}
132
133/// Reads a Prophesee `.dat` recording. `sensor_size` overrides the header geometry; the
134/// timestamp unit is always microseconds. `max_events` caps how many events are kept.
135pub fn read_dat(path: impl AsRef<Path>, options: &LoadOptions) -> Result<EventStream, IoError> {
136    read_dat_from(BufReader::new(File::open(path)?), options)
137}
138
139/// Writes a Prophesee `.dat` CD recording a window at a time — the inverse of [`read_dat`].
140///
141/// The `%` header and the event-type/size pair are written on the first non-empty append, because
142/// the header carries `% Width`/`% Height` and those come from the first stream's sensor size.
143/// After that every append is a run of 8-byte records with nothing to fix up, so the file is
144/// readable at any point.
145///
146/// Timestamps are `u32` microseconds, which is the format's own width: it wraps after ~71 minutes
147/// and the format offers no high half, so a recording longer than that is rejected rather than
148/// silently wrapped.
149pub struct DatEventSink {
150    writer: BufWriter<File>,
151    header_written: bool,
152    n_events: usize,
153}
154
155impl DatEventSink {
156    pub fn create(path: impl AsRef<Path>) -> Result<Self, IoError> {
157        Ok(Self {
158            writer: BufWriter::new(File::create(path).map_err(IoError::Io)?),
159            header_written: false,
160            n_events: 0,
161        })
162    }
163}
164
165/// Packs `x`, `y` and polarity into a `.dat` CD word, the inverse of the unpacking in
166/// [`DatSource::next_event`].
167fn cd_word(x: u16, y: u16, polarity: bool) -> u32 {
168    (u32::from(x) & 0x3FFF) | ((u32::from(y) & 0x3FFF) << 14) | ((polarity as u32) << 28)
169}
170
171impl super::EventSink for DatEventSink {
172    fn append(&mut self, stream: &EventStream) -> Result<(), IoError> {
173        if stream.is_empty() {
174            return Ok(());
175        }
176        if !self.header_written {
177            let (width, height) = stream.sensor_size();
178            write!(
179                self.writer,
180                "% Date 1970-01-01 00:00:00\n% Version 2\n% Width {width}\n% Height {height}\n"
181            )
182            .map_err(IoError::Io)?;
183            // Event type 0x0C (CD_LOW/CD event) and the 8-byte record size the reader requires.
184            self.writer
185                .write_all(&[0x0C, CD_EVENT_SIZE])
186                .map_err(IoError::Io)?;
187            self.header_written = true;
188        }
189        let (xs, ys, ts, ps) = (stream.xs(), stream.ys(), stream.ts(), stream.ps());
190        for index in 0..stream.len() {
191            let timestamp = u32::try_from(ts[index]).map_err(|_| {
192                IoError::Unsupported(format!(
193                    "Prophesee .dat timestamps are 32-bit microseconds, so {} does not fit; the \
194                     format cannot hold a recording longer than ~71 minutes or one starting at a \
195                     Unix epoch — subtract the start time first (stream.time_shift(-t0))",
196                    ts[index]
197                ))
198            })?;
199            self.writer
200                .write_all(&timestamp.to_le_bytes())
201                .map_err(IoError::Io)?;
202            self.writer
203                .write_all(&cd_word(xs[index], ys[index], ps[index]).to_le_bytes())
204                .map_err(IoError::Io)?;
205        }
206        self.n_events += stream.len();
207        Ok(())
208    }
209
210    fn n_events(&self) -> usize {
211        self.n_events
212    }
213
214    fn flush(&mut self) -> Result<(), IoError> {
215        self.writer.flush().map_err(IoError::Io)
216    }
217
218    fn finish(mut self: Box<Self>) -> Result<(), IoError> {
219        self.writer.flush().map_err(IoError::Io)
220    }
221}
222
223#[cfg(test)]
224mod tests {
225    use std::io::Cursor;
226
227    use super::*;
228
229    fn cd_word(x: u32, y: u32, polarity: bool) -> u32 {
230        (x & 0x3FFF) | ((y & 0x3FFF) << 14) | ((polarity as u32) << 28)
231    }
232
233    fn record(timestamp: u32, data: u32) -> [u8; 8] {
234        let mut bytes = [0u8; 8];
235        bytes[0..4].copy_from_slice(&timestamp.to_le_bytes());
236        bytes[4..8].copy_from_slice(&data.to_le_bytes());
237        bytes
238    }
239
240    fn file_with(records: &[[u8; 8]]) -> Vec<u8> {
241        let mut data = b"% Date 2024-01-01\n% Width 640\n% Height 480\n% Version 2\n".to_vec();
242        data.extend_from_slice(&[0x00, CD_EVENT_SIZE]); // event type + size
243        for record in records {
244            data.extend_from_slice(record);
245        }
246        data
247    }
248
249    fn read(data: &[u8], options: &LoadOptions) -> Result<EventStream, IoError> {
250        read_dat_from(Cursor::new(data.to_vec()), options)
251    }
252
253    #[test]
254    fn decodes_cd_events() {
255        let data = file_with(&[
256            record(1_000, cd_word(10, 20, true)),
257            record(1_005, cd_word(600, 400, false)),
258        ]);
259        let stream = read(&data, &LoadOptions::default()).unwrap();
260
261        assert_eq!(stream.sensor_size(), (640, 480));
262        assert_eq!(stream.len(), 2);
263        assert_eq!(stream.xs(), &[10, 600]);
264        assert_eq!(stream.ys(), &[20, 400]);
265        assert_eq!(stream.ts(), &[1_000, 1_005]);
266        assert_eq!(stream.ps(), &[true, false]);
267    }
268
269    #[test]
270    fn header_only_file_is_empty() {
271        let data = b"% Width 640\n% Height 480\n".to_vec();
272        let stream = read(&data, &LoadOptions::default()).unwrap();
273        assert!(stream.is_empty());
274        assert_eq!(stream.sensor_size(), (640, 480));
275    }
276
277    #[test]
278    fn out_of_bounds_events_are_dropped() {
279        let data = file_with(&[
280            record(1, cd_word(1, 1, true)),
281            record(2, cd_word(700, 1, true)), // x >= width -> dropped
282        ]);
283        let options = LoadOptions {
284            sensor_size: Some((4, 4)),
285            ..LoadOptions::default()
286        };
287        assert_eq!(read(&data, &options).unwrap().len(), 1);
288    }
289
290    #[test]
291    fn non_cd_event_size_is_unsupported() {
292        let mut data = b"% Width 640\n% Height 480\n".to_vec();
293        data.extend_from_slice(&[0x0C, 16]); // 16-byte events -> unsupported
294        match read(&data, &LoadOptions::default()) {
295            Err(IoError::Unsupported(message)) => assert!(message.contains("event size")),
296            other => panic!("expected unsupported event-size error, got {other:?}"),
297        }
298    }
299
300    #[test]
301    fn missing_geometry_without_override_is_unsupported() {
302        let mut data = b"% Date 2024-01-01\n% Version 2\n".to_vec();
303        data.extend_from_slice(&[0x00, CD_EVENT_SIZE]);
304        data.extend_from_slice(&record(1, cd_word(1, 1, true)));
305        match read(&data, &LoadOptions::default()) {
306            Err(IoError::Unsupported(message)) => assert!(message.contains("sensor_size")),
307            other => panic!("expected unsupported geometry error, got {other:?}"),
308        }
309    }
310
311    #[test]
312    fn explicit_sensor_size_overrides_the_header() {
313        let data = file_with(&[record(1, cd_word(1, 1, true))]);
314        let options = LoadOptions {
315            sensor_size: Some((1280, 720)),
316            ..LoadOptions::default()
317        };
318        assert_eq!(read(&data, &options).unwrap().sensor_size(), (1280, 720));
319    }
320}
eventcv_core/io/prophesee.rs

eventcv_core/io/
prophesee.rs
