diff --git a/send_udp.sh b/send_udp.sh new file mode 100755 index 0000000..88acf6d --- /dev/null +++ b/send_udp.sh @@ -0,0 +1,3 @@ +#!/bin/sh + +ffmpeg -re -i audio/modulated.wav -f s16le -acodec pcm_s16le udp://127.0.0.1:8080 diff --git a/src/main.rs b/src/main.rs index ff86859..57eb8c6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -12,8 +12,9 @@ mod units; mod windows; use egui_plot::{Legend, Line, Plot}; -use rand::{rand_core::le, seq::index::sample}; -use std::{collections::VecDeque, io::{stdout, Write}, time::Duration}; +use hound::WavWriter; +use rand::{rand_core::le, seq::index::sample, Rng}; +use std::{cell::{Cell, RefCell}, collections::VecDeque, env::{self, args}, fs::File, io::{stdout, BufWriter, Sink, Write}, ops::DerefMut, sync::Arc, time::Duration}; use tokio::{join, net::UdpSocket, select, time::timeout}; use crate::{ @@ -25,7 +26,7 @@ use crate::{ ted::elg::ELGate, units::frequency::hz_to_rad_per_sample, }; -use eframe::{egui::{self, Color32}, glow::SAMPLE_MASK_VALUE}; +use eframe::{egui::{self, mutex::{Mutex, RwLock}, panel::Side, CentralPanel, Color32, SidePanel}, glow::SAMPLE_MASK_VALUE}; use tokio::sync::mpsc::{Receiver, Sender, channel}; const BAUD_RATE: u32 = 1000; @@ -37,39 +38,124 @@ const DEVIATION: f32 = 500.; pub trait SampleSender { - fn open_link(&mut self); - fn send_samples(&mut self, samples: &[f32]); - fn close_link(&mut self); + async fn open_link(&mut self); + async fn send_sample(&mut self, sample: f32); + async fn close_link(&mut self); +} + +pub struct ChannelSampleSender +{ + pub sender: Sender, + pub open: Arc>, +} + +impl SampleSender for ChannelSampleSender +{ + async fn open_link(&mut self) + { + *self.open.write() = true; + } + + async fn send_sample(&mut self, sample: f32) { + self.sender.send(sample).await.unwrap(); + } + + async fn close_link(&mut self) + { + *self.open.write() = false; + } +} + +fn create_dummy_channel() -> ((Receiver, ChannelSampleSender), (Receiver, ChannelSampleSender)) +{ + let a_open = Arc::new(RwLock::new(false)); + let b_open = Arc::new(RwLock::new(false)); + let a_open_cln = a_open.clone(); + let b_open_cln = b_open.clone(); + + // Up link + let (a_up_tx, mut a_up_rx) = channel::(1024); + let (b_up_tx, mut b_up_rx) = channel::(1024); + + // Down link + let (a_down_tx, a_down_rx) = channel::(1024); + let (b_down_tx, b_down_rx) = channel::(1024); + + tokio::spawn(async move { + loop + { + let sample = select! + { + Some(x) = a_up_rx.recv() => x, + Some(x) = b_up_rx.recv() => x + }; + + if !(*a_open_cln.read()) + { + a_down_tx.send(sample).await.unwrap(); + } + + if !(*b_open_cln.read()) + { + b_down_tx.send(sample).await.unwrap(); + } + } + }); + + + ( + ( + a_down_rx, + ChannelSampleSender + { + open: a_open, + sender: a_up_tx + } + ), + ( + b_down_rx, + ChannelSampleSender + { + open: b_open, + sender: b_up_tx + } + ) + ) + } struct WavSampleSender { - + writer: Option>> +} + +impl Default for WavSampleSender +{ + fn default() -> Self { + Self { writer: None } + } } impl SampleSender for WavSampleSender { - fn open_link(&mut self) { - } - - fn send_samples(&mut self, samples: &[f32]) { + async fn open_link(&mut self) { let spec = hound::WavSpec { channels: 1, sample_rate: SAMPLE_RATE, bits_per_sample: 16, sample_format: hound::SampleFormat::Int, }; - let mut writer = hound::WavWriter::create("audio/modulated.wav", spec).unwrap(); - - for s in samples - { - let out_sample = (s * i16::MAX as f32) as i16; - writer.write_sample(out_sample).unwrap(); - } - writer.finalize().unwrap(); + self.writer = Some(hound::WavWriter::create("audio/modulated.wav", spec).unwrap()); } - fn close_link(&mut self) { + async fn send_sample(&mut self, sample: f32) { + + let out_sample = (sample * i16::MAX as f32) as i16; + self.writer.as_mut().unwrap().write_sample(out_sample).unwrap(); + } + + async fn close_link(&mut self) { + self.writer = None; } } @@ -80,7 +166,7 @@ impl Transceiver { mut sample_stream: Receiver, mut tx_stream: Receiver>, mut rx_stream: Sender>, - sample_sender: &mut T, + mut sample_sender: T, mut eye_sender: Sender> ) { let mut resend: Option> = None; @@ -88,6 +174,7 @@ impl Transceiver { select! { _ = Self::squelch_detector(&mut sample_stream) => { + println!("Squelch up"); select! { x = Self::receive(&mut sample_stream, &mut eye_sender) => @@ -102,7 +189,7 @@ impl Transceiver { Ok(Frame::Data(data)) => { rx_stream.send(data).await.unwrap(); - Self::transmit(Frame::Ack, sample_sender).await; + Self::transmit(Frame::Ack, &mut sample_sender).await; } } }, @@ -127,7 +214,8 @@ impl Transceiver { { if let Some(data) = data_opt { - Self::transmit(Frame::Data(data.clone()), sample_sender).await; + println!("Sending data"); + Self::transmit(Frame::Data(data.clone()), &mut sample_sender).await; resend = Some(data); } } @@ -166,16 +254,12 @@ impl Transceiver { ); let up_lo = Nco::new(hz_to_rad_per_sample(CENTER_FREQ, SAMPLE_RATE as f32)); - - let mut sample_buffer = vec![]; + samples_sender.open_link().await; for (m, up) in modulator.zip(up_lo) { let sample = m * up; - sample_buffer.push(sample.re); // Project IQ + samples_sender.send_sample(sample.re).await; } - - samples_sender.open_link(); - samples_sender.send_samples(&sample_buffer); - samples_sender.close_link(); + samples_sender.close_link().await; } async fn receive(sample_stream: &mut Receiver, eye_sender: &mut Sender>) -> Result { @@ -380,8 +464,8 @@ impl Frame { #[tokio::main] async fn main() { - let rand = rand::rng(); - Transceiver::transmit(Frame::Data("Skibditoilet".repeat(100).bytes().collect::>()), &mut WavSampleSender{}).await; + //Transceiver::transmit(Frame::Data("Skibditoilet".repeat(100).bytes().collect::>()), &mut WavSampleSender{}).await; + Transceiver::transmit(Frame::Ack, &mut WavSampleSender::default()).await; //return; let native_options = eframe::NativeOptions::default(); @@ -397,102 +481,103 @@ async fn main() { struct DummySampleSender(); impl SampleSender for DummySampleSender { - fn open_link(&mut self) {} - fn send_samples(&mut self, samples: &[f32]) {} - fn close_link(&mut self) {} + async fn open_link(&mut self) {} + async fn send_sample(&mut self, _sample: f32) {} + async fn close_link(&mut self) {} } struct EguiApp { - eye_receiver: Receiver>, - eyes: VecDeque> + eye_receiver_a: Receiver>, + eye_receiver_b: Receiver>, + eyes_a: VecDeque>, + eyes_b: VecDeque>, + up_a_tx: Sender> } impl EguiApp { fn new(_cc: &eframe::CreationContext<'_>) -> Self { - let (eye_tx, mut eye_rx) = channel::>(1024); - let (eye_red_tx, eye_red_rx) = channel::>(1024); + let (eye_a_tx, mut eye_a_rx) = channel::>(1024); + let (eye_b_tx, mut eye_b_rx) = channel::>(1024); + + let (eye_a_red_tx, eye_a_red_rx) = channel::>(1024); + let (eye_b_red_tx, eye_b_red_rx) = channel::>(1024); let ctx = _cc.egui_ctx.clone(); tokio::spawn(async move { loop { - if let Some(eye) = eye_rx.recv().await + let _ = select! { - ctx.request_repaint(); - eye_red_tx.send(eye).await; - } + Some(eye) = eye_a_rx.recv() => {eye_a_red_tx.send(eye).await.unwrap()} + Some(eye) = eye_b_rx.recv() => {eye_b_red_tx.send(eye).await.unwrap()} + }; + ctx.request_repaint(); } } ); - let (sample_tx, sample_rx) = channel::(1024); + let ((sample_down_a, sample_up_a), (sample_down_b, sample_up_b)) = create_dummy_channel(); + let (up_a_tx, up_a_rx) = channel::>(1024); + let (down_a_tx, down_a_rx) = channel::>(1024); - let (transmit_tx, transmit_rx) = channel::>(1024); - let (receive_tx, mut receive_rx) = channel::>(1024); + let (up_b_tx, up_b_rx) = channel::>(1024); + let (down_b_tx, down_b_rx) = channel::>(1024); - tokio::spawn(async move { - Transceiver::start(sample_rx, transmit_rx, receive_tx, &mut DummySampleSender(), eye_tx).await; - }); + tokio::spawn(Transceiver::start(sample_down_a, up_a_rx, down_a_tx, sample_up_a, eye_a_tx)); + tokio::spawn(Transceiver::start(sample_down_b, up_b_rx, down_b_tx, sample_up_b, eye_b_tx)); - tokio::spawn(async move { - loop - { - let bytes = receive_rx.recv().await.unwrap(); - let str = String::from_utf8(bytes).unwrap(); - println!("{}", str); - - } - }); + // tokio::spawn(async move { + // loop + // { + // let bytes = receive_rx.recv().await.unwrap(); + // let str = String::from_utf8(bytes).unwrap(); + // println!("{}", str); + // + // } + // }); + // - tokio::spawn(async move { - let sock = UdpSocket::bind("0.0.0.0:8080").await.unwrap(); - let mut buf = [0u8; 2048]; - let mut sample = [0u8; 2]; - let mut byte_index = 0; - - loop { - let len = sock.recv(&mut buf).await.unwrap(); - for x in buf.iter().take(len) { - sample[byte_index] = *x; - byte_index += 1; - if byte_index >= 2 { - let pcm16_sample = i16::from_le_bytes(sample); - sample_tx - .send(pcm16_sample as f32 / i16::MAX as f32) - .await - .unwrap(); - byte_index = 0; - } - } - } - }); Self { - eye_receiver: eye_red_rx, - eyes: VecDeque::with_capacity(100) + eye_receiver_a: eye_a_red_rx, + eye_receiver_b: eye_b_red_rx, + eyes_a: VecDeque::with_capacity(100), + eyes_b: VecDeque::with_capacity(100), + + up_a_tx, } } } impl eframe::App for EguiApp { fn update(&mut self, ctx: &egui::Context, _frame: &mut eframe::Frame) { - egui::CentralPanel::default().show(ctx, |ui| { + egui::CentralPanel::default() + .show(ctx, |ui| { let max_eyes = 100; - while let Ok(eye) = self.eye_receiver.try_recv() { - self.eyes.push_back(eye); + while let Ok(eye) = self.eye_receiver_a.try_recv() { + self.eyes_a.push_back(eye); } - while self.eyes.len() > max_eyes { - self.eyes.pop_front(); + while let Ok(eye) = self.eye_receiver_b.try_recv() { + self.eyes_b.push_back(eye); } - Plot::new("Eye") + while self.eyes_a.len() > max_eyes { + self.eyes_a.pop_front(); + } + + while self.eyes_b.len() > max_eyes { + self.eyes_b.pop_front(); + } + + ui.columns(2, |uis| { + Plot::new("EyeA") .legend(Legend::default()) - .show(ui, |plot_ui| { + .show(&mut uis[0], |plot_ui| { //plot_ui.set_auto_bounds(Vec2b { x: false, y: false }); - for eye in self.eyes.iter() { + for eye in self.eyes_a.iter() { let line = Line::new( - "Eye", + "EyeA", eye.iter() .enumerate() .map(|(i, x)| [i as f64, *x as f64]) @@ -502,7 +587,33 @@ impl eframe::App for EguiApp { plot_ui.line(line); } }); - }); + + if uis[0].button("Start").clicked() + { + let snd = self.up_a_tx.clone(); + tokio::spawn(async move { + let _ = snd.send("Skibditoilet".repeat(100).as_bytes().to_vec()).await; + }); + } + + Plot::new("EyeB") + .legend(Legend::default()) + .show(&mut uis[1], |plot_ui| { + //plot_ui.set_auto_bounds(Vec2b { x: false, y: false }); + for eye in self.eyes_b.iter() { + let line = Line::new( + "EyeB", + eye.iter() + .enumerate() + .map(|(i, x)| [i as f64, *x as f64]) + .collect::>(), + ) + .color(Color32::LIGHT_GREEN); + plot_ui.line(line); + } + }); + }); + }); } }