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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
//! Provides stream messages that come from a running stream
use crate::config::*;
use crate::daq::Qty;
use crate::siggen::Siggen;
use anyhow::{bail, Result};
use cpal::Sample;
use crossbeam::channel::Sender;
use reinterpret::{reinterpret_slice, reinterpret_vec};
use std::any::TypeId;
use std::sync::{Arc, RwLock};
use std::u128::MAX;
use strum_macros::Display;
use super::*;
/// Raw stream data coming from a stream.
#[derive(Clone, Debug)]
pub enum RawStreamData {
/// 8-bits integer
Datai8(Vec<i8>),
/// 16-bits integer
Datai16(Vec<i16>),
/// 32-bits integer
Datai32(Vec<i32>),
/// 32-bits float
Dataf32(Vec<f32>),
/// 64-bits float
Dataf64(Vec<f64>),
/// A stream error occured
StreamError(StreamError),
}
impl RawStreamData {
pub fn toFloat(&self, _nchannels: usize) -> Dmat {
// match &self {
// RawStreamData::Datai8(c) => {
// Dmat::zeros((2, 2));
// }
// RawStreamData::Datai16(c) => {
// Dmat::zeros((2, 2));
// }
// }
todo!()
}
}
impl RawStreamData {
/// Get reference to raw data buffer
pub fn getRef<T>(&self) -> &[T]
where
T: Sample + 'static,
{
let thetype: TypeId = TypeId::of::<T>();
match &self {
RawStreamData::Datai8(t) => {
let i8type: TypeId = TypeId::of::<i8>();
assert!(thetype == i8type);
let v: &[T] = unsafe { reinterpret_slice(t) };
v
}
RawStreamData::Datai16(t) => {
let i16type: TypeId = TypeId::of::<i16>();
assert!(thetype == i16type);
let v: &[T] = unsafe { reinterpret_slice(t) };
v
}
RawStreamData::Datai32(t) => {
let i32type: TypeId = TypeId::of::<i32>();
assert!(thetype == i32type);
let v: &[T] = unsafe { reinterpret_slice(t) };
v
}
RawStreamData::Dataf32(t) => {
let f32type: TypeId = TypeId::of::<f32>();
assert!(thetype == f32type);
let v: &[T] = unsafe { reinterpret_slice(t) };
v
}
RawStreamData::Dataf64(t) => {
let f64type: TypeId = TypeId::of::<f64>();
assert!(thetype == f64type);
let v: &[T] = unsafe { reinterpret_slice(t) };
v
}
_ => panic!("Cannot getRef from "),
}
}
}
// Create InStreamData object from
impl<T> From<&[T]> for RawStreamData
where
T: num::ToPrimitive + Clone + 'static,
{
fn from(input: &[T]) -> RawStreamData {
// Apparently, this code does not work with a match. I have searched around and have not found the
// reason for this. So this is a bit of stupid boilerplate.
let i8type: TypeId = TypeId::of::<i8>();
let i16type: TypeId = TypeId::of::<i16>();
let i32type: TypeId = TypeId::of::<i32>();
let f32type: TypeId = TypeId::of::<f32>();
let f64type: TypeId = TypeId::of::<f64>();
let thetype: TypeId = TypeId::of::<T>();
if i8type == thetype {
let v: Vec<i8> = unsafe { reinterpret_slice(input).to_vec() };
RawStreamData::Datai8(v)
} else if i16type == thetype {
let v: Vec<i16> = unsafe { reinterpret_slice(input).to_vec() };
RawStreamData::Datai16(v)
} else if i16type == thetype {
let v: Vec<i16> = unsafe { reinterpret_slice(input).to_vec() };
RawStreamData::Datai16(v)
} else if i32type == thetype {
let v: Vec<i32> = unsafe { reinterpret_slice(input).to_vec() };
RawStreamData::Datai32(v)
} else if f32type == thetype {
let v: Vec<f32> = unsafe { reinterpret_slice(input).to_vec() };
RawStreamData::Dataf32(v)
} else if f64type == thetype {
let v: Vec<f64> = unsafe { reinterpret_slice(input).to_vec() };
RawStreamData::Dataf64(v)
} else {
panic!("Not implemented sample type!")
}
}
}
// Create InStreamData object from
impl<T> From<Vec<T>> for RawStreamData
where
T: num::ToPrimitive + Clone + 'static,
{
fn from(input: Vec<T>) -> RawStreamData {
// Apparently, this code does not work with a match. I have searched around and have not found the
// reason for this. So this is a bit of stupid boilerplate.
let i8type: TypeId = TypeId::of::<i8>();
let i16type: TypeId = TypeId::of::<i16>();
let i32type: TypeId = TypeId::of::<i32>();
let f32type: TypeId = TypeId::of::<f32>();
let f64type: TypeId = TypeId::of::<f64>();
let thetype: TypeId = TypeId::of::<T>();
if i8type == thetype {
let v: Vec<i8> = unsafe { reinterpret_vec(input) };
RawStreamData::Datai8(v)
} else if i16type == thetype {
let v: Vec<i16> = unsafe { reinterpret_vec(input) };
RawStreamData::Datai16(v)
} else if i16type == thetype {
let v: Vec<i16> = unsafe { reinterpret_vec(input) };
RawStreamData::Datai16(v)
} else if i32type == thetype {
let v: Vec<i32> = unsafe { reinterpret_vec(input) };
RawStreamData::Datai32(v)
} else if f32type == thetype {
let v: Vec<f32> = unsafe { reinterpret_vec(input) };
RawStreamData::Dataf32(v)
} else if f64type == thetype {
let v: Vec<f64> = unsafe { reinterpret_vec(input) };
RawStreamData::Dataf64(v)
} else {
panic!("Not implemented sample type!")
}
}
}
/// Stream metadata. All information required for properly interpreting the raw
/// data that is coming from the stream.
#[derive(Clone, Debug)]
pub struct StreamMetaData {
/// Information for each channel in the stream
pub channelInfo: Vec<DaqChannel>,
/// The data type of the device [Number / voltage / Acoustic pressure / ...]
pub rawDatatype: DataType,
/// Sample rate in [Hz]
pub samplerate: Flt,
/// The number of frames per block of data that comes in. Multiplied by
/// channelInfo.len() we get the total number of samples that come in at
/// each callback.
pub framesPerBlock: usize,
}
impl StreamMetaData {
/// Create new metadata object.
/// ///
/// # Args
///
pub fn new(
channelInfo: &[DaqChannel],
rawdtype: DataType,
sr: Flt,
framesPerBlock: usize,
) -> Result<StreamMetaData> {
Ok(StreamMetaData {
channelInfo: channelInfo.to_vec(),
rawDatatype: rawdtype,
samplerate: sr,
framesPerBlock,
})
}
/// Returns the number of channels in the stream metadata.
pub fn nchannels(&self) -> usize {
self.channelInfo.len()
}
}
/// Stream data (audio / other) coming from a stream or to be send to a stream
#[derive(Debug)]
pub struct StreamData {
/// Package counter. Should always increase monotonically.
pub ctr: usize,
/// Stream metadata. All info required for properly interpreting the raw data.
pub meta: Arc<StreamMetaData>,
/// This is typically what is stored when recording
pub raw: RawStreamData,
// Converted to floating point format. Used for further real time
// processing. Stored in an rw-lock. The first thread that acesses this data
// will perform the conversion. All threads after that will get the data.
converted: RwLock<Option<Arc<Dmat>>>,
}
impl StreamData {
/// Create new stream data object.
pub fn new(ctr: usize, meta: Arc<StreamMetaData>, raw: RawStreamData) -> StreamData {
StreamData {
ctr,
meta,
raw,
converted: RwLock::new(None),
}
}
/// Get the data in floating point format. If already converted, uses the
/// cached float data.
pub fn getFloatData(&self) -> Arc<Dmat> {
if let Some(dat) = self.converted.read().unwrap().as_ref() {
return dat.clone();
}
// In case we reach here, the data has not yet be converted to floating
// point, so we do this.
let mut o = self.converted.write().unwrap();
// It might be that another thread was 'first', and already performed
// the conversion. In that case, we still do an early return, and we
// just openend the lock twice for writing. Not a problem.
if let Some(dat) = o.as_ref() {
return dat.clone();
}
// Perform the actual conversion
let converted_data = Arc::new(self.raw.toFloat(self.meta.nchannels()));
// Replace the option with the Some
o.replace(converted_data.clone());
converted_data
}
}
#[cfg(test)]
mod test {
use num::traits::sign;
use cpal::Sample;
use super::*;
#[test]
fn test() {
const fs: Flt = 20.;
// Number of samples per channel
const Nframes: usize = 20;
const Nch: usize = 2;
let mut signal = [0.; Nch*Nframes];
let mut siggen = Siggen::newSine(Nch, 1.);
siggen.reset(fs);
siggen.setMute(&[false, true]);
siggen.genSignal(&mut signal);
let raw: Vec<i16> = Vec::from_iter(signal.iter().map(
|o| o.to_sample::<i16>()));
let ms1 = raw.iter().step_by(2).map(|s1| {*s1 as f64 * *s1 as f64}).sum::<f64>() / Nframes as f64;
let i16maxsq = (i16::MAX as f64).powf(2.);
// println!("ms1: {} {}", ms1, i16maxsq/2.);
// println!("{:?}", raw.iter().cloned().step_by(2).collect::<Vec<i16>>());
// println!("{:?}", i16::EQUILIBRIUM);
assert!(f64::abs(ms1 - i16maxsq/2.)/i16maxsq < 1e-3);
}
}