1200 bauds ?

src/main.rs

@@ -12,6 +12,7 @@ mod ted;
 mod units;
 mod windows;
 
+use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
 use egui_plot::{Legend, Line, Plot};
 use hound::WavWriter;
 use rand::{Rng, rand_core::le, seq::index::sample};
@@ -19,18 +20,20 @@ use std::{
     cell::{Cell, RefCell},
     collections::VecDeque,
     env::{self, args},
+    fmt::Display,
     fs::File,
     io::{BufWriter, Read, Sink, Write, stdout},
     ops::DerefMut,
-    sync::{Arc, mpsc::RecvTimeoutError},
+    sync::{Arc, atomic::AtomicU64, mpsc::RecvTimeoutError},
     time::Duration,
+    u64,
 };
 use tokio::{
     io::{self, AsyncReadExt, AsyncWriteExt},
     join,
     net::{TcpSocket, TcpStream, UdpSocket},
     select,
-    sync::mpsc::error::TryRecvError,
+    sync::mpsc::{UnboundedSender, error::TryRecvError, unbounded_channel},
     time::timeout,
 };
 
@@ -44,11 +47,11 @@ use crate::{
     ted::elg::ELGate,
     units::frequency::hz_to_rad_per_sample,
 };
-use eframe::egui::{self, CentralPanel, Color32};
+use eframe::egui::{self, CentralPanel, Color32, RichText};
 use tokio::sync::RwLock;
 use tokio::sync::mpsc::{Receiver, Sender, channel};
 
-const BAUD_RATE: u32 = 1000;
+const BAUD_RATE: u32 = 1200;
 const SAMPLE_RATE: u32 = 48000;
 
 // Modulation parameters
@@ -105,71 +108,51 @@ impl SampleSender for WavSampleSender {
 }
 
 struct FSKReceiver {
-    pos_correllator: FIRFilter,
-    neg_correllator: FIRFilter,
     eye_sender: Sender<Vec<f32>>,
-    matched_lowpass: FIRFilter,
+    phase_lowpass: FIRFilter,
     elg: ELGate,
-    dc_block: DCBlocker,
     last_byte: u8,
     frame_constructor: FrameConstructor,
     bit_count: Option<u32>,
+    last_sample: Complex32,
 }
 
 impl FSKReceiver {
     fn new(eye_sender: Sender<Vec<f32>>) -> Self {
         let samples_per_symbol = (SAMPLE_RATE as f32) / (BAUD_RATE as f32);
 
-        let correllator_length = samples_per_symbol as usize;
-        let mut pos_nco = Nco::new(hz_to_rad_per_sample(DEVIATION, SAMPLE_RATE as f32));
-        let mut neg_nco = Nco::new(hz_to_rad_per_sample(-DEVIATION, SAMPLE_RATE as f32));
-        let pos_ir = (0..correllator_length).map(|i| {
-            pos_nco.step();
-            pos_nco.cexp() * windows::blackmann(i as f32 / correllator_length as f32)
-        });
-        let neg_ir = (0..correllator_length).map(|i| {
-            neg_nco.step();
-            neg_nco.cexp() * windows::blackmann(i as f32 / correllator_length as f32)
-        });
-        let mut pos_correllator = FIRFilter::new(&pos_ir.collect::<Vec<_>>());
-        let mut neg_correllator = FIRFilter::new(&neg_ir.collect::<Vec<_>>());
-        pos_correllator.normalize_freq(hz_to_rad_per_sample(DEVIATION, SAMPLE_RATE as f32));
-        neg_correllator.normalize_freq(hz_to_rad_per_sample(-DEVIATION, SAMPLE_RATE as f32));
-
-        let mut matched_lowpass = FIRFilter::new(&vec![
+        let mut phase_lowpass = FIRFilter::new(&vec![
             Complex32::new(1., 0.);
             samples_per_symbol as usize / 2
         ]);
-        matched_lowpass.normalize_freq(hz_to_rad_per_sample(DEVIATION, SAMPLE_RATE as f32));
+        phase_lowpass.normalize_dc();
         //let mut dc_block = DCBlocker::new(0.999);
-        let mut dc_block = DCBlocker::new(1.);
+        //let mut dc_block = DCBlocker::new(1.);
 
-        let loop_i = 0.0;
+        let loop_i = 0.03;
         let loop_p = 0.1;
-        let mut loop_ir = vec![Complex32::new(loop_i, 0.); samples_per_symbol as usize];
+        let mut loop_ir = vec![Complex32::new(loop_i, 0.); samples_per_symbol as usize / 2];
         loop_ir.push(Complex32::new(loop_p, 0.));
-        let mut elg = ELGate::new(samples_per_symbol, FIRFilter::new(&loop_ir));
+        let elg = ELGate::new(samples_per_symbol, FIRFilter::new(&loop_ir));
         Self {
             //iq_sampler: IQSampler::new(hz_to_rad_per_sample(CENTER_FREQ, SAMPLE_RATE as f32)),
-            pos_correllator,
-            neg_correllator,
-            matched_lowpass,
-            dc_block,
+            phase_lowpass,
             elg,
             last_byte: 0x00u8,
             frame_constructor: FrameConstructor::new(),
             bit_count: None,
             eye_sender,
+            last_sample: Complex32::new(1., 0.),
         }
     }
 
     async fn receive(&mut self, iq: Complex32) -> Result<Option<Frame>, FrameConstructionError> {
         // Frame reconstruction
-        let matched =
-            self.matched_lowpass.next_real(self.dc_block.next_real(
-                self.pos_correllator.next(iq).mag() - self.neg_correllator.next(iq).mag(),
-            ));
-        if let Some((bit_sample, eye)) = self.elg.next_eye(matched) {
+        let dphi = self
+            .phase_lowpass
+            .next_real((self.last_sample * iq.conj()).arg());
+        self.last_sample = iq;
+        if let Some((bit_sample, eye)) = self.elg.next_eye(dphi) {
             let _ = self.eye_sender.send(eye).await;
             self.last_byte >>= 1;
             self.last_byte |= ((bit_sample > 0.) as u8) << 7;
@@ -199,11 +182,22 @@ impl FSKReceiver {
     }
 }
 
+#[derive(Debug)]
+enum TransceiverState {
+    Waiting,
+    Receiving,
+    EOT,
+    SendingAck,
+    Sending,
+    Listening,
+}
+
 struct Transceiver {
     tx_stream: Sender<Vec<u8>>,
     rx_stream: Receiver<Vec<u8>>,
 
     eye_receiver: Receiver<Vec<f32>>,
+    state_receiver: Receiver<TransceiverState>,
 }
 
 impl Transceiver {
@@ -223,12 +217,15 @@ impl Transceiver {
         self.eye_receiver.try_recv()
     }
 
-    pub fn start(
-        mut sample_stream: Receiver<f32>,
-        mut sample_sender: Sender<SampleSenderCommand>,
-    ) -> Self {
+    pub fn try_recv_state(&mut self) -> Result<TransceiverState, TryRecvError> {
+        self.state_receiver.try_recv()
+    }
+
+    pub fn start(mut sample_stream: Receiver<f32>, mut sample_sender: Sender<Vec<f32>>) -> Self {
         let mut resend: Option<Vec<u8>> = None;
         let (mut eyes_tx, eyes_rx) = channel::<Vec<f32>>(1024);
+        let (mut state_tx, state_rx) = channel::<TransceiverState>(1024);
+        state_tx.try_send(TransceiverState::Waiting);
 
         let (rx_stream_sender, mut rx_stream_receiver) = channel::<Vec<u8>>(1024);
         let (tx_stream_sender, mut tx_stream_receiver) = channel::<Vec<u8>>(1024);
@@ -247,22 +244,29 @@ impl Transceiver {
                     }
                     =>
                     {
+                        state_tx.try_send(TransceiverState::Receiving);
                         // Wait for end of tranmission
                         let mut recv = Some(FSKReceiver::new(eyes_tx.clone()));
+                        let mut send_ack = false;
                         while let Some(iq) = squelch.next(iq_sampler.sample(sample_stream.recv().await.unwrap()))
                         {
                             if recv.as_ref().is_some()
                             {
                                 match recv.as_mut().unwrap().receive(iq).await
                                 {
-                                    Ok(Some(Frame::Data(_))) => {println!("GOT DATA"); Self::transmit(Frame::Ack, &mut sample_sender).await; recv = None;},
-                                    Ok(Some(Frame::Ack)) => {current_message = None; recv = None;},
+                                    Ok(Some(Frame::Data(_))) => {println!("GOT DATA"); send_ack = true; recv = None; state_tx.try_send(TransceiverState::EOT);},
+                                    Ok(Some(Frame::Ack)) => {current_message = None; recv = None; state_tx.try_send(TransceiverState::EOT);},
                                     Err(()) => {recv = None;},
                                     _ => {}
                                 }
                             }
                         }
-                        println!("EOT");
+                        if send_ack
+                        {
+                            state_tx.try_send(TransceiverState::SendingAck);
+                            Self::transmit(Frame::Ack, &mut sample_sender).await;
+                        }
+                        state_tx.try_send(TransceiverState::Waiting);
                     },
                     message = async
                     {
@@ -270,14 +274,17 @@ impl Transceiver {
                         {
                             current_message = Some((tx_stream_receiver).recv().await.unwrap());
                         }
-                        tokio::time::sleep(Duration::from_secs(rand::random_range(1..3))).await;
+                        state_tx.try_send(TransceiverState::Listening);
+                        tokio::time::sleep(Duration::from_millis(500 * rand::random_range(1..6))).await;
                         current_message.as_ref().unwrap()
                     } =>
                     {
+                        state_tx.try_send(TransceiverState::Sending);
                         println!("Sending message");
                         Self::transmit(Frame::Data(message.clone()), &mut sample_sender).await;
                         //current_message = None;
                         println!("Sent message");
+                        state_tx.try_send(TransceiverState::Waiting);
                     }
                 };
             }
@@ -285,30 +292,59 @@ impl Transceiver {
 
         Self {
             eye_receiver: eyes_rx,
+            state_receiver: state_rx,
 
             tx_stream: tx_stream_sender,
             rx_stream: rx_stream_receiver,
         }
     }
 
-    pub async fn transmit(frame: Frame, samples_sender: &mut Sender<SampleSenderCommand>) {
+    pub async fn transmit(frame: Frame, samples_sender: &mut Sender<Vec<f32>>) {
         let bytes = frame.bytes();
-        let mut bit_stream = bytes.iter().flat_map(|x| byte_to_bits(*x));
-        let modulator = BFSKMod::new(
-            (SAMPLE_RATE as f32 / BAUD_RATE as f32).round() as u32,
-            hz_to_rad_per_sample(DEVIATION, SAMPLE_RATE as f32),
-            &mut bit_stream,
-        );
+        let data = bytes
+            .iter()
+            .flat_map(|x| byte_to_bits(*x))
+            .collect::<Vec<_>>();
 
-        let up_lo = Nco::new(hz_to_rad_per_sample(CENTER_FREQ, SAMPLE_RATE as f32));
-        samples_sender.send(SampleSenderCommand::Open).await;
-        for (m, up) in modulator.zip(up_lo) {
-            let sample = m * up;
-            samples_sender
-                .send(SampleSenderCommand::Sample(sample.re))
-                .await;
+        let sample_per_symbols = SAMPLE_RATE / BAUD_RATE;
+        let bitstream = (0..(bytes.len() * 8 * sample_per_symbols as usize)).map(|i| {
+            if data[i / sample_per_symbols as usize] {
+                1.
+            } else {
+                -1.
+            }
+        });
+
+        // Synthesise impulse response
+        let mut impulse_response =
+            vec![Complex32::zero(); sample_per_symbols as usize].into_boxed_slice();
+        for (i, x) in impulse_response.iter_mut().enumerate() {
+            *x = Complex32::new(
+                windows::gaussian(0.3, i as f32 / sample_per_symbols as f32),
+                0.,
+            );
         }
-        samples_sender.send(SampleSenderCommand::Close).await;
+
+        let mut gaussian_filter = FIRFilter::new(&impulse_response);
+        gaussian_filter.normalize_dc();
+        let filtered_bitstream = bitstream.map(|x| gaussian_filter.next_real(x));
+
+        let mut nco = Nco::new(0.);
+        let mut lo = Nco::new(hz_to_rad_per_sample(CENTER_FREQ, SAMPLE_RATE as f32));
+
+        // Generate passband
+        let samples = filtered_bitstream
+            .map(|f| {
+                nco.set_frequency(hz_to_rad_per_sample(f * DEVIATION, SAMPLE_RATE as f32));
+                nco.step_n(1);
+                lo.step_n(1);
+                (nco.cexp() * lo.cexp()).re
+            })
+            .collect::<Vec<_>>();
+
+        let len = samples.len();
+        samples_sender.send(samples).await.unwrap();
+        tokio::time::sleep(Duration::from_secs_f32(len as f32 / SAMPLE_RATE as f32)).await;
     }
 }
 
@@ -389,7 +425,7 @@ impl Frame {
         let mut output_bytes = vec![];
 
         // Initial training sequence
-        output_bytes.append(&mut vec![0b01010101; 64]);
+        output_bytes.append(&mut vec![0b01010101; 32]);
 
         // Preamble byte
         output_bytes.push(0xD8);
@@ -417,7 +453,7 @@ impl Frame {
         }
 
         // SEND EOT
-        output_bytes.extend(std::iter::repeat_n(4, 32));
+        output_bytes.extend(std::iter::repeat_n(4, 16));
         output_bytes
     }
 }
@@ -460,10 +496,11 @@ struct EguiApp {
     transceiver: Transceiver,
 
     eyes: VecDeque<Vec<f32>>,
+    current_state: TransceiverState,
 }
 impl EguiApp {
     fn new(_cc: &eframe::CreationContext<'_>) -> Self {
-        let (up_sender, mut up_receiver) = channel::<SampleSenderCommand>(1024);
+        let (up_sender, mut up_receiver) = channel::<Vec<f32>>(16);
         let (down_sender, down_receiver) = channel::<f32>(1024);
 
         let transceiver = Transceiver::start(down_receiver, up_sender);
@@ -482,63 +519,75 @@ impl EguiApp {
             //     .await
             //     .unwrap();
 
-            let mut stream;
-
-            if instance_id == 0 {
-                let socket = TcpSocket::new_v4().unwrap();
-                socket.bind("0.0.0.0:9000".parse().unwrap()).unwrap();
-                stream = socket.listen(1).unwrap().accept().await.unwrap().0;
-            } else {
-                stream = TcpStream::connect("127.0.0.1:9000").await.unwrap();
-            }
-
-            println!("Connected");
-
-            let (mut sock_recv, mut sock_send) = io::split(stream);
             // Receiving end
-            tokio::spawn(async move {
-                let mut buf = [0u8; 65536];
-                while let Ok(size) = sock_recv.read(&mut buf).await {
-                    assert!(buf.len() % 4 == 0);
-                    for x in buf.chunks(4).take(size / 4) {
-                        let _ = down_sender.try_send(f32::from_ne_bytes(x.try_into().unwrap()));
-                    }
-                }
-            });
+            let host = cpal::default_host();
 
-            // Sending end
-            let mut sending = false;
-            let mut wait_countdown = 0;
-            loop {
-                // Up link
-                //let mut sample = 0.;
-                let noise = rand::random::<f32>() * 0.1;
-                if let Ok(x) = up_receiver.try_recv() {
-                    match x {
-                        SampleSenderCommand::Open => {
-                            sending = true;
+            let device = host.default_input_device().expect("No input device");
+            let mut config = device
+                .supported_input_configs()
+                .unwrap()
+                .next()
+                .unwrap()
+                .with_sample_rate(cpal::SampleRate(48000));
+
+            let stream = device
+                .build_input_stream(
+                    &config.into(),
+                    move |data: &[f32], _| {
+                        for x in data.iter() {
+                            let _ = down_sender.blocking_send(*x * 30.); // non-blocking send
                         }
-                        SampleSenderCommand::Close => {
-                            sending = false;
-                        }
-                        SampleSenderCommand::Sample(x) => {
-                            if sending {
-                                let _ = sock_send.write(&(x + noise).to_ne_bytes()).await;
-                                wait_countdown += 1;
+                    },
+                    move |err| eprintln!("Stream error: {}", err),
+                    None,
+                )
+                .unwrap();
+            stream.play().unwrap();
+
+            let device = host.default_output_device().unwrap();
+            let mut supported_configs_range = device.supported_output_configs().unwrap();
+            let supported_config = supported_configs_range
+                .find(|config| {
+                    config.sample_format() == cpal::SampleFormat::F32
+                        && config.min_sample_rate().0 <= 48000
+                        && config.max_sample_rate().0 >= 48000
+                })
+                .expect("Device does not support 48kHz f32 output");
+            let config = supported_config
+                .with_sample_rate(cpal::SampleRate(48_000))
+                .config();
+
+            while let Some(stream) = up_receiver.recv().await {
+                let stream_len = stream.len();
+                let progression = Arc::new(AtomicU64::new(0));
+                let (finished_tx, mut finished_rx) = channel::<()>(16);
+
+                let send_stream = device
+                    .build_output_stream(
+                        &config,
+                        move |data: &mut [f32], _: &cpal::OutputCallbackInfo| {
+                            for d in data.iter_mut() {
+                                if progression.load(std::sync::atomic::Ordering::Relaxed) as usize
+                                    == stream.len()
+                                {
+                                    let _ = finished_tx.blocking_send(());
+                                    break;
+                                }
+
+                                *d = stream[progression
+                                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
+                                    as usize]
+                                    * 0.1; // TODO
                             }
-                        }
-                    }
-                }
-
-                if !sending {
-                    let _ = sock_send.write(&noise.to_ne_bytes()).await;
-                    wait_countdown += 1;
-                }
-
-                if wait_countdown >= 480 {
-                    tokio::time::sleep(Duration::from_millis(10)).await;
-                    wait_countdown = 0;
-                }
+                        },
+                        move |err| {
+                            eprintln!("Stream error: {}", err);
+                        },
+                        None,
+                    )
+                    .unwrap();
+                send_stream.play().unwrap();
+                let _ = finished_rx.recv().await;
             }
         });
 
@@ -546,6 +595,7 @@ impl EguiApp {
             transceiver,
 
             eyes: VecDeque::new(),
+            current_state: TransceiverState::Waiting,
         }
     }
 }
@@ -561,6 +611,28 @@ impl eframe::App for EguiApp {
             while self.eyes.len() > max_eyes {
                 self.eyes.pop_front();
             }
+            if let Ok(new_state) = self.transceiver.try_recv_state() {
+                self.current_state = new_state;
+            }
+
+            ui.horizontal(|ui| {
+                if ui.button("Start").clicked() {
+                    let snd = self.transceiver.get_sender();
+                    let data = (0..rand::random_range(50..250))
+                        .map(|_| rand::random::<char>() as u8)
+                        .collect::<Vec<_>>();
+
+                    tokio::spawn(async move {
+                        let _ = snd.send(data).await;
+                    });
+                }
+
+                ui.label(
+                    RichText::new(format!("{:?}", self.current_state))
+                        .size(35.)
+                        .color(Color32::LIGHT_GREEN),
+                );
+            });
 
             Plot::new("EyeA")
                 .legend(Legend::default())
@@ -578,15 +650,6 @@ impl eframe::App for EguiApp {
                         plot_ui.line(line);
                     }
                 });
-
-            if ui.button("Start").clicked() {
-                let snd = self.transceiver.get_sender();
-                tokio::spawn(async move {
-                    let _ = snd
-                        .send("Skibditoilet".repeat(10).as_bytes().to_vec())
-                        .await;
-                });
-            }
         }); // Central panel
 
         std::thread::sleep(Duration::from_millis(16));

src/windows.rs

@@ -10,18 +10,20 @@ pub fn bartlett(t: f32) -> f32 {
     if t < 0.5 { 2. * t } else { 2. - 2. * t }
 }
 
-pub fn hann(t: f32) -> f32
-{
+pub fn hann(t: f32) -> f32 {
     0.5 - 0.5 * (2. * PI * t).cos()
 }
 
-pub fn hamming(t: f32) -> f32
-{
+pub fn hamming(t: f32) -> f32 {
     0.54 - 0.46 * (2. * PI * t).cos()
 }
 
-pub fn blackmann(t: f32) -> f32
-{
+pub fn blackmann(t: f32) -> f32 {
     let x = 2. * PI * t;
     0.45 - 0.5 * x.cos() + 0.08 * (2. * x).cos()
 }
+
+pub fn gaussian(sigma: f32, t: f32) -> f32 {
+    let sq = (t - 0.5) / sigma;
+    (-sq * sq).exp()
+}