fm mod init

src/agc.rs

@@ -1,9 +1,9 @@
-use num_complex::Complex;
-
 use crate::iq_reader::IqChunk;
+use crate::iq_reader::IqSample;
 
 // Automatic Gain Control
-pub struct Agc {
+pub struct Agc<I> {
+    inner: I,
     // Previous power estimate
     pub power_estimate: f32,
     // Previous gain
@@ -16,8 +16,14 @@ pub struct Agc {
     pub max_gain: f32,
 }
 
-impl Agc {
-    pub fn new(sample_rate: f32, target_power: f32, min_gain: f32, max_gain: f32) -> Self {
+impl<I> Agc<I> {
+    pub fn new(
+        inner: I,
+        sample_rate: f32,
+        target_power: f32,
+        min_gain: f32,
+        max_gain: f32,
+    ) -> Self {
         // Target attack time 5 ms
         let tau_attack = 0.005;
 
@@ -33,6 +39,7 @@ impl Agc {
         let current_gain = 1.0;
 
         Self {
+            inner,
             power_estimate,
             current_gain,
             target_power,
@@ -77,7 +84,24 @@ impl Agc {
 
             self.current_gain = final_gain;
 
-            *z = Complex::new(i * final_gain, q * final_gain);
+            *z = IqSample::new(i * final_gain, q * final_gain);
+        }
+    }
+}
+
+impl<I, E> Iterator for Agc<I>
+where
+    I: Iterator<Item = Result<IqChunk, E>>,
+{
+    type Item = Result<IqChunk, E>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.inner.next()? {
+            Ok(mut chunk) => {
+                self.process_chunk(&mut chunk);
+                Some(Ok(chunk))
+            }
+            Err(e) => Some(Err(e)),
         }
     }
 }

src/fir.rs

@@ -1,27 +1,66 @@
 // Finite Impulse response + Decimation
-use crate::utils::ring_buffer::RingBuffer;
-use num_complex::Complex32;
+use crate::{iq_reader::IqChunk, iq_reader::IqSample, utils::ring_buffer::RingBuffer};
 
-pub struct Fir<const N: usize> {
+pub struct Fir<I, const N: usize> {
+    inner: I,
     // Filter coefs
     pub taps: [f32; N],
 
     // Ring Buffer of samples
-    pub history: RingBuffer<Complex32>,
+    pub history: RingBuffer<IqSample>,
 
-    decimation_factor: usize,
+    // Factor of decimation
+    pub decimation_factor: usize,
 
-    // When to keep a sample
-    decimator_counter: usize,
+    // Track decimation
+    pub decimation_index: usize,
 }
 
-impl<const N: usize> Fir<N> {
-    fn new(taps: [f32; N], decimation_factor: usize) -> Self {
+impl<I, const N: usize> Fir<I, N> {
+    pub fn new(inner: I, taps: [f32; N], decimation_factor: usize) -> Self {
         Self {
+            inner,
             taps,
-            history: RingBuffer::new(N),
+            history: RingBuffer::<IqSample>::new(N),
             decimation_factor,
-            decimator_counter: 0,
+            decimation_index: 0,
+        }
+    }
+
+    pub fn process_chunk(&mut self, chunk: &IqChunk) -> IqChunk {
+        let mut chunk_out = IqChunk::new();
+
+        for iq in chunk.samples.iter() {
+            self.history.push(*iq);
+
+            // 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];
+                    }
+                }
+
+                chunk_out.samples.push(y_n);
+            }
+
+            self.decimation_index += 1;
+        }
+        chunk_out
+    }
+}
+
+impl<I, E, const N: usize> Iterator for Fir<I, N>
+where
+    I: Iterator<Item = Result<IqChunk, E>>,
+{
+    type Item = Result<IqChunk, 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)),
         }
     }
 }

src/fm_demod.rs

@@ -0,0 +1,39 @@
+use std::{os::unix::process, slice::Chunks};
+
+use crate::iq_reader::{IqChunk, IqSample};
+
+pub struct FmDemod<I> {
+    pub inner: I,
+    pub prev_sample: IqSample,
+}
+
+impl<I> FmDemod<I> {
+    pub fn new(inner: I) -> Self {
+        Self {
+            inner,
+            prev_sample: IqSample::new(1.0, 0.0),
+        }
+    }
+
+    pub fn process_chunk(&mut self, chunk: &mut IqChunk) {
+        // TODO: FM Demodulation
+        todo!();
+    }
+}
+
+impl<I, E> Iterator for FmDemod<I>
+where
+    I: Iterator<Item = Result<IqChunk, E>>,
+{
+    type Item = Result<IqChunk, E>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.inner.next()? {
+            Ok(mut chunk) => {
+                self.process_chunk(&mut chunk);
+                Some(Ok(chunk))
+            }
+            Err(e) => Some(Err(e)),
+        }
+    }
+}

src/iq_reader.rs

@@ -11,6 +11,14 @@ pub struct IqChunk {
     pub samples: Vec<IqSample>,
 }
 
+impl IqChunk {
+    pub fn new() -> Self {
+        IqChunk {
+            samples: Vec::<IqSample>::new(),
+        }
+    }
+}
+
 pub struct FileSource {
     // Buffer
     pub reader: BufReader<File>,

src/main.rs

@@ -1,26 +1,27 @@
-use crate::agc::Agc;
 use crate::iq_reader::FileSource;
+use crate::pipeline::DspPipelineExt;
 use std::error::Error;
 
 mod agc;
 mod fir;
+mod fm_demod;
 mod iq_reader;
+mod pipeline;
 mod utils;
 
 fn main() -> Result<(), Box<dyn Error>> {
     let source = FileSource::new("test.iq", 32769)?;
 
-    // for chunk in source {
-    //     println!("{chunk :?}");
-    // }
+    // Fir coefs
+    let taps = [0.5; 64];
 
-    // 20 MSps
-    let mut agc = Agc::new(20_000_000.0, 0.1, 0.001, 100.0);
+    let pipeline = source
+        .agc(20_000_000.0, 1.0, 0.001, 100.0)
+        .fir::<64>(taps, 4);
 
-    // Apply Auto Gain Control
-    for chunk_r in source {
-        let mut chunk = chunk_r?;
-        agc.process_chunk(&mut chunk);
+    for chunk_r in pipeline {
+        let chunk = chunk_r?;
+        println!("size : {}", chunk.samples.len());
     }
 
     Ok(())

src/pipeline.rs

@@ -0,0 +1,15 @@
+use crate::agc::Agc;
+use crate::fir::Fir;
+use crate::iq_reader::IqChunk;
+
+pub trait DspPipelineExt<E>: Iterator<Item = Result<IqChunk, 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)
+    }
+
+    fn fir<const N: usize>(self, taps: [f32; N], decimation_factor: usize) -> Fir<Self, N> {
+        Fir::new(self, taps, decimation_factor)
+    }
+}
+
+impl<I, E> DspPipelineExt<E> for I where I: Iterator<Item = Result<IqChunk, E>> {}

src/utils/ring_buffer.rs

@@ -21,13 +21,13 @@ impl<T: Copy + Default> RingBuffer<T> {
         }
     }
 
-    pub fn write(&mut self, value: T) {
+    pub fn push(&mut self, value: T) {
         self.data[self.head] = value;
         self.head = (self.head + 1) % self.capacity;
         self.size = (self.size + 1).min(self.capacity);
     }
 
-    pub fn read(&mut self) -> Option<T> {
+    pub fn pop(&mut self) -> Option<T> {
         if self.size == 0 {
             return None;
         }
@@ -55,7 +55,7 @@ impl<T: Copy + Default> RingBuffer<T> {
 
     pub fn write_read(&mut self, value: T, delay: usize) -> Option<T> {
         let delayed = self.read_at(delay).copied();
-        self.write(value);
+        self.push(value);
         delayed
     }