use std::fs::File;
use std::io::{BufRead, BufReader, BufWriter, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::str::{FromStr, SplitWhitespace};
use super::{read_all, read_capped, EventSource, IoError, LoadOptions, RawEvent, SliceSource};
use crate::{EventStream, EventStreamBuilder};
pub fn write_text_stream(path: impl AsRef<Path>, stream: &EventStream) -> Result<(), IoError> {
let mut writer = BufWriter::new(File::create(path).map_err(IoError::Io)?);
let (xs, ys, ts, ps) = (stream.xs(), stream.ys(), stream.ts(), stream.ps());
for index in 0..stream.len() {
writeln!(
writer,
"{} {} {} {}",
ts[index],
xs[index],
ys[index],
u8::from(ps[index])
)
.map_err(IoError::Io)?;
}
writer.flush().map_err(IoError::Io)
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TimeUnit {
Seconds,
Milliseconds,
Microseconds,
Nanoseconds,
}
impl TimeUnit {
fn to_microseconds(self, value: f64) -> i64 {
let microseconds = match self {
Self::Seconds => value * 1e6,
Self::Milliseconds => value * 1e3,
Self::Microseconds => value,
Self::Nanoseconds => value / 1e3,
};
microseconds.round() as i64
}
#[cfg_attr(not(feature = "hdf5"), allow(dead_code))]
pub(crate) fn from_scale_ms(scale_ms: f64) -> Option<TimeUnit> {
for (unit, expected) in [
(TimeUnit::Nanoseconds, 1e-6),
(TimeUnit::Microseconds, 1e-3),
(TimeUnit::Milliseconds, 1.0),
(TimeUnit::Seconds, 1e3),
] {
if (scale_ms - expected).abs() <= expected * 1e-9 {
return Some(unit);
}
}
None
}
pub(crate) fn infer_from_span(span: i64) -> TimeUnit {
let span = span.max(0) as f64;
if span * 1e-9 >= 1.0 {
TimeUnit::Nanoseconds
} else if span * 1e-6 >= 1.0 {
TimeUnit::Microseconds
} else if span * 1e-3 >= 1.0 {
TimeUnit::Milliseconds
} else {
TimeUnit::Seconds
}
}
#[cfg(feature = "hdf5")]
pub(crate) fn microseconds_from_int(self, value: i64) -> i64 {
let value = i128::from(value);
let microseconds = match self {
Self::Seconds => value * 1_000_000,
Self::Milliseconds => value * 1_000,
Self::Microseconds => value,
Self::Nanoseconds => value / 1_000,
};
microseconds.clamp(i64::MIN as i128, i64::MAX as i128) as i64
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum ColumnOrder {
#[default]
Txyp,
Xytp,
}
#[derive(Clone, Copy, Debug)]
pub struct TextOptions {
pub width: usize,
pub height: usize,
pub time_unit: TimeUnit,
pub order: ColumnOrder,
}
impl TextOptions {
pub fn new(width: usize, height: usize) -> Self {
Self {
width,
height,
time_unit: TimeUnit::Seconds,
order: ColumnOrder::Txyp,
}
}
}
#[derive(Debug)]
pub struct TextReader<R> {
reader: R,
options: TextOptions,
buffer: String,
line: usize,
}
impl<R: BufRead> TextReader<R> {
pub fn new(reader: R, options: TextOptions) -> Result<Self, IoError> {
if options.width == 0 || options.height == 0 {
return Err(IoError::InvalidSensorSize);
}
Ok(Self {
reader,
options,
buffer: String::new(),
line: 0,
})
}
fn parse_line(&self, line: &str) -> Result<RawEvent, IoError> {
let mut fields = line.split_whitespace();
let (t, x, y, p) = match self.options.order {
ColumnOrder::Txyp => {
let t = self.field(&mut fields, "t")?;
(
t,
self.field(&mut fields, "x")?,
self.field(&mut fields, "y")?,
self.field(&mut fields, "p")?,
)
}
ColumnOrder::Xytp => {
let x = self.field(&mut fields, "x")?;
let y = self.field(&mut fields, "y")?;
(
self.field(&mut fields, "t")?,
x,
y,
self.field(&mut fields, "p")?,
)
}
};
Ok(RawEvent {
x: self.parse(x, "x")?,
y: self.parse(y, "y")?,
t: self
.options
.time_unit
.to_microseconds(self.parse::<f64>(t, "t")?),
p: self.parse::<i32>(p, "p")? > 0,
})
}
fn field<'a>(&self, fields: &mut SplitWhitespace<'a>, name: &str) -> Result<&'a str, IoError> {
fields.next().ok_or_else(|| IoError::Parse {
line: self.line,
message: format!("missing {name}"),
})
}
fn parse<T: FromStr>(&self, value: &str, field: &str) -> Result<T, IoError> {
value.parse().map_err(|_| IoError::Parse {
line: self.line,
message: format!("invalid {field}: {value:?}"),
})
}
}
impl<R: BufRead> EventSource for TextReader<R> {
fn sensor_size(&self) -> (usize, usize) {
(self.options.width, self.options.height)
}
fn timestamp_scale_ms(&self) -> f64 {
0.001
}
fn next_event(&mut self) -> Result<Option<RawEvent>, IoError> {
loop {
self.buffer.clear();
self.line += 1;
if self.reader.read_line(&mut self.buffer)? == 0 {
return Ok(None);
}
let trimmed = self.buffer.trim();
if trimmed.is_empty() || trimmed.starts_with('#') {
continue;
}
return self.parse_line(trimmed).map(Some);
}
}
}
pub fn open(
path: impl AsRef<Path>,
options: TextOptions,
) -> Result<TextReader<BufReader<File>>, IoError> {
TextReader::new(BufReader::new(File::open(path)?), options)
}
pub fn read_text(path: impl AsRef<Path>, options: TextOptions) -> Result<EventStream, IoError> {
read_all(open(path, options)?)
}
pub struct RawRow {
pub x: u16,
pub y: u16,
pub t: f64,
pub p: bool,
}
pub fn load_text(path: impl AsRef<Path>, options: &LoadOptions) -> Result<EventStream, IoError> {
if let (Some((width, height)), Some(time_unit)) = (options.sensor_size, options.time_unit) {
let text_options = TextOptions {
width,
height,
time_unit,
order: options.order,
};
return read_capped(open(path.as_ref(), text_options)?, options.max_events);
}
let rows = read_raw_rows(path.as_ref(), options.order)?;
load_rows(&rows, options)
}
pub fn load_rows(rows: &[RawRow], options: &LoadOptions) -> Result<EventStream, IoError> {
let (width, height) = options
.sensor_size
.unwrap_or_else(|| infer_sensor_size(rows));
let time_unit = options.time_unit.unwrap_or_else(|| infer_time_unit(rows));
if width == 0 || height == 0 {
return Err(IoError::InvalidSensorSize);
}
let mut builder = EventStreamBuilder::new(width, height, 0.001);
for row in rows {
builder.push(row.x, row.y, time_unit.to_microseconds(row.t), row.p);
if options.max_events.is_some_and(|max| builder.len() >= max) {
break;
}
}
Ok(builder.build())
}
fn infer_sensor_size(rows: &[RawRow]) -> (usize, usize) {
let width = rows.iter().map(|row| usize::from(row.x)).max();
let height = rows.iter().map(|row| usize::from(row.y)).max();
match (width, height) {
(Some(width), Some(height)) => (width + 1, height + 1),
_ => (1, 1),
}
}
fn infer_time_unit(rows: &[RawRow]) -> TimeUnit {
if rows.iter().any(|row| row.t.fract() != 0.0) {
return TimeUnit::Seconds;
}
let min = rows.iter().map(|row| row.t).fold(f64::INFINITY, f64::min);
let max = rows
.iter()
.map(|row| row.t)
.fold(f64::NEG_INFINITY, f64::max);
if min.is_finite() {
TimeUnit::infer_from_span((max - min) as i64)
} else {
TimeUnit::Seconds
}
}
fn parse_raw_row(trimmed: &str, order: ColumnOrder, number: usize) -> Result<RawRow, IoError> {
let mut fields = trimmed.split_whitespace();
let mut next = |name: &str| {
fields.next().ok_or(IoError::Parse {
line: number,
message: format!("missing {name}"),
})
};
let (t, x, y, p) = match order {
ColumnOrder::Txyp => {
let t = next("t")?;
(t, next("x")?, next("y")?, next("p")?)
}
ColumnOrder::Xytp => {
let x = next("x")?;
let y = next("y")?;
(next("t")?, x, y, next("p")?)
}
};
let parse = |value: &str, field: &str| {
value.parse::<f64>().map_err(|_| IoError::Parse {
line: number,
message: format!("invalid {field}: {value:?}"),
})
};
Ok(RawRow {
x: parse(x, "x")? as u16,
y: parse(y, "y")? as u16,
t: parse(t, "t")?,
p: parse(p, "p")? > 0.0,
})
}
fn read_raw_rows(path: &Path, order: ColumnOrder) -> Result<Vec<RawRow>, IoError> {
let reader = BufReader::new(File::open(path)?);
let mut rows = Vec::new();
for (index, line) in reader.lines().enumerate() {
let line = line?;
let trimmed = line.trim();
if trimmed.is_empty() || trimmed.starts_with('#') {
continue;
}
rows.push(parse_raw_row(trimmed, order, index + 1)?);
}
Ok(rows)
}
const TEXT_INDEX_STRIDE: usize = 4096;
#[derive(Clone, Copy)]
struct TextIndexEntry {
offset: u64,
count: usize,
t_us: i64,
}
pub struct TextSliceSource {
path: PathBuf,
order: ColumnOrder,
time_unit: TimeUnit,
sensor: (usize, usize),
total: usize,
span_us: (i64, i64),
index: Vec<TextIndexEntry>,
}
pub fn open_text_slice(
path: impl AsRef<Path>,
options: &LoadOptions,
) -> Result<TextSliceSource, IoError> {
let path = path.as_ref();
let mut reader = BufReader::new(File::open(path)?);
let mut buffer = String::new();
let mut offset = 0u64;
let mut line_no = 0usize;
let mut kept = 0usize;
let mut samples: Vec<(u64, usize, f64)> = Vec::new();
let (mut max_x, mut max_y) = (0u16, 0u16);
let (mut min_t, mut max_t) = (f64::INFINITY, f64::NEG_INFINITY);
let mut fractional = false;
let mut sorted = true;
let mut previous_t = f64::NEG_INFINITY;
loop {
buffer.clear();
let line_start = offset;
let bytes = reader.read_line(&mut buffer)?;
if bytes == 0 {
break;
}
offset += bytes as u64;
line_no += 1;
let trimmed = buffer.trim();
if trimmed.is_empty() || trimmed.starts_with('#') {
continue;
}
let row = parse_raw_row(trimmed, options.order, line_no)?;
if let Some((width, height)) = options.sensor_size {
if usize::from(row.x) >= width || usize::from(row.y) >= height {
continue; }
}
if kept.is_multiple_of(TEXT_INDEX_STRIDE) {
samples.push((line_start, kept, row.t));
}
max_x = max_x.max(row.x);
max_y = max_y.max(row.y);
min_t = min_t.min(row.t);
max_t = max_t.max(row.t);
fractional |= row.t.fract() != 0.0;
sorted &= row.t >= previous_t;
previous_t = row.t;
kept += 1;
}
let sensor = options
.sensor_size
.unwrap_or((usize::from(max_x) + 1, usize::from(max_y) + 1));
if sensor.0 == 0 || sensor.1 == 0 {
return Err(IoError::InvalidSensorSize);
}
if !sorted {
return Err(IoError::Format(
"text timestamps are not sorted; in-place slicing requires time-ordered events"
.to_owned(),
));
}
let time_unit = options.time_unit.unwrap_or_else(|| {
if fractional || !min_t.is_finite() {
TimeUnit::Seconds
} else {
TimeUnit::infer_from_span((max_t - min_t) as i64)
}
});
let index = samples
.into_iter()
.map(|(offset, count, t)| TextIndexEntry {
offset,
count,
t_us: time_unit.to_microseconds(t),
})
.collect();
let span_us = if kept == 0 {
(0, 0)
} else {
(
time_unit.to_microseconds(min_t),
time_unit.to_microseconds(max_t),
)
};
Ok(TextSliceSource {
path: path.to_path_buf(),
order: options.order,
time_unit,
sensor,
total: kept,
span_us,
index,
})
}
impl TextSliceSource {
fn reader_at(&self, offset: u64) -> Result<TextReader<BufReader<File>>, IoError> {
let mut file = File::open(&self.path)?;
file.seek(SeekFrom::Start(offset))?;
TextReader::new(
BufReader::new(file),
TextOptions {
width: self.sensor.0,
height: self.sensor.1,
time_unit: self.time_unit,
order: self.order,
},
)
}
fn keeps(&self, x: u16, y: u16) -> bool {
usize::from(x) < self.sensor.0 && usize::from(y) < self.sensor.1
}
}
impl SliceSource for TextSliceSource {
fn sensor_size(&self) -> (usize, usize) {
self.sensor
}
fn timestamp_scale_ms(&self) -> f64 {
0.001
}
fn n_events(&self) -> usize {
self.total
}
fn time_span(&self) -> (i64, i64) {
self.span_us
}
fn slice_index(&self, i0: usize, i1: usize) -> Result<EventStream, IoError> {
let i0 = i0.min(self.total);
let i1 = i1.clamp(i0, self.total);
let mut builder = EventStreamBuilder::new(self.sensor.0, self.sensor.1, 0.001);
if i0 == i1 || self.index.is_empty() {
return Ok(builder.build());
}
let entry = self.index[self.index.partition_point(|e| e.count <= i0) - 1];
let mut reader = self.reader_at(entry.offset)?;
let mut index = entry.count;
while index < i1 {
let Some(event) = reader.next_event()? else {
break;
};
if !self.keeps(event.x, event.y) {
continue; }
if index >= i0 {
builder.push(event.x, event.y, event.t, event.p);
}
index += 1;
}
Ok(builder.build())
}
fn slice_time(&self, t0: i64, t1: i64) -> Result<EventStream, IoError> {
let mut builder = EventStreamBuilder::new(self.sensor.0, self.sensor.1, 0.001);
if self.index.is_empty() {
return Ok(builder.build());
}
let entry = self.index[self
.index
.partition_point(|e| e.t_us < t0)
.saturating_sub(1)];
let mut reader = self.reader_at(entry.offset)?;
while let Some(event) = reader.next_event()? {
if event.t >= t1 {
break; }
if event.t >= t0 {
builder.push(event.x, event.y, event.t, event.p);
}
}
Ok(builder.build())
}
}
#[cfg(test)]
mod tests {
use std::io::Cursor;
use super::{load_text, open_text_slice, ColumnOrder, TextOptions, TextReader, TimeUnit};
use crate::io::{read_all, IoError, LoadOptions, SliceSource};
use crate::EventStream;
fn read(data: &str, options: TextOptions) -> Result<EventStream, IoError> {
read_all(TextReader::new(Cursor::new(data), options).unwrap())
}
#[test]
fn parses_txyp_seconds_skips_noise_and_drops_out_of_bounds() {
let data = "0.0 1 2 1\n0.000002 3 0 0\n\n# comment\n0.00001 0 4 1\n0.00002 4 0 1\n";
let stream = read(data, TextOptions::new(4, 5)).unwrap();
assert_eq!(stream.len(), 3); assert_eq!(stream.xs(), &[1, 3, 0]);
assert_eq!(stream.ys(), &[2, 0, 4]);
assert_eq!(stream.ts(), &[0, 2, 10]);
assert_eq!(stream.ps(), &[true, false, true]);
assert_eq!(stream.sensor_size(), (4, 5));
assert_eq!(stream.timestamp_scale_ms(), 0.001);
}
#[test]
fn supports_xytp_order_and_negative_polarity() {
let options = TextOptions {
order: ColumnOrder::Xytp,
..TextOptions::new(8, 8)
};
let stream = read("1 2 0.5 -1\n", options).unwrap();
assert_eq!(stream.xs(), &[1]);
assert_eq!(stream.ys(), &[2]);
assert_eq!(stream.ts(), &[500_000]); assert_eq!(stream.ps(), &[false]); }
#[test]
fn converts_time_units_to_microseconds() {
for (unit, raw, expected) in [
(TimeUnit::Microseconds, "7", 7_i64),
(TimeUnit::Milliseconds, "2", 2_000),
(TimeUnit::Nanoseconds, "2400", 2),
] {
let data = format!("{raw} 0 0 1\n");
let options = TextOptions {
time_unit: unit,
..TextOptions::new(4, 4)
};
assert_eq!(
read(&data, options).unwrap().ts(),
&[expected],
"unit {unit:?}"
);
}
}
#[test]
fn reports_parse_errors_with_line_numbers() {
let error = read("0.0 1 2 1\n0.0 nope 2 1\n", TextOptions::new(4, 4)).unwrap_err();
match error {
IoError::Parse { line, .. } => assert_eq!(line, 2),
other => panic!("expected parse error, got {other:?}"),
}
}
#[test]
fn reports_missing_fields() {
let error = read("0.0 1 2\n", TextOptions::new(4, 4)).unwrap_err();
assert!(matches!(error, IoError::Parse { line: 1, .. }));
}
#[test]
fn rejects_zero_sensor_size() {
let error = TextReader::new(Cursor::new(""), TextOptions::new(0, 4)).unwrap_err();
assert!(matches!(error, IoError::InvalidSensorSize));
}
#[test]
fn empty_input_yields_empty_stream() {
let stream = read("\n\n# only comments\n", TextOptions::new(4, 4)).unwrap();
assert!(stream.is_empty());
assert_eq!(stream.sensor_size(), (4, 4));
}
fn write_temp(tag: &str, contents: &str) -> std::path::PathBuf {
let dir = std::env::temp_dir().join(format!("eventcv-{tag}-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let path = dir.join("events.txt");
std::fs::write(&path, contents).unwrap();
path
}
#[test]
fn load_text_infers_size_and_microseconds() {
let path = write_temp("txtus", "1000000 0 0 1\n3000000 3 1 0\n6000000 1 2 1\n");
let stream = load_text(&path, &LoadOptions::default()).unwrap();
assert_eq!(stream.sensor_size(), (4, 3));
assert_eq!(stream.len(), 3); assert_eq!(stream.ts(), &[1_000_000, 3_000_000, 6_000_000]);
std::fs::remove_dir_all(path.parent().unwrap()).ok();
}
#[test]
fn load_text_infers_seconds_from_a_fractional_value() {
let path = write_temp("txtsec", "0.0 0 0 1\n0.5 1 1 0\n");
let stream = load_text(&path, &LoadOptions::default()).unwrap();
assert_eq!(stream.ts(), &[0, 500_000]); std::fs::remove_dir_all(path.parent().unwrap()).ok();
}
#[test]
fn explicit_options_override_inference() {
let path = write_temp("txtexp", "7 0 0 1\n");
let options = LoadOptions {
sensor_size: Some((4, 4)),
time_unit: Some(TimeUnit::Microseconds),
..LoadOptions::default()
};
let stream = load_text(&path, &options).unwrap();
assert_eq!(stream.sensor_size(), (4, 4));
assert_eq!(stream.ts(), &[7]); std::fs::remove_dir_all(path.parent().unwrap()).ok();
}
#[test]
fn text_slice_source_matches_load() {
let mut data = String::new();
for i in 0..10 {
data.push_str(&format!("{} {} {} {}\n", i * 1000, i, i % 5, i % 2));
}
let path = write_temp("txtslice", &data);
let options = LoadOptions {
time_unit: Some(TimeUnit::Microseconds),
..LoadOptions::default()
};
let source = open_text_slice(&path, &options).unwrap();
let full = load_text(&path, &options).unwrap();
assert_eq!(source.n_events(), full.len());
assert_eq!(source.sensor_size(), full.sensor_size()); assert_eq!(source.time_span(), (0, 9000));
assert_eq!(
source.slice_time(2000, 6000).unwrap().ts(),
&[2000, 3000, 4000, 5000]
);
assert_eq!(
source.slice_index(3, 7).unwrap().ts(),
&[3000, 4000, 5000, 6000]
);
let whole = source.slice_index(0, source.n_events()).unwrap();
assert_eq!(whole.xs(), full.xs());
assert_eq!(whole.ts(), full.ts());
assert_eq!(whole.ps(), full.ps());
std::fs::remove_dir_all(path.parent().unwrap()).ok();
}
#[test]
fn text_slice_rejects_unsorted_timestamps() {
let path = write_temp("txtunsorted", "0 0 0 1\n5000 1 1 0\n2000 2 2 1\n");
match open_text_slice(&path, &LoadOptions::default()) {
Err(IoError::Format(message)) => assert!(message.contains("not sorted")),
Err(other) => panic!("expected a not-sorted error, got {other:?}"),
Ok(_) => panic!("expected unsorted text to be rejected"),
}
std::fs::remove_dir_all(path.parent().unwrap()).ok();
}
#[test]
fn text_slice_keeps_same_timestamp_events_spanning_the_index_stride() {
let dense = super::TEXT_INDEX_STRIDE * 2 + 100;
let mut data = String::from("0 0 0 1\n"); for _ in 0..dense {
data.push_str("1000 1 1 1\n");
}
data.push_str("2000 2 2 0\n"); let path = write_temp("txtdense", &data);
let options = LoadOptions {
time_unit: Some(TimeUnit::Microseconds),
..LoadOptions::default()
};
let source = open_text_slice(&path, &options).unwrap();
assert_eq!(source.slice_time(1000, 2000).unwrap().len(), dense);
let tiled = source.slice_time(0, 1000).unwrap().len()
+ source.slice_time(1000, 2000).unwrap().len()
+ source.slice_time(2000, 3000).unwrap().len();
assert_eq!(tiled, source.n_events());
std::fs::remove_dir_all(path.parent().unwrap()).ok();
}
#[test]
fn text_slice_empty_file() {
let path = write_temp("txtsliceempty", "\n# only a comment\n");
let source = open_text_slice(&path, &LoadOptions::default()).unwrap();
assert_eq!(source.n_events(), 0);
assert_eq!(source.time_span(), (0, 0));
assert!(source.slice_time(0, 1000).unwrap().is_empty());
assert!(source.slice_index(0, 10).unwrap().is_empty());
std::fs::remove_dir_all(path.parent().unwrap()).ok();
}
#[test]
fn write_text_stream_round_trips_at_event_level() {
let mut builder = crate::EventStreamBuilder::new(16, 12, 0.001);
for &(x, y, t, p) in &[(0u16, 0u16, 5i64, true), (15, 11, 2_500_000, false)] {
builder.push(x, y, t, p);
}
let stream = builder.build();
let dir = std::env::temp_dir().join(format!("eventcv-txtrt-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let path = dir.join("events.txt");
super::write_text_stream(&path, &stream).unwrap();
let options = LoadOptions {
sensor_size: Some((16, 12)),
time_unit: Some(TimeUnit::Microseconds),
..LoadOptions::default()
};
let loaded = load_text(&path, &options).unwrap();
assert_eq!(loaded.xs(), stream.xs());
assert_eq!(loaded.ys(), stream.ys());
assert_eq!(loaded.ts(), stream.ts());
assert_eq!(loaded.ps(), stream.ps());
assert_eq!(loaded.sensor_size(), (16, 12));
std::fs::remove_dir_all(&dir).ok();
}
}