rtlola_streamir_interpreter/
csv.rs

1//! Provides an interface for reading inputs from a csv file and writing verdicts as CSV.
2
3use std::collections::HashMap;
4use std::io::{Read, Write};
5use std::time::Duration;
6
7use itertools::Itertools;
8use rtlola_streamir::ir::{StreamIr, StreamReference};
9use thiserror::Error;
10
11use crate::value::{Value, ValueConvertError};
12use crate::verdict::{Change, TotalIncremental};
13use crate::Inputs;
14
15pub(crate) type ParseValueFn = Box<dyn Fn(&[u8]) -> Result<Option<Value>, ValueConvertError>>;
16
17/// Contains a csv file and can produce new values according to the rows in the file.
18pub struct CsvEventSource<R: Read> {
19    reader: csv::ByteRecordsIntoIter<R>,
20    parser: Vec<ParseValueFn>,
21}
22
23impl<R: Read> std::fmt::Debug for CsvEventSource<R> {
24    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
25        f.debug_struct("CsvEventSource").finish()
26    }
27}
28
29#[derive(Error, Debug)]
30/// An error that can happen while reading or writing from or to a csv file
31pub enum CsvError {
32    #[error("Error while parsing csv file")]
33    /// An error happened while reading/writing CSV
34    CsvError(#[from] csv::Error),
35    #[error("Error while converting to value")]
36    /// An error occured while converting the CSV value into a [Value]
37    ValueError(#[from] ValueConvertError),
38    #[error("The csv file was missing a column")]
39    /// The csv file is missing a column
40    MissingColumn,
41    #[error("The csv output includes parameterized stream \"{0}\"")]
42    /// The csv writer tried to write a parameterized output stream
43    Parameterized(String),
44}
45
46impl<R: Read> CsvEventSource<R> {
47    /// Construct a new [CsvEventSource] reading CSV rows from `read`.
48    pub fn new(read: R, ir: &StreamIr) -> Self {
49        let parser = ir
50            .inputs()
51            .sorted()
52            .map(|i| {
53                let ty = ir.stream_memory(StreamReference::In(i)).ty.clone();
54                Value::bytes_parser(ty)
55            })
56            .collect();
57
58        Self {
59            reader: csv::Reader::from_reader(read).into_byte_records(),
60            parser,
61        }
62    }
63
64    /// Return the next event (with the associated timestamp) from the CSV file.
65    pub fn next_event(&mut self) -> Result<Option<(Inputs, Duration)>, CsvError> {
66        let Some(record) = self.reader.next() else {
67            return Ok(None);
68        };
69
70        let record = record?;
71        let mut record = record.into_iter();
72        let inputs = self
73            .parser
74            .iter()
75            .map(|p| Ok(p(record.next().ok_or(CsvError::MissingColumn)?)?))
76            .collect::<Result<Vec<_>, CsvError>>()?;
77
78        let ts = record.next().ok_or(CsvError::MissingColumn)?;
79        let ts: f64 = String::from_utf8_lossy(ts).trim().parse().unwrap();
80        let ts = Duration::from_secs_f64(ts);
81
82        Ok(Some((Inputs(inputs), ts)))
83    }
84}
85
86/// Provides a way to write verdicts as CSV rows
87#[derive(Debug)]
88pub struct CsvVerdictSink<W: Write> {
89    writer: csv::Writer<W>,
90    fields: HashMap<StreamReference, usize>,
91    has_input: bool,
92    has_output: bool,
93}
94
95impl<W: Write> CsvVerdictSink<W> {
96    /// Construct a new [CsvVerdictSink] writing to `write`.
97    pub fn new(write: W, ir: &StreamIr, fields: &[StreamReference]) -> Result<Self, CsvError> {
98        let field_map = fields.iter().enumerate().map(|(i, sr)| (*sr, i)).collect();
99
100        let mut writer = csv::Writer::from_writer(write);
101        let header = fields
102            .iter()
103            .map(|sr| ir.name(*sr))
104            .chain(std::iter::once("time"));
105
106        writer.write_record(header)?;
107
108        let has_input = fields
109            .iter()
110            .any(|field| matches!(field, StreamReference::In(_)));
111        let has_output = fields
112            .iter()
113            .any(|field| matches!(field, StreamReference::Out(_)));
114
115        for field in fields {
116            if let StreamReference::Out(_) = field {
117                if ir.stream_memory(*field).parameters().is_some() {
118                    return Err(CsvError::Parameterized(ir.name(*field).to_owned()));
119                }
120            }
121        }
122
123        Ok(Self {
124            writer,
125            fields: field_map,
126            has_input,
127            has_output,
128        })
129    }
130
131    /// Accepts a verdict returned by the monitor and writes it to CSV.
132    pub fn accept_verdict(
133        &mut self,
134        ts: Duration,
135        verdict: TotalIncremental,
136    ) -> Result<(), CsvError> {
137        if !(self.has_input || self.has_output) {
138            return Ok(());
139        }
140
141        let mut fields = vec![None; self.fields.len()];
142        if self.has_input {
143            for (sr, value) in &verdict.inputs {
144                if let Some(idx) = self.fields.get(&StreamReference::In(*sr)) {
145                    fields[*idx] = Some(value)
146                }
147            }
148        }
149        if self.has_output {
150            for (sr, changes) in &verdict.outputs {
151                if let Some(idx) = self.fields.get(&sr.sr()) {
152                    if let Some(value) = changes.iter().find_map(|change| match change {
153                        Change::Value(inst, value) => {
154                            debug_assert!(inst.is_none());
155                            Some(value)
156                        }
157                        _ => None,
158                    }) {
159                        fields[*idx] = Some(value)
160                    }
161                }
162            }
163        }
164
165        if fields.iter().all(|field| field.is_none()) {
166            return Ok(());
167        }
168
169        let fields = fields
170            .into_iter()
171            .map(|value| match value {
172                Some(value) => value.to_string(),
173                None => "#".into(),
174            })
175            .chain(std::iter::once(ts.as_secs_f64().to_string()));
176
177        self.writer.write_record(fields)?;
178        self.writer.flush().unwrap();
179        Ok(())
180    }
181}
rtlola_streamir_interpreter/csv.rs

rtlola_streamir_interpreter/
csv.rs
