chunk to sample process

src/agc.rs

@@ -1,4 +1,3 @@
-use crate::iq_reader::IqChunk;
 use crate::iq_reader::IqSample;
 
 // Automatic Gain Control
@@ -51,56 +50,51 @@ impl<I> Agc<I> {
         }
     }
 
-    pub fn process_chunk(&mut self, chunk: &mut IqChunk) {
-        for z in chunk.samples.iter_mut() {
-            let i = z.re;
-            let q = z.im;
+    pub fn process_sample(&mut self, sample: IqSample) -> IqSample {
+        let i = sample.re;
+        let q = sample.im;
 
-            // Instant Power
-            let inst_power = i * i + q * q;
+        // Instant Power
+        let inst_power = i * i + q * q;
 
-            let alpha = if inst_power > self.power_estimate {
-                self.alpha_attack
-            } else {
-                self.alpha_release
-            };
+        let alpha = if inst_power > self.power_estimate {
+            self.alpha_attack
+        } else {
+            self.alpha_release
+        };
 
-            // IIR filter
-            let power_estimate = alpha * inst_power + (1.0 - alpha) * self.power_estimate;
+        // IIR filter
+        let power_estimate = alpha * inst_power + (1.0 - alpha) * self.power_estimate;
 
-            // Update Power
-            self.power_estimate = power_estimate.max(1e-10);
+        // Update Power
+        self.power_estimate = power_estimate.max(1e-10);
 
-            let raw_gain = (self.target_power / self.power_estimate).sqrt();
+        let raw_gain = (self.target_power / self.power_estimate).sqrt();
 
-            // Gain in [min_gain ; max_gain]
-            let raw_gain = match raw_gain {
-                g if g < self.min_gain => self.min_gain,
-                g if g > self.max_gain => self.max_gain,
-                _ => raw_gain,
-            };
+        // Gain in [min_gain ; max_gain]
+        let raw_gain = match raw_gain {
+            g if g < self.min_gain => self.min_gain,
+            g if g > self.max_gain => self.max_gain,
+            _ => raw_gain,
+        };
 
-            let final_gain = self.beta * self.current_gain + (1.0 - self.beta) * raw_gain;
+        let final_gain = self.beta * self.current_gain + (1.0 - self.beta) * raw_gain;
 
-            self.current_gain = final_gain;
+        self.current_gain = final_gain;
 
-            *z = IqSample::new(i * final_gain, q * final_gain);
-        }
+        IqSample::new(i * final_gain, q * final_gain)
     }
 }
 
 impl<I, E> Iterator for Agc<I>
 where
-    I: Iterator<Item = Result<IqChunk, E>>,
+    I: Iterator<Item = Result<IqSample, E>>,
 {
-    type Item = Result<IqChunk, E>;
+    type Item = Result<IqSample, E>;
 
     fn next(&mut self) -> Option<Self::Item> {
         match self.inner.next()? {
-            Ok(mut chunk) => {
-                self.process_chunk(&mut chunk);
-                Some(Ok(chunk))
-            }
+            Ok(mut sample) => Some(Ok(self.process_sample(sample))),
             Err(e) => Some(Err(e)),
         }
     }

src/fir.rs

@@ -1,5 +1,5 @@
 // Finite Impulse response + Decimation
-use crate::{iq_reader::IqChunk, iq_reader::IqSample, utils::ring_buffer::RingBuffer};
+use crate::{iq_reader::IqSample, utils::ring_buffer::RingBuffer};
 
 pub struct Fir<I, const N: usize> {
     inner: I,
@@ -27,40 +27,43 @@ impl<I, const N: usize> Fir<I, N> {
         }
     }
 
-    pub fn process_chunk(&mut self, chunk: &IqChunk) -> IqChunk {
-        let mut chunk_out = IqChunk::new();
+    // None if not emited
+    pub fn process_sample(&mut self, sample: IqSample) -> Option<IqSample> {
+        self.history.push(sample);
 
-        for iq in chunk.samples.iter() {
-            self.history.push(*iq);
+        self.decimation_index += 1;
 
-            // Decimation
-            if self.decimation_index.is_multiple_of(self.decimation_factor) {
-                let mut y_n = IqSample::default();
-                for k in 0..N {
-                    if let Some(sample) = self.history.read_at(k) {
-                        y_n += *sample * self.taps[k];
-                    }
+        // Decimation
+        if self.decimation_index.is_multiple_of(self.decimation_factor) {
+            let mut y_n = IqSample::default();
+            for k in 0..N {
+                if let Some(s) = self.history.read_at(k) {
+                    y_n += *s * self.taps[k];
                 }
-
-                chunk_out.samples.push(y_n);
             }
-
-            self.decimation_index += 1;
+            Some(y_n)
+        } else {
+            None
         }
-        chunk_out
     }
 }
 
 impl<I, E, const N: usize> Iterator for Fir<I, N>
 where
-    I: Iterator<Item = Result<IqChunk, E>>,
+    I: Iterator<Item = Result<IqSample, E>>,
 {
-    type Item = Result<IqChunk, E>;
+    type Item = Result<IqSample, E>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        match self.inner.next()? {
-            Ok(chunk) => Some(Ok(self.process_chunk(&chunk))),
-            Err(e) => Some(Err(e)),
+        loop {
+            match self.inner.next()? {
+                Ok(sample) => {
+                    if let Some(y_n) = self.process_sample(sample) {
+                        return Some(Ok(y_n));
+                    }
+                }
+                Err(e) => return Some(Err(e)),
+            }
         }
     }
 }

src/fm_demod.rs

@@ -1,5 +1,3 @@
-use std::{os::unix::process, slice::Chunks};
-
 use crate::iq_reader::{IqChunk, IqSample};
 
 pub struct FmDemod<I> {
@@ -15,7 +13,7 @@ impl<I> FmDemod<I> {
         }
     }
 
-    pub fn process_chunk(&mut self, chunk: &mut IqChunk) {
+    pub fn process_sample(&mut self, sample: IqSample) -> IqSample {
         // TODO: FM Demodulation
         todo!();
     }
@@ -23,16 +21,13 @@ impl<I> FmDemod<I> {
 
 impl<I, E> Iterator for FmDemod<I>
 where
-    I: Iterator<Item = Result<IqChunk, E>>,
+    I: Iterator<Item = Result<IqSample, E>>,
 {
-    type Item = Result<IqChunk, E>;
+    type Item = Result<IqSample, E>;
 
     fn next(&mut self) -> Option<Self::Item> {
         match self.inner.next()? {
-            Ok(mut chunk) => {
-                self.process_chunk(&mut chunk);
-                Some(Ok(chunk))
-            }
+            Ok(sample) => Some(Ok(self.process_sample(sample))),
             Err(e) => Some(Err(e)),
         }
     }

src/iq_reader.rs

@@ -26,6 +26,8 @@ pub struct FileSource {
     pub raw_buffer: Vec<u8>,
     // Size of a sample chunk
     pub chunk_samples_size: usize,
+    // raw_buffer position (in samples)
+    pub cursor: usize,
 }
 
 impl FileSource {
@@ -42,40 +44,38 @@ impl FileSource {
             reader,
             raw_buffer: raw_reader,
             chunk_samples_size,
+            cursor: chunk_samples_size,
         })
     }
 }
 
 impl Iterator for FileSource {
-    type Item = Result<IqChunk, Box<dyn Error>>;
+    type Item = Result<IqSample, Box<dyn Error>>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        // Buffer read
-        match self.reader.read_exact(&mut self.raw_buffer) {
-            Ok(_) => {}
+        if self.cursor >= self.chunk_samples_size {
+            // Buffer read chunk
+            match self.reader.read_exact(&mut self.raw_buffer) {
+                Ok(_) => {}
 
-            // EOF
-            Err(e) if e.kind() == ErrorKind::UnexpectedEof => {
-                return None;
-            }
+                // EOF
+                Err(e) if e.kind() == ErrorKind::UnexpectedEof => {
+                    return None;
+                }
 
-            Err(e) => {
-                return Some(Err(e.into()));
+                Err(e) => {
+                    return Some(Err(e.into()));
+                }
             }
+            self.cursor = 0;
         }
 
-        // Output samples
-        let mut samples = Vec::with_capacity(self.chunk_samples_size);
+        // Buffer read sample
+        let i = (self.raw_buffer[self.cursor * 2] as i8) as f32 / 128.0;
+        let q = (self.raw_buffer[self.cursor * 2 + 1] as i8) as f32 / 128.0;
 
-        // Buffer read
-        for iq in self.raw_buffer.chunks(2) {
-            let i = iq[0] as i8;
-            let q = iq[1] as i8;
-            let i_f32 = (i as f32) / 128.0;
-            let q_f32 = (q as f32) / 128.0;
-            samples.push(Complex::new(i_f32, q_f32));
-        }
+        self.cursor += 1;
 
-        Some(Ok(IqChunk { samples }))
+        Some(Ok(IqSample::new(i, q)))
     }
 }

src/main.rs

@@ -19,9 +19,9 @@ fn main() -> Result<(), Box<dyn Error>> {
         .agc(20_000_000.0, 1.0, 0.001, 100.0)
         .fir::<64>(taps, 4);
 
-    for chunk_r in pipeline {
-        let chunk = chunk_r?;
-        println!("size : {}", chunk.samples.len());
+    for sample_r in pipeline {
+        let sample = sample_r?;
+        println!("sample : {} + i{}", sample.re, sample.im);
     }
 
     Ok(())

src/pipeline.rs

@@ -1,8 +1,8 @@
 use crate::agc::Agc;
 use crate::fir::Fir;
-use crate::iq_reader::IqChunk;
+use crate::iq_reader::IqSample;
 
-pub trait DspPipelineExt<E>: Iterator<Item = Result<IqChunk, E>> + Sized {
+pub trait DspPipelineExt<E>: Iterator<Item = Result<IqSample, E>> + Sized {
     fn agc(self, sample_rate: f32, target_power: f32, min_gain: f32, max_gain: f32) -> Agc<Self> {
         Agc::new(self, sample_rate, target_power, min_gain, max_gain)
     }
@@ -12,4 +12,4 @@ pub trait DspPipelineExt<E>: Iterator<Item = Result<IqChunk, E>> + Sized {
     }
 }
 
-impl<I, E> DspPipelineExt<E> for I where I: Iterator<Item = Result<IqChunk, E>> {}
+impl<I, E> DspPipelineExt<E> for I where I: Iterator<Item = Result<IqSample, E>> {}