[READ-ONLY] Mirror of https://github.com/probablykasper/csv-pipeline.
Add sum transformer
+101
-58
+22
-16
README.md
+22
-16
README.md
···
5
5
[](https://crates.io/crates/csv-pipeline)
6
6
[](https://docs.rs/csv-pipeline)
7
7
8
-
## Basic Example
8
+
## Example
9
+
9
10
```rs
10
11
use csv_pipeline::{Pipeline, Transformer};
11
12
12
-
// First create a pipeline from a CSV file path
13
-
let csv = Pipeline::from_path("test/Countries.csv")
13
+
let source = "\
14
+
Person,Score\n\
15
+
A,1\n\
16
+
A,8\n\
17
+
B,3\n\
18
+
B,4\n";
19
+
let reader = csv::Reader::from_reader(source.as_bytes());
20
+
let csv = Pipeline::from_reader(reader)
14
21
.unwrap()
15
-
// Add a column with values computed from a closure
16
-
.add_col("Language", |headers, row| {
17
-
match headers.get_field(row, "Country") {
18
-
Some("Norway") => Ok("Norwegian".into()),
19
-
_ => Ok("Unknown".into()),
20
-
}
22
+
.map(|_headers, row| Ok(row))
23
+
// Transform into a new csv
24
+
.transform_into(|| {
25
+
vec![
26
+
// Keep every Person
27
+
Transformer::new("Person").keep_unique(),
28
+
// Sum the scores into a "Total score" column
29
+
Transformer::new("Total score").from_col("Score").sum(0),
30
+
]
21
31
})
22
-
// Make the "Country" column uppercase
23
-
.rename_col("Country", "COUNTRY")
24
-
.map_col("COUNTRY", |id_str| Ok(id_str.to_uppercase()))
25
-
// Collect the csv into a string
26
32
.collect_into_string()
27
33
.unwrap();
28
34
29
35
assert_eq!(
30
36
csv,
31
-
"ID,COUNTRY,Language\n\
32
-
1,NORWAY,Norwegian\n\
33
-
2,TUVALU,Unknown\n"
37
+
"Person,Total score\n\
38
+
A,9\n\
39
+
B,7\n"
34
40
);
35
41
```
36
42
+1
-7
src/lib.rs
+1
-7
src/lib.rs
···
63
63
//! // Keep every Person
64
64
//! Transformer::new("Person").keep_unique(),
65
65
//! // Sum the scores into a "Total score" column
66
-
//! Transformer::new("Total score").from_col("Score").reduce(
67
-
//! |accumulator, current| {
68
-
//! let score: u64 = current.parse().unwrap();
69
-
//! Ok(accumulator + score)
70
-
//! },
71
-
//! 0,
72
-
//! ),
66
+
//! Transformer::new("Total score").from_col("Score").sum(0),
73
67
//! ]
74
68
//! })
75
69
//! .collect_into_string()
+5
-5
src/pipeline.rs
+5
-5
src/pipeline.rs
···
136
136
/// ## Example
137
137
///
138
138
/// ```
139
-
/// use csv_pipeline::{Pipeline, StringTarget};
139
+
/// use csv_pipeline::Pipeline;
140
140
///
141
141
/// let csv = Pipeline::from_path("test/AB.csv")
142
142
/// .unwrap()
···
159
159
/// ## Example
160
160
///
161
161
/// ```
162
-
/// use csv_pipeline::{Pipeline, StringTarget};
162
+
/// use csv_pipeline::Pipeline;
163
163
///
164
164
/// let csv = Pipeline::from_path("test/AB.csv")
165
165
/// .unwrap()
···
195
195
/// ## Example
196
196
///
197
197
/// ```
198
-
/// use csv_pipeline::{Pipeline, StringTarget};
198
+
/// use csv_pipeline::Pipeline;
199
199
///
200
200
/// let csv = Pipeline::from_path("test/AB.csv")
201
201
/// .unwrap()
···
258
258
/// ## Example
259
259
///
260
260
/// ```
261
-
/// use csv_pipeline::{Pipeline, StringTarget};
261
+
/// use csv_pipeline::{Pipeline, Target};
262
262
///
263
263
/// let mut csv = String::new();
264
264
/// Pipeline::from_path("test/AB.csv")
265
265
/// .unwrap()
266
-
/// .flush(StringTarget::new(&mut csv))
266
+
/// .flush(Target::string(&mut csv))
267
267
/// .run()
268
268
/// .unwrap();
269
269
///
+73
-30
src/transform.rs
+73
-30
src/transform.rs
···
2
2
use core::fmt::Display;
3
3
use std::collections::hash_map::DefaultHasher;
4
4
use std::hash::{Hash, Hasher};
5
+
use std::ops::AddAssign;
6
+
use std::str::FromStr;
5
7
6
8
/// For grouping and reducing rows.
7
9
pub trait Transform {
···
50
52
value: "".to_string(),
51
53
})
52
54
}
55
+
/// Sum the values in this column
56
+
pub fn sum<'a, N>(self, init: N) -> Box<dyn Transform + 'a>
57
+
where
58
+
N: Display + AddAssign + FromStr + Clone + 'a,
59
+
{
60
+
Box::new(Sum {
61
+
name: self.name,
62
+
from_col: self.from_col,
63
+
value: init,
64
+
})
65
+
}
53
66
/// Reduce the values from this column into a single value using a closure
54
67
pub fn reduce<'a, R, V>(self, reduce: R, init: V) -> Box<dyn Transform + 'a>
55
68
where
56
69
R: FnMut(V, &str) -> Result<V, Error> + 'a,
57
70
V: Display + Clone + 'a,
58
71
{
59
-
Box::new(Closure {
72
+
Box::new(Reduce {
60
73
name: self.name,
61
74
from_col: self.from_col,
62
75
reduce,
···
65
78
}
66
79
}
67
80
68
-
struct Closure<F, V> {
69
-
name: String,
70
-
from_col: String,
71
-
reduce: F,
72
-
value: V,
73
-
}
74
-
impl<F, V> Transform for Closure<F, V>
75
-
where
76
-
F: FnMut(V, &str) -> Result<V, Error>,
77
-
V: Display + Clone,
78
-
{
79
-
fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> {
80
-
let field = headers
81
-
.get_field(row, &self.from_col)
82
-
.ok_or(Error::MissingColumn(self.from_col.clone()))?
83
-
.to_string();
84
-
self.value = (self.reduce)(self.value.clone(), &field)?;
85
-
Ok(())
86
-
}
87
-
88
-
fn value(&self) -> String {
89
-
self.value.to_string()
90
-
}
91
-
92
-
fn name(&self) -> String {
93
-
self.name.clone()
94
-
}
95
-
}
96
-
97
81
struct KeepUnique {
98
82
name: String,
99
83
from_col: String,
···
139
123
}
140
124
Ok(hasher.finish())
141
125
}
126
+
127
+
struct Reduce<F, V> {
128
+
name: String,
129
+
from_col: String,
130
+
reduce: F,
131
+
value: V,
132
+
}
133
+
impl<F, V> Transform for Reduce<F, V>
134
+
where
135
+
F: FnMut(V, &str) -> Result<V, Error>,
136
+
V: Display + Clone,
137
+
{
138
+
fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> {
139
+
let field = headers
140
+
.get_field(row, &self.from_col)
141
+
.ok_or(Error::MissingColumn(self.from_col.clone()))?
142
+
.to_string();
143
+
self.value = (self.reduce)(self.value.clone(), &field)?;
144
+
Ok(())
145
+
}
146
+
147
+
fn value(&self) -> String {
148
+
self.value.to_string()
149
+
}
150
+
151
+
fn name(&self) -> String {
152
+
self.name.clone()
153
+
}
154
+
}
155
+
156
+
struct Sum<N> {
157
+
name: String,
158
+
from_col: String,
159
+
value: N,
160
+
}
161
+
impl<V> Transform for Sum<V>
162
+
where
163
+
V: Display + AddAssign + FromStr + Clone,
164
+
{
165
+
fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> {
166
+
let field = headers
167
+
.get_field(row, &self.from_col)
168
+
.ok_or(Error::MissingColumn(self.from_col.clone()))?
169
+
.to_string();
170
+
let new: V = match field.parse() {
171
+
Ok(v) => v,
172
+
Err(_) => return Err(Error::InvalidField(field)),
173
+
};
174
+
self.value += new;
175
+
Ok(())
176
+
}
177
+
178
+
fn value(&self) -> String {
179
+
self.value.to_string()
180
+
}
181
+
fn name(&self) -> String {
182
+
self.name.clone()
183
+
}
184
+
}