diff --git a/src/iced-app/Cargo.toml b/src/iced-app/Cargo.toml index e26a3a8..9edbf5b 100644 --- a/src/iced-app/Cargo.toml +++ b/src/iced-app/Cargo.toml @@ -19,6 +19,7 @@ tracing = { workspace = true } tracing-subscriber = { workspace = true } chrono = { workspace = true } cpal = { version = "0.15", optional = true } +audiopus = { version = "0.3.0-rc.0", optional = true } shared = { workspace = true } tscore = { workspace = true } @@ -28,4 +29,4 @@ tsproto-packets = { workspace = true } [features] default = [] -audio = ["dep:cpal"] +audio = ["dep:cpal", "dep:audiopus"] diff --git a/src/iced-app/src/audio.rs b/src/iced-app/src/audio.rs index 0e0d8fa..1a7debe 100644 --- a/src/iced-app/src/audio.rs +++ b/src/iced-app/src/audio.rs @@ -1,10 +1,12 @@ use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; use tsclientlib::audio::AudioHandler; use tsclientlib::ClientId; -use tsproto_packets::packets::{AudioData, InAudioBuf}; +use tsproto_packets::packets::{AudioData, CodecType, InAudioBuf, OutAudio}; const SAMPLE_RATE: u32 = 48000; const CHANNELS: u16 = 2; +const FRAME_SIZE: usize = 960; // 20ms at 48kHz mono +const OPUS_MAX_PACKET: usize = 1275; type PacketSender = std::sync::mpsc::Sender; @@ -14,7 +16,6 @@ struct PlaybackState { _stream: cpal::Stream, } -// cpal::Stream is safe to send across threads (it's a handle to the audio device) unsafe impl Send for PlaybackState {} unsafe impl Sync for PlaybackState {} @@ -125,23 +126,27 @@ impl AudioPlayback { } } -pub struct AudioCapture { - _stream: Option, +struct CaptureState { + _stream: cpal::Stream, input_device: String, } -impl AudioCapture { +unsafe impl Send for CaptureState {} +unsafe impl Sync for CaptureState {} + +pub struct Microphone { + state: Option, +} + +impl Microphone { pub fn new() -> Self { - Self { - _stream: None, - input_device: String::new(), - } + Self { state: None } } pub fn start( &mut self, device_name: Option<&str>, - audio_tx: tokio::sync::mpsc::Sender>, + sample_tx: std::sync::mpsc::Sender>, ) -> Result<(), String> { let host = cpal::default_host(); @@ -168,7 +173,7 @@ impl AudioCapture { .build_input_stream( &config, move |data: &[f32], _: &cpal::InputCallbackInfo| { - let _ = audio_tx.blocking_send(data.to_vec()); + let _ = sample_tx.send(data.to_vec()); }, |err| tracing::error!("Audio input error: {err}"), None, @@ -179,23 +184,84 @@ impl AudioCapture { .play() .map_err(|e| format!("Failed to start capture: {e}"))?; - self._stream = Some(stream); - self.input_device = device_name_str; + self.state = Some(CaptureState { + _stream: stream, + input_device: device_name_str, + }); - tracing::info!("Audio capture started"); + tracing::info!("Microphone started"); Ok(()) } pub fn stop(&mut self) { - self._stream = None; - tracing::info!("Audio capture stopped"); + self.state = None; + tracing::info!("Microphone stopped"); } pub fn input_device(&self) -> &str { - &self.input_device + self.state + .as_ref() + .map(|s| s.input_device.as_str()) + .unwrap_or("") } pub fn is_active(&self) -> bool { - self._stream.is_some() + self.state.is_some() + } +} + +pub struct OpusEncoderState { + encoder: audiopus::coder::Encoder, + packet_id: u16, + opus_buf: [u8; OPUS_MAX_PACKET], + sample_buf: Vec, +} + +impl OpusEncoderState { + pub fn new() -> Result { + let encoder = audiopus::coder::Encoder::new( + audiopus::SampleRate::Hz48000, + audiopus::Channels::Mono, + audiopus::Application::Voip, + ) + .map_err(|e| format!("Failed to create Opus encoder: {e}"))?; + + Ok(Self { + encoder, + packet_id: 0, + opus_buf: [0u8; OPUS_MAX_PACKET], + sample_buf: Vec::with_capacity(FRAME_SIZE * 2), + }) + } + + pub fn encode_and_send( + &mut self, + samples: &[f32], + audio_tx: &tokio::sync::mpsc::Sender>, + ) { + self.sample_buf.extend_from_slice(samples); + + while self.sample_buf.len() >= FRAME_SIZE { + let frame: Vec = self.sample_buf.drain(..FRAME_SIZE).collect(); + match self.encoder.encode_float(&frame, &mut self.opus_buf) { + Ok(len) => { + let packet_data = self.opus_buf[..len].to_vec(); + let _ = audio_tx.blocking_send(packet_data); + self.packet_id = self.packet_id.wrapping_add(1); + } + Err(e) => { + tracing::debug!("Opus encode error: {e}"); + } + } + } + } + + pub fn packet_id(&self) -> u16 { + self.packet_id + } + + pub fn reset(&mut self) { + self.sample_buf.clear(); + self.packet_id = 0; } } diff --git a/src/iced-app/src/main.rs b/src/iced-app/src/main.rs index df197a2..4cf299a 100644 --- a/src/iced-app/src/main.rs +++ b/src/iced-app/src/main.rs @@ -10,6 +10,8 @@ use tsclientlib::sync::{SyncConnection, SyncConnectionHandle, SyncStreamItem}; use tsclientlib::{ChannelId, ClientId, Connection, DisconnectOptions, MessageTarget}; use tsclientlib::events::{Event, PropertyId}; use tsclientlib::prelude::*; +#[cfg(feature = "audio")] +use tsproto_packets::packets::{AudioData, CodecType, OutAudio}; mod theme; @@ -54,6 +56,12 @@ enum Message { RunServerQuery, QueryResponse(String), QueryError(String), + #[cfg(feature = "audio")] + PttPressed, + #[cfg(feature = "audio")] + PttReleased, + #[cfg(feature = "audio")] + MicSamples(Vec), Noop, } @@ -136,6 +144,14 @@ struct App { session_id: u64, #[cfg(feature = "audio")] audio: Arc>, + #[cfg(feature = "audio")] + mic: Arc>, + #[cfg(feature = "audio")] + encoder: Arc>>, + #[cfg(feature = "audio")] + ptt_active: bool, + #[cfg(feature = "audio")] + audio_send_tx: Option>>, } impl App { @@ -186,6 +202,14 @@ impl App { session_id: 0, #[cfg(feature = "audio")] audio: Arc::new(Mutex::new(audio::AudioPlayback::new())), + #[cfg(feature = "audio")] + mic: Arc::new(Mutex::new(audio::Microphone::new())), + #[cfg(feature = "audio")] + encoder: Arc::new(Mutex::new(None)), + #[cfg(feature = "audio")] + ptt_active: false, + #[cfg(feature = "audio")] + audio_send_tx: None, }; (app, Task::none()) @@ -518,6 +542,97 @@ impl App { self.query_error = Some(e); Task::none() } + #[cfg(feature = "audio")] + Message::PttPressed => { + if !self.ptt_active && self.connected { + self.ptt_active = true; + let mic = self.mic.clone(); + let encoder = self.audio_send_tx.clone(); + let (sample_tx, sample_rx) = std::sync::mpsc::channel(); + let h = self.handle.clone(); + + tokio::spawn(async move { + { + let mut mic_guard = mic.lock().await; + if mic_guard.start(None, sample_tx).is_err() { + return; + } + } + + let mut enc = match audio::OpusEncoderState::new() { + Ok(e) => e, + Err(_) => return, + }; + + let (audio_tx, mut audio_rx) = tokio::sync::mpsc::channel::>(100); + + let send_handle = h.clone(); + tokio::spawn(async move { + while let Some(data) = audio_rx.recv().await { + let mut guard = send_handle.lock().await; + if let Some(ref mut handle) = *guard { + let _ = handle + .with_connection(move |con| { + let packet = OutAudio::new(&AudioData::C2S { + id: 0, + codec: CodecType::OpusVoice, + data: &data, + }); + let _ = con.send_audio(packet); + Ok::<(), ()>(()) + }) + .await; + } + } + }); + + loop { + match sample_rx.recv() { + Ok(samples) => { + enc.encode_and_send(&samples, &audio_tx); + } + Err(_) => break, + } + } + + let mut mic_guard = mic.lock().await; + mic_guard.stop(); + }); + } + Task::none() + } + #[cfg(feature = "audio")] + Message::PttReleased => { + if self.ptt_active { + self.ptt_active = false; + let mic = self.mic.clone(); + let h = self.handle.clone(); + tokio::spawn(async move { + { + let mut mic_guard = mic.lock().await; + mic_guard.stop(); + } + // Send end-of-transmission + let mut guard = h.lock().await; + if let Some(ref mut handle) = *guard { + let _ = handle + .with_connection(move |con| { + let packet = OutAudio::new(&AudioData::C2S { + id: 0, + codec: CodecType::OpusVoice, + data: &[], + }); + let _ = con.send_audio(packet); + Ok::<(), ()>(()) + }) + .await; + } + }); + } + Task::none() + } + #[cfg(feature = "audio")] + Message::MicSamples(_) => Task::none(), Message::Noop => Task::none(), } } @@ -665,37 +780,63 @@ impl App { } fn subscription(&self) -> Subscription { - if !self.connected { - return Subscription::none(); - } - let event_rx = self.event_rx.clone(); - let session_id = self.session_id; - Subscription::run_with_id( - session_id, - iced::stream::channel(100, move |mut sender| async move { - loop { - let event = { - let mut guard = event_rx.lock().await; - match guard.as_mut() { - Some(rx) => rx.recv().await, - None => break, + let mut subs = Vec::new(); + + // Keyboard events for PTT + #[cfg(feature = "audio")] + subs.push(iced::event::listen().map(|event| { + match event { + iced::Event::Keyboard(iced::keyboard::Event::KeyPressed { ref key, .. }) => { + if let iced::keyboard::Key::Character(c) = key { + if c.as_str().to_lowercase() == "v" { + return Message::PttPressed; } - }; - match event { - Some(event) => { - let is_disconnect = matches!(event, TsEvent::Disconnected | TsEvent::Error(_)); - if sender.send(Message::TsEvent(event)).await.is_err() { - break; - } - if is_disconnect { - break; - } - } - None => break, } } - }), - ) + iced::Event::Keyboard(iced::keyboard::Event::KeyReleased { ref key, .. }) => { + if let iced::keyboard::Key::Character(c) = key { + if c.as_str().to_lowercase() == "v" { + return Message::PttReleased; + } + } + } + _ => {} + } + Message::Noop + })); + + if self.connected { + let event_rx = self.event_rx.clone(); + let session_id = self.session_id; + subs.push(Subscription::run_with_id( + session_id, + iced::stream::channel(100, move |mut sender| async move { + loop { + let event = { + let mut guard = event_rx.lock().await; + match guard.as_mut() { + Some(rx) => rx.recv().await, + None => break, + } + }; + match event { + Some(event) => { + let is_disconnect = matches!(event, TsEvent::Disconnected | TsEvent::Error(_)); + if sender.send(Message::TsEvent(event)).await.is_err() { + break; + } + if is_disconnect { + break; + } + } + None => break, + } + } + }), + )); + } + + Subscription::batch(subs) } fn view(&self) -> Element<'_, Message> {