iguazu 0.0.1

Tools for viewing, storing, and sharing mixed-signal time series 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

use std::{pin::Pin, sync::Arc};
use std::future;
use crate::import::OptionDescription;
use crate::schema::EntityStream;
use crate::{io::ReadableFile, schema::{EntitySchema, attribute}, storage::{Pool, FlatFileOpts, FlatFileStream}};

use super::{ImportError, Importer};

/// Importer for flat files that are directly read from disk.
pub struct FlatFileImporter {
    schema: EntitySchema,
    opts: FlatFileOpts,
}

impl FlatFileImporter {
    pub fn new(schema: EntitySchema) -> Self {
        Self {
            schema,
            opts: FlatFileOpts::default(),
        }
    }

    pub fn binary() -> Self {
        Self::new(EntitySchema::bytes())
    }

    pub fn logic8() -> Self {
        Self::new(EntitySchema::logic8())
    }

    pub fn offset(&self) -> u64 {
        self.opts.offset
    }

    pub fn with_offset(mut self, offset: u64) -> Self {
        self.opts.offset = offset;
        self
    }

    pub fn count(&self) -> Option<u64> {
        self.opts.count
    }

    pub fn with_count(mut self, count: Option<u64>) -> Self {
        self.opts.count = count;
        self
    }

    pub fn block_size(&self) -> usize {
        self.opts.block_size
    }

    pub fn with_block_size(mut self, block_size: usize) -> Self {
        self.opts.block_size = block_size;
        self
    }
}

impl Importer for FlatFileImporter {
    fn options(&self) -> &'static [ super::OptionDescription ] {
        &[
            OptionDescription {
                name: "offset",
                description: "Byte offset in the file where the data starts.",
            },
            OptionDescription {
                name: "count",
                description: "Number of elements to read from the file. Empty means to read until the end of the file.",
            },
            OptionDescription {
                name: "block_size",
                description: "Number of elements to read in each block.",
            },
            OptionDescription {
                name: "sample_rate",
                description: "Sample rate",
            },
        ]
    }

    fn set(&mut self, option: &str, value: &str) -> Result<(), String> {
        match option {
            "offset" => self.opts.offset = value.parse().map_err(|_| "Invalid integer")?,
            "count" => self.opts.count = if value.is_empty() { None } else { Some(value.parse().map_err(|_| "Invalid integer")?) },
            "block_size" => self.opts.block_size = value.parse().map_err(|_| "Invalid integer")?,
            "sample_rate" => {
                if value.is_empty() {
                    self.schema.remove_attribute(attribute::core::SAMPLE_RATE);
                } else {
                    let val = value.parse().map_err(|_| "Invalid float")?;
                    self.schema.set_attribute(attribute::core::SAMPLE_RATE, val);
                }
            }
            _ => return Err("Unknown option".into()),
        }
        Ok(())
    }

    fn get(&self, option: &str) -> Option<String> {
        Some(match option {
            "offset" => self.opts.offset.to_string(),
            "count" => self.opts.count.map_or_else(|| "".into(), |c| c.to_string()),
            "block_size" => self.opts.block_size.to_string(),
            "sample_rate" => self.schema.attribute(attribute::core::SAMPLE_RATE)?.to_string(),
            _ => return None,
        })
    }

    fn load_schema(&self, _file: Arc<dyn ReadableFile>) -> Pin<Box<dyn Future<Output = Result<EntitySchema, ImportError>> + Send>> {
        Box::pin(future::ready(Ok(self.schema.clone())))
    }

    fn import(&self, file: Arc<dyn ReadableFile>, schema: Option<EntitySchema>, pool: Arc<Pool>) -> Pin<Box<dyn Future<Output = Result<(EntityStream, Pin<Box<dyn Future<Output = Result<(), ImportError>> + Send>>), ImportError>> + Send + '_>> {
        Box::pin(async move {
            let schema = schema.unwrap_or_else(|| self.schema.clone());
            let entity = FlatFileStream::entity(file, pool, schema, &self.opts).await?;
            Ok((entity, Box::pin(async move {Ok(())}) as Pin<Box<_>>))
        })
    }
}