octagonal/oxydsp
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Working fsk transmitter

Browse Source
This commit is contained in:
Albin Chaboissier
2026-03-22 19:34:21 +01:00
parent f1f769e0e6
commit 6429685cd2
16 changed files with 921 additions and 406 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
example/Cargo.toml
View File

@ -11,3 +11,5 @@ egui_plot = "0.34.1"
eframe = { version = "0.33.3", features = ["default_fonts", "wayland"] }
num = "0.4.3"
hound = "3.5.1"
rand = "0.10.0"
cpal = "0.17.3"

BIN
example/mod.wav
View File

Binary file not shown.

30
example/out.dot
View File

@ -3,21 +3,23 @@
node [shape=record];
rankdir=TB;
IterSource_0 [label="{ IterSource |{<o0> output} }"];
Map_1 [label="{ {<i0> input}| Map |{<o0> output} }"];
Repeat_2 [label="{ {<i0> input}| Repeat |{<o0> output} }"];
Nco_3 [label="{ {<i0> frequency}| Nco |{<o0> output} }"];
OscillatorSource_4 [label="{ OscillatorSource |{<o0> output} }"];
Multiplier_5 [label="{ {<i0> input_a|<i1> input_b}| Multiplier |{<o0> output} }"];
MapResultTagged_6 [label="{ {<i0> input}| MapResultTagged |{<o0> output} }"];
NullSink_7 [label="{ {<i0> input}| NullSink }"];
FirFilter_1 [label="{ {<i0> input}| FirFilter |{<o0> output} }"];
Map_2 [label="{ {<i0> input}| Map |{<o0> output} }"];
Repeat_3 [label="{ {<i0> input}| Repeat |{<o0> output} }"];
Nco_4 [label="{ {<i0> frequency}| Nco |{<o0> output} }"];
OscillatorSource_5 [label="{ OscillatorSource |{<o0> output} }"];
Multiplier_6 [label="{ {<i0> input_a|<i1> input_b}| Multiplier |{<o0> output} }"];
MapResultTagged_7 [label="{ {<i0> input}| MapResultTagged |{<o0> output} }"];
NullSink_8 [label="{ {<i0> input}| NullSink }"];
IterSource_0:o0 -> Map_1:i0 [label="bool"];
Map_1:o0 -> Repeat_2:i0 [label="oxydsp_dsp::units::DigitalFrequency"];
Repeat_2:o0 -> Nco_3:i0 [label="oxydsp_dsp::units::DigitalFrequency"];
Nco_3:o0 -> Multiplier_5:i0 [label="num_complex::Complex<f32>"];
OscillatorSource_4:o0 -> Multiplier_5:i1 [label="num_complex::Complex<f32>"];
Multiplier_5:o0 -> MapResultTagged_6:i0 [label="num_complex::Complex<f32>"];
MapResultTagged_6:o0 -> NullSink_7:i0 [label="num_complex::Complex<f32>"];
IterSource_0:o0 -> Repeat_3:i0 [label="f32"];
FirFilter_1:o0 -> Map_2:i0 [label="f32"];
Map_2:o0 -> Nco_4:i0 [label="oxydsp_dsp::units::DigitalFrequency"];
Repeat_3:o0 -> FirFilter_1:i0 [label="f32"];
Nco_4:o0 -> Multiplier_6:i0 [label="num_complex::Complex<f32>"];
OscillatorSource_5:o0 -> Multiplier_6:i1 [label="num_complex::Complex<f32>"];
Multiplier_6:o0 -> MapResultTagged_7:i0 [label="num_complex::Complex<f32>"];
MapResultTagged_7:o0 -> NullSink_8:i0 [label="num_complex::Complex<f32>"];
}

376
example/src/main.rs
View File

@ -1,394 +1,62 @@
use std::fmt::Display;
use std::collections::VecDeque;
use std::fs::File;
use std::io;
use std::io::Write;
use std::sync::mpsc;
use cpal::traits::DeviceTrait;
use cpal::traits::HostTrait;
use eframe::NativeOptions;
use egui::Color32;
use egui_plot::Line;
use egui_plot::MarkerShape;
use egui_plot::PlotPoints;
use egui_plot::Points;
use egui_plot::Polygon;
use egui_plot::VLine;
use num::Complex;
use num::Zero;
use oxydsp_dsp::blocks::filtering::fir::FirFilter;
use oxydsp_dsp::blocks::iq::zero_if::ZeroIf;
use oxydsp_dsp::blocks::math::basic::Adder;
use oxydsp_dsp::blocks::math::basic::Multiplier;
use oxydsp_dsp::blocks::synthesis::Nco;
use oxydsp_dsp::blocks::synthesis::OscillatorSource;
use oxydsp_dsp::blocks::ted::early_late::EarlyLateGate;
use oxydsp_dsp::blocks::utilities::adapters::Map;
use oxydsp_dsp::blocks::utilities::adapters::MapResult;
use oxydsp_dsp::blocks::utilities::adapters::MapResultTagged;
use oxydsp_dsp::blocks::utilities::adapters::NullSink;
use oxydsp_dsp::blocks::utilities::adapters::Repeat;
use oxydsp_dsp::blocks::utilities::adapters::Scan;
use oxydsp_dsp::blocks::utilities::channels::TxSink;
use oxydsp_dsp::blocks::utilities::adapters::ScanTagged;
use oxydsp_dsp::blocks::utilities::channels::RxSource;
use oxydsp_dsp::blocks::utilities::iter::IterSource;
use oxydsp_dsp::blocks::utilities::squelch::Squelch;
use oxydsp_dsp::filtering::fir::Fir;
use oxydsp_dsp::units::DigitalFrequency;
use oxydsp_flowgraph::BlockIO;
use oxydsp_flowgraph::block::Block;
use oxydsp_flowgraph::block::BlockResult;
use oxydsp_flowgraph::block::SyncBlock;
use oxydsp_flowgraph::flowgraph;
use oxydsp_flowgraph::graph::FlowGraph;
use oxydsp_flowgraph::io::In;
use oxydsp_flowgraph::io::Out;
use oxydsp_flowgraph::io::PopIterable;
use oxydsp_flowgraph::io::stream;
use oxydsp_flowgraph::sync_block;
use oxydsp_flowgraph::tag::Tag;
use oxydsp_flowgraph::tag::Tagged;
use rand::random;
#[derive(BlockIO)]
//#[sync_block(tagged)]
pub struct Printer<T: 'static + Display>
{
#[input]
input: In<T>,
use crate::transmitter::Transmitter;
n: usize,
}
impl<T: 'static + Display> Block for Printer<T>
{
fn work(&mut self) -> BlockResult
{
for x in self.input.pop_iter()
{
if self.n.is_multiple_of(100_000)
{
if x.has_tag()
{
let tag = x.1.unwrap();
let valuea: usize = *tag.retrieve("valuea").unwrap().downcast().unwrap();
let valueb: usize = *tag.retrieve("valueb").unwrap().downcast().unwrap();
println!(
"{} TAGGED {}, {} amoziefjmoazijfmoazeijfmoazeifjmozeijfmoizfmojzaemfojzaemofjzeamofimazoijefmzoaeijfmoazeifj",
x.0, valuea, valueb
);
}
else
{
println!("{} NO TAG", x.0);
}
}
self.n += 1
}
BlockResult::Ok
}
}
// impl<'view, T: 'static + Display> SyncBlock<'view> for Printer<T>
// {
// fn sync_work(state: Self::StateView, input: Self::Input) -> Option<Self::Output>
// {
// // if state.n.is_multiple_of(100_000)
// // {
// if input.has_tag()
// {
// println!("{} TAGGED", input.0);
// }
// else
// {
// println!("{} NO TAG", input.0);
// }
// //}
// *state.n += 1;
// Some(())
// }
// }
#[derive(BlockIO)]
//#[sync_block(tagged)]
pub struct SourceTag
{
#[output]
output: Out<usize>,
tk: String,
n: usize,
}
impl SourceTag
{
pub fn new(tk: String) -> (Self, In<usize>)
{
let (output, input) = stream();
(Self { output, tk, n: 0 }, input)
}
}
// impl<'view> SyncBlock<'view> for SourceTag
// {
// fn sync_work(state: Self::StateView, _input: Self::Input) -> Option<Self::Output>
// {
// *state.n += 1;
//
// let t = Tag::default();
// Some((*state.n, t).into())
// }
// }
impl Block for SourceTag
{
fn work(&mut self) -> BlockResult
{
self.output.push_iter((0usize..).map(|_| {
self.n += 1;
(
self.n - 1,
if (self.n - 1).is_multiple_of(1_000)
{
let tag = Tag::default();
tag.tag(&self.tk, self.n);
Some(tag)
}
else
{
None
},
)
.into()
}));
BlockResult::Ok
}
}
// mod printer_synchronous_block
// {
// struct PrinterView<'view, T>
// {
// n: &'view mut usize,
// _sync_block_phantom: std::marker::PhantomData<'view, T>,
// }
// impl<'view, T: 'static> oxydsp_flowgraph::block::SyncBlockIO<'view> for super::Printer<T>
// {
// type StateView = PrinterView<'view, T>;
// type Input = ();
// type Output = ();
// }
// }
impl<T: 'static + Display> Printer<T>
{
pub fn new(input: In<T>) -> Self
{
Self { input, n: 0 }
}
}
pub mod receiver;
pub mod transmitter;
fn main()
{
main_demod()
}
let tx = Transmitter::start_new();
fn main_tst()
{
let (sourcea, a) = SourceTag::new("valuea".to_string());
let (sourceb, b) = SourceTag::new("valueb".to_string());
let (adder, a) = Adder::new(a, b);
let printer = Printer::new(a);
let fg = flowgraph![sourcea, sourceb, adder, printer];
let _ = fg.run().join();
}
const SAMPLE_RATE: usize = 48_000;
const SAMPLE_PER_SYMBOL: usize = 96;
const DEVIATION: f64 = 500.;
const CARRIER: f64 = 1000.;
fn main_demod()
{
let carrier = DigitalFrequency::from_time_frequency(CARRIER, SAMPLE_RATE as f64);
let mut reader = hound::WavReader::open("mod.wav").unwrap();
let sqr_sum = reader
.samples::<i16>()
.map(|sample| (sample.unwrap() as f32) / (i16::MAX as f32))
.collect::<Vec<_>>();
let (iter_source, signal) = IterSource::new(sqr_sum.into_iter());
// Make an iq sampler
let (lo, lo_signal) = OscillatorSource::new(carrier.into());
let (mixer, iq) = Multiplier::new(signal, lo_signal);
let (iq_bandpass, iq) = FirFilter::<Complex<f32>, Complex<f32>, Complex<f32>>::new(
iq,
Fir::lowpass(carrier, 100).normalized(),
);
let (arg_extract, arg) = Scan::new(iq, Complex::zero(), |state, sample| {
let angle = *state / sample;
*state = sample;
angle.arg()
});
let (sig_lowpass, arg) =
FirFilter::<f32, f32, f32>::new(arg, Fir(vec![1.; SAMPLE_PER_SYMBOL]).normalized());
let (tx, rx) = mpsc::channel();
let elg_loop = Fir(vec![1.0f32; 10]);
let mut elg_loop = elg_loop.normalized();
elg_loop.0[0] = 0.5;
let (elg, arg) = EarlyLateGate::new(arg, elg_loop, SAMPLE_PER_SYMBOL);
let (sender, arg) = MapResultTagged::new(arg, move |x| {
let _ = tx.send((
x.0,
x.1.as_ref()
.is_some_and(|t| t.retrieve("elg_symbol").is_some()),
));
if x.1
.is_some_and(|t| t.retrieve("itersource_finished").is_some())
{
println!("FINISHED !");
(x.0.into(), BlockResult::Exit)
}
else
{
(x.0.into(), BlockResult::Ok)
}
});
let null_sink = NullSink::new(arg);
let graph = flowgraph![
iter_source,
lo,
mixer,
iq_bandpass,
arg_extract,
sig_lowpass,
elg,
sender,
null_sink
];
let j = graph.run();
let mut output = vec![];
while let Ok(x) = rx.recv()
loop
{
output.push(x);
let mut user_input = String::new();
io::stdin().read_line(&mut user_input).unwrap();
println!("Transmitting ...");
tx.transmit(user_input.as_bytes().to_vec());
}
let _ = j.join();
eframe::run_simple_native("Plot", NativeOptions::default(), move |ctx, _frame| {
egui::CentralPanel::default().show(ctx, |ui| {
egui_plot::Plot::new("hello").show(ui, |plot_ui| {
plot_ui.line(Line::new(
"samples",
output
.iter()
.enumerate()
.map(|(i, s)| [i as f64, s.0 as f64])
.collect::<PlotPoints>(),
));
plot_ui.points(
Points::new(
"symbols",
output
.iter()
.enumerate()
.filter(|(_, (_, x))| *x)
.map(|(i, (s, _))| [i as f64, *s as f64])
.collect::<Vec<_>>(),
)
.id("symbols")
.radius(5.)
.shape(MarkerShape::Diamond),
);
});
ctx.request_repaint();
});
})
.unwrap();
}
fn main_mod()
{
let carrier = DigitalFrequency::from_time_frequency(CARRIER, SAMPLE_RATE as f64);
let deviation = DigitalFrequency::from_time_frequency(DEVIATION, SAMPLE_RATE as f64);
let data = (0..255u8).flat_map(to_bits).collect::<Vec<_>>();
let (bit_stream, bits) = IterSource::new(data.into_iter());
let (to_freq, freq) = Map::new(bits, move |x| [-deviation, deviation][x as usize]);
let (repeat, freq) = Repeat::new(freq, SAMPLE_PER_SYMBOL);
let (base_oscillator, baseband) = Nco::<f32>::new(freq);
let (local_oscillator, lo) = OscillatorSource::<f32>::new(carrier.into());
let (frontend, passband) = Multiplier::new(baseband, lo);
let (tx, rx) = mpsc::channel::<Complex<f32>>();
let (sender, passband) = MapResultTagged::new(passband, move |x| {
let _ = tx.send(x.0);
if x.1
.is_some_and(|t| t.retrieve("itersource_finished").is_some())
{
println!("FINISHED !");
(x.0.into(), BlockResult::Exit)
}
else
{
(x.0.into(), BlockResult::Ok)
}
});
let null_sink = NullSink::new(passband);
let graph = flowgraph![
bit_stream,
to_freq,
repeat,
base_oscillator,
local_oscillator,
frontend,
sender,
null_sink,
];
File::create("out.dot")
.unwrap()
.write_all(graph.get_dot().as_bytes())
.unwrap();
let j = graph.run();
let mut output = vec![];
while let Ok(x) = rx.recv()
{
output.push(x);
}
let _ = j.join();
// Write signal
let spec = hound::WavSpec {
channels: 1,
sample_rate: SAMPLE_RATE as u32,
bits_per_sample: 16,
sample_format: hound::SampleFormat::Int,
};
let mut writer = hound::WavWriter::create("mod.wav", spec).unwrap();
for x in output.iter()
{
let amplitude = i16::MAX as f32;
writer.write_sample((x.re * amplitude) as i16).unwrap();
}
writer.finalize().unwrap();
//
eframe::run_simple_native("Plot", NativeOptions::default(), move |ctx, _frame| {
egui::CentralPanel::default().show(ctx, |ui| {
egui_plot::Plot::new("hello").show(ui, |plot_ui| {
plot_ui.line(Line::new(
"samples",
output
.iter()
.enumerate()
.map(|(i, s)| [i as f64, s.re as f64])
.collect::<PlotPoints>(),
));
});
ctx.request_repaint();
});
})
.unwrap();
}
pub const SAMPLE_RATE: usize = 48_000;
pub const SAMPLE_PER_SYMBOL: usize = 96;
pub const DEVIATION: f64 = 500.;
pub const CARRIER: f64 = 1700.;
pub fn to_bits(n: u8) -> [bool; 8]
{

178
example/src/receiver.rs Normal file
View File

@ -0,0 +1,178 @@
use std::collections::VecDeque;
use std::sync::mpsc;
use cpal::traits::DeviceTrait;
use cpal::traits::HostTrait;
use eframe::NativeOptions;
use egui::Color32;
use egui_plot::Line;
use egui_plot::PlotPoints;
use num::Complex;
use num::Zero;
use oxydsp_dsp::blocks::filtering::fir::FirFilter;
use oxydsp_dsp::blocks::iq::zero_if::ZeroIf;
use oxydsp_dsp::blocks::ted::early_late::EarlyLateGate;
use oxydsp_dsp::blocks::utilities::adapters::NullSink;
use oxydsp_dsp::blocks::utilities::adapters::Scan;
use oxydsp_dsp::blocks::utilities::adapters::ScanTagged;
use oxydsp_dsp::blocks::utilities::channels::RxSource;
use oxydsp_dsp::blocks::utilities::squelch::Squelch;
use oxydsp_dsp::filtering::fir::Fir;
use oxydsp_dsp::units::DigitalFrequency;
use oxydsp_flowgraph::flowgraph;
use oxydsp_flowgraph::graph::FlowGraph;
use crate::CARRIER;
use crate::DEVIATION;
use crate::SAMPLE_PER_SYMBOL;
use crate::SAMPLE_RATE;
fn main_demod()
{
let carrier = DigitalFrequency::from_time_frequency(CARRIER, SAMPLE_RATE as f64);
let mut reader = hound::WavReader::open("mod.wav").unwrap();
let sqr_sum = reader
.samples::<i16>()
.map(|sample| (sample.unwrap() as f32) / (i16::MAX as f32))
.collect::<Vec<_>>();
let (audio_tx, audio_rx) = mpsc::channel();
//let (source, signal) = IterSource::new(sqr_sum.into_iter().cycle());
let (source, signal) = RxSource::new(audio_rx);
let (mut zero_if, iq) = ZeroIf::new(signal, carrier.into());
zero_if.set_fir(Fir::lowpass(
DigitalFrequency::from_time_frequency(DEVIATION * 2. + 100., SAMPLE_RATE as f64),
200,
));
let (squelch, iq) = Squelch::new(iq, 5., 100);
let (arg_extract, arg) = Scan::new(iq, Complex::zero(), |state, sample| {
let angle: Complex<f32> = sample / *state;
*state = sample;
angle.arg() * 14.
});
let (sig_lowpass, arg) =
FirFilter::<f32, f32, f32>::new(arg, Fir(vec![1.; SAMPLE_PER_SYMBOL]).normalized());
let mut elg_loop = Fir(vec![1. / 30.; 30]);
//let mut elg_loop = elg_loop.normalized();
*elg_loop.0.last_mut().unwrap() = 0.4;
//*elg_loop.0.first_mut().unwrap() = 0.001;
let (elg, arg) = EarlyLateGate::new(arg, elg_loop, SAMPLE_PER_SYMBOL);
// Eye diagram
let (tx, rx) = mpsc::channel();
let (eye_sender, arg) = ScanTagged::new(arg, VecDeque::new(), move |history, x| {
if history.len() == SAMPLE_PER_SYMBOL
{
history.pop_back();
}
let mut error: f32 = 0.;
let is_symbol_center = x.1.as_ref().is_some_and(|t| {
if let Some(err) = t.retrieve("elg_symbol")
{
error = *err.downcast().unwrap();
true
}
else
{
false
}
});
history.push_front(((is_symbol_center, error), x.0));
if history.len() > SAMPLE_PER_SYMBOL / 2 && history[SAMPLE_PER_SYMBOL / 2].0.0
{
let _ = tx.send((
history.iter().map(|(_, x)| *x).collect::<Vec<_>>(),
history[SAMPLE_PER_SYMBOL / 2].0.1,
));
}
x.0.into()
});
let null_sink = NullSink::new(arg);
let graph = flowgraph![
source,
squelch,
zero_if,
arg_extract,
sig_lowpass,
elg,
eye_sender,
null_sink
];
// Setup input
let host = cpal::default_host();
let device = host.default_input_device().expect("No input device");
let mut supported_configs_range = device
.supported_input_configs()
.expect("error while querying configs");
let supported_config = supported_configs_range
.next()
.expect("no supported config?!")
.with_sample_rate(SAMPLE_RATE as u32);
let stream = device.build_input_stream(
&supported_config.into(),
move |data: &[f32], _: &cpal::InputCallbackInfo| {
for x in data.iter()
{
let _ = audio_tx.send(*x);
}
},
move |err| {
panic!() // react to errors here.
},
None, // None=blocking, Some(Duration)=timeout
);
let j = graph.run();
let mut eyes = VecDeque::new();
eframe::run_simple_native("Plot", NativeOptions::default(), move |ctx, _frame| {
while let Ok(eye) = rx.try_recv()
{
if eyes.len() >= 100
{
let _ = eyes.pop_back();
}
eyes.push_front(eye);
}
egui::CentralPanel::default().show(ctx, |ui| {
egui_plot::Plot::new("hello").show(ui, |plot_ui| {
for eye in eyes.iter()
{
plot_ui.line(
Line::new(
"eyes",
eye.0
.iter()
.enumerate()
.map(|(i, s)| [i as f64, *s as f64])
.collect::<PlotPoints>(),
)
.id("eyes")
.color(color_from_err(eye.1, 0.3)),
);
}
});
ctx.request_repaint();
});
})
.unwrap();
}
pub fn color_from_err(error: f32, max: f32) -> Color32
{
Color32::RED
.linear_multiply(error.abs() / max)
.blend(Color32::GREEN.linear_multiply((1. - error.abs() / max).max(0.)))
}

268
example/src/transmitter.rs Normal file
View File

@ -0,0 +1,268 @@
use std::collections::VecDeque;
use std::fs::File;
use std::io::Write;
use std::iter::FusedIterator;
use std::ops::BitXor;
use std::sync::mpsc;
use std::sync::mpsc::Receiver;
use std::sync::mpsc::Sender;
use std::sync::mpsc::SyncSender;
use std::sync::mpsc::sync_channel;
use std::thread::JoinHandle;
use cpal::Stream;
use cpal::traits::DeviceTrait;
use cpal::traits::HostTrait;
use eframe::NativeOptions;
use egui::Color32;
use egui::output;
use egui_plot::Line;
use egui_plot::PlotPoints;
use num::Complex;
use num::Zero;
use oxydsp_dsp::blocks::filtering::fir::FirFilter;
use oxydsp_dsp::blocks::iq::zero_if::ZeroIf;
use oxydsp_dsp::blocks::math::basic::Adder;
use oxydsp_dsp::blocks::math::basic::Multiplier;
use oxydsp_dsp::blocks::synthesis::Nco;
use oxydsp_dsp::blocks::synthesis::OscillatorSource;
use oxydsp_dsp::blocks::ted::early_late::EarlyLateGate;
use oxydsp_dsp::blocks::utilities::adapters::Map;
use oxydsp_dsp::blocks::utilities::adapters::MapResultTagged;
use oxydsp_dsp::blocks::utilities::adapters::NullSink;
use oxydsp_dsp::blocks::utilities::adapters::Repeat;
use oxydsp_dsp::blocks::utilities::adapters::Scan;
use oxydsp_dsp::blocks::utilities::adapters::ScanTagged;
use oxydsp_dsp::blocks::utilities::channels::RxSource;
use oxydsp_dsp::blocks::utilities::channels::TxSink;
use oxydsp_dsp::blocks::utilities::iter::IterSource;
use oxydsp_dsp::blocks::utilities::squelch::Squelch;
use oxydsp_dsp::filtering::fir::Fir;
use oxydsp_dsp::units::DigitalFrequency;
use oxydsp_flowgraph::BlockIO;
use oxydsp_flowgraph::block::Block;
use oxydsp_flowgraph::block::BlockResult;
use oxydsp_flowgraph::flowgraph;
use oxydsp_flowgraph::graph::FlowGraph;
use oxydsp_flowgraph::io::In;
use oxydsp_flowgraph::io::Out;
use rand::random;
use crate::CARRIER;
use crate::DEVIATION;
use crate::SAMPLE_PER_SYMBOL;
use crate::SAMPLE_RATE;
use crate::gaussian;
use crate::to_bits;
#[derive(BlockIO)]
pub struct FlatMap<I, O, F>
where
I: 'static,
O: IntoIterator + 'static,
O::IntoIter: FusedIterator,
F: Fn(I) -> O,
{
#[input]
input: In<I>,
#[output]
output: Out<O::Item>,
current_iter: Option<O::IntoIter>,
map: F,
}
impl<I, O, F> FlatMap<I, O, F>
where
I: 'static,
O: IntoIterator + 'static,
O::IntoIter: FusedIterator,
F: Fn(I) -> O,
{
pub fn new(input: In<I>, map: F) -> (Self, In<O::Item>)
{
let (output, port) = oxydsp_flowgraph::io::stream();
(
Self {
input,
output,
current_iter: None,
map,
},
port,
)
}
}
impl<I, O, F> Block for FlatMap<I, O, F>
where
I: 'static,
O: IntoIterator + 'static,
O::IntoIter: FusedIterator,
F: Fn(I) -> O,
{
fn work(&mut self) -> BlockResult
{
let writer = self.output.write();
let reader = self.input.read();
let max_write = writer.len();
let mut written = 0;
while written < max_write
{
if let Some(current_iter) = self.current_iter.as_mut()
{
if let Some(next_elt) = current_iter.next()
{
let _ = writer.push((next_elt, None).into());
written += 1;
continue;
}
else
{
// Iterator empty
self.current_iter = None;
}
}
if self.current_iter.is_none()
{
// Get input
if let Some(input) = reader.pop()
{
let mut new_iter = (self.map)(input.0).into_iter();
if let Some(first_elt) = new_iter.next()
{
self.current_iter = Some(new_iter);
let _ = writer.push((first_elt, input.1).into());
written += 1;
}
else
{
// Iterator empty
self.current_iter = None;
continue;
}
}
else
{
// Cannot continue
break;
}
}
}
BlockResult::Ok
}
}
pub struct Transmitter
{
flowgraph_handle: JoinHandle<()>,
packet_sender: SyncSender<Vec<u8>>,
stream: Stream,
}
impl Transmitter
{
pub fn start_new() -> Self
{
let carrier = DigitalFrequency::from_time_frequency(CARRIER, SAMPLE_RATE as f64);
let deviation = DigitalFrequency::from_time_frequency(DEVIATION, SAMPLE_RATE as f64);
let (packet_tx, packet_rx): (_, Receiver<Vec<u8>>) = sync_channel(128);
let (packet_rec, packets): (_, In<Vec<u8>>) = RxSource::new(packet_rx);
let (linearizer, bits) = FlatMap::new(packets, |packet| {
// +1 for chksum
let packet_length = (packet.len() + 1) as u16;
let checksum = packet.iter().copied().reduce(BitXor::bitxor).unwrap();
// Learning sequence
let mut frame = vec![0b10101010; 8];
frame.push(packet_length.to_le_bytes()[0]);
frame.push(packet_length.to_le_bytes()[1]);
frame.extend(packet.iter());
frame.push(checksum);
frame
.into_iter()
.flat_map(to_bits)
.map(|x| if x { 1. } else { -1.0f32 })
});
let (repeat, bits) = Repeat::new(bits, SAMPLE_PER_SYMBOL);
// gaussian fir
let fir = Fir((0..SAMPLE_PER_SYMBOL)
.map(|x| gaussian(0.3, x as f32 / SAMPLE_PER_SYMBOL as f32))
.collect())
.normalized();
let (bit_filter, bits) = FirFilter::new(bits, fir);
let (to_freq, freq) = Map::new(bits, move |x| {
DigitalFrequency::from_time_frequency(DEVIATION * x as f64, SAMPLE_RATE as f64)
});
let (base_oscillator, baseband) = Nco::<f32>::new(freq);
let (local_oscillator, lo) = OscillatorSource::<f32>::new(carrier.into());
let (frontend, passband) = Multiplier::new(baseband, lo);
let (audio_tx, audio_rx) = mpsc::channel::<Complex<f32>>();
let tx_sink = TxSink::new(passband, audio_tx);
let graph = flowgraph![
packet_rec,
linearizer,
bit_filter,
to_freq,
repeat,
base_oscillator,
local_oscillator,
frontend,
tx_sink,
];
// Open output device
let host = cpal::default_host();
let device = host
.default_output_device()
.expect("no output device available");
let mut supported_configs_range = device
.supported_output_configs()
.expect("error while querying configs");
let supported_config = supported_configs_range
.next()
.expect("no supported config?!")
.with_sample_rate(SAMPLE_RATE as u32);
let stream = device
.build_output_stream(
&supported_config.into(),
move |data: &mut [f32], _: &cpal::OutputCallbackInfo| {
for x in data.iter_mut()
{
if let Ok(y) = audio_rx.try_recv()
{
*x = y.re;
}
else
{
*x = 0.;
}
}
},
move |err| panic!(),
None, // None=blocking, Some(Duration)=timeout
)
.unwrap();
Self {
flowgraph_handle: graph.run(),
packet_sender: packet_tx,
stream,
}
}
pub fn transmit(&self, data: Vec<u8>)
{
let _ = self.packet_sender.send(data);
}
}