increased mic logging

client/Cargo.toml

@@ -28,6 +28,7 @@ raw-window-handle       = "0.6"
 serde                   = { version = "1.0", features = ["derive"] }
 serde_json              = "1.0"
 async-stream            = "0.3"
+shell-escape            = "0.1"
 
 [build-dependencies]
 prost-build             = "0.13"

client/src/app.rs

@@ -14,7 +14,8 @@ use winit::{
 };
 
 use lesavka_common::lesavka::{
-    relay_client::RelayClient, KeyboardReport, MonitorRequest, MouseReport, VideoPacket,
+    relay_client::RelayClient, KeyboardReport, 
+    MonitorRequest, MouseReport, VideoPacket, AudioPacket
 };
 
 use crate::{input::inputs::InputAggregator,
@@ -51,7 +52,7 @@ impl LesavkaClientApp {
         /*────────── persistent gRPC channels ──────────*/
         let hid_ep = Channel::from_shared(self.server_addr.clone())?
             .tcp_nodelay(true)
-            .concurrency_limit(1)
+            .concurrency_limit(4)
             .http2_keep_alive_interval(Duration::from_secs(15))
             .connect_lazy();
 
@@ -256,32 +257,34 @@ impl LesavkaClientApp {
     async fn mic_loop(ep: Channel, mic: Arc<MicrophoneCapture>) {
         loop {
             let mut cli = RelayClient::new(ep.clone());
-
+    
+            // 1. create a Tokio MPSC channel
+            let (tx, rx) = tokio::sync::mpsc::channel::<AudioPacket>(256);
+            let (stop_tx, stop_rx) = std::sync::mpsc::channel::<()>();
+    
+            // 2. spawn a real thread that does the blocking `pull()`
             let mic_clone = mic.clone();
-            let stream = async_stream::stream! {
+            std::thread::spawn(move || {
-                loop {
+                while stop_rx.try_recv().is_err() {
                     if let Some(pkt) = mic_clone.pull() {
-                        yield pkt;
+                        trace!("🎤📤 cli  {} bytes → gRPC", pkt.data.len());
-                    } else {
+                        let _ = tx.blocking_send(pkt);
-                        break;                    // EOS (should never happen)
                     }
                 }
-            };
+            });
+    
+            // 3. turn `rx` into an async stream for gRPC
+            let outbound = tokio_stream::wrappers::ReceiverStream::new(rx);
 
-            match cli.stream_microphone(Request::new(stream)).await {
+            match cli.stream_microphone(Request::new(outbound)).await {
                 Ok(mut resp) => {
-                    // Drain the (mostly empty) response stream
+                    // Drain and ignore Empty replies
-                    while let Some(res) = resp.get_mut().message().await.transpose() {
+                    while resp.get_mut().message().await.transpose().is_some() {}
-                        if let Err(e) = res {
-                            warn!("🎤 server err: {e}");
-                            break;
-                        }
-                    }
                 }
                 Err(e) => warn!("🎤 connect failed: {e}"),
             }
-
+            let _ = stop_tx.send(());
-            tokio::time::sleep(Duration::from_secs(1)).await;   // retry
+            tokio::time::sleep(Duration::from_secs(1)).await;
         }
-    }
+    }    
 }

client/src/input/inputs.rs

@@ -7,8 +7,7 @@ use tracing::{debug, info, warn};
 
 use lesavka_common::lesavka::{KeyboardReport, MouseReport};
 
-use super::{keyboard::KeyboardAggregator, mouse::MouseAggregator,
+use super::{keyboard::KeyboardAggregator, mouse::MouseAggregator, camera::CameraCapture};
-            camera::CameraCapture,  microphone::MicrophoneCapture};
 use crate::layout::{Layout, apply as apply_layout};
 
 pub struct InputAggregator {

client/src/input/microphone.rs

@@ -7,7 +7,9 @@ use gstreamer as gst;
 use gstreamer_app as gst_app;
 use gst::prelude::*;
 use lesavka_common::lesavka::AudioPacket;
-use tracing::{debug, error, info, warn};
+use tracing::{debug, error, info, warn, trace};
+use shell_escape::unix::escape;
+use std::sync::atomic::{AtomicU64, Ordering};
 
 pub struct MicrophoneCapture {
     pipeline: gst::Pipeline,
@@ -19,15 +21,24 @@ impl MicrophoneCapture {
         gst::init().ok();                        // idempotent
 
         /* pulsesrc (default mic) → AAC/ADTS → appsink  -------------------*/
-        let desc = concat!(
+        // Optional override: LESAVKA_MIC_SOURCE=<pulse‑device‑name>
-            "pulsesrc do-timestamp=true ! ",
+        let device_arg = match std::env::var("LESAVKA_MIC_SOURCE") {
-            "audio/x-raw,format=S16LE,channels=2,rate=48000 ! ",
+            Ok(s) if !s.is_empty() => {
-            "audioconvert ! audioresample ! avenc_aac bitrate=128000 ! ",
+                let full = Self::pulse_source_by_substr(&s).unwrap_or(s);
-            "aacparse ! capsfilter caps=audio/mpeg,stream-format=adts,rate=48000,channels=2 ! ",
+                format!("device={}", escape(full.into()))
-            "appsink name=asink emit-signals=true max-buffers=50 drop=true"
+            }
+            _ => String::new(),
+        };
+        debug!("🎤 device: {device_arg}");
+        let desc = format!(
+            "pulsesrc {device_arg} do-timestamp=true ! \
+            audio/x-raw,format=S16LE,channels=2,rate=48000 ! \
+            audioconvert ! audioresample ! avenc_aac bitrate=128000 ! \
+            aacparse ! capsfilter caps=audio/mpeg,stream-format=adts,rate=48000,channels=2 ! \
+            appsink name=asink emit-signals=true max-buffers=50 drop=true"
         );
 
-        let pipeline: gst::Pipeline = gst::parse::launch(desc)?
+        let pipeline: gst::Pipeline = gst::parse::launch(&desc)?
             .downcast()
             .expect("pipeline");
         let sink: gst_app::AppSink = pipeline
@@ -69,9 +80,29 @@ impl MicrophoneCapture {
                 let buf = sample.buffer().unwrap();
                 let map = buf.map_readable().unwrap();
                 let pts = buf.pts().unwrap_or(gst::ClockTime::ZERO).nseconds() / 1_000;
+                static CNT: AtomicU64 = AtomicU64::new(0);
+                let n = CNT.fetch_add(1, Ordering::Relaxed);
+                if n < 10 || n % 300 == 0 {
+                    trace!("🎤⇧ cli  pkt#{n}  {} bytes", map.len());
+                }
                 Some(AudioPacket { id: 0, pts, data: map.as_slice().to_vec() })
+
             }
             Err(_) => None,
         }
     }
+
+    fn pulse_source_by_substr(fragment: &str) -> Option<String> {
+        use std::process::Command;
+        let out = Command::new("pactl").args(["list", "short", "sources"])
+            .output().ok()?;
+        let list = String::from_utf8_lossy(&out.stdout);
+        list.lines()
+            .find_map(|ln| {
+                let mut cols = ln.split_whitespace();
+                let _id   = cols.next()?;
+                let name  = cols.next()?;             // column #1
+                if name.contains(fragment) { Some(name.to_owned()) } else { None }
+            })
+    }
 }

client/src/tests/integration_test.rs

@@ -9,7 +9,7 @@ mod tests {
     fn test_keycode_mapping() {
         assert_eq!(keycode_to_usage(Key::KEY_A), Some(0x04));
         assert_eq!(keycode_to_usage(Key::KEY_Z), Some(0x1D));
-        assert_eq!(keycode_to_usage(Key::KEY_LEFTCTRL), Some(0)); // this is "handled" by is_modifier
+        assert_eq!(keycode_to_usage(Key::KEY_LEFTCTRL), Some(0));
         assert!(is_modifier(Key::KEY_LEFTCTRL).is_some());
     }
-}
+}

scripts/install/server.sh

@@ -20,7 +20,8 @@ sudo pacman -Syq --needed --noconfirm git \
                                       gst-plugins-bad-libs \
                                       gst-plugins-ugly \
                                       gst-libav \
-                                      tcpdump
+                                      tcpdump \
+                                      lsof
 if ! command -v yay >/dev/null 2>&1; then
   echo "==> 1b. installing yay from AUR ..."
   sudo -u "$ORIG_USER" bash -c '
@@ -123,6 +124,7 @@ ExecStart=/usr/local/bin/lesavka-server
 Restart=always
 Environment=RUST_LOG=lesavka_server=trace,lesavka_server::audio=trace,lesavka_server::video=trace,lesavka_server::gadget=trace
 Environment=RUST_BACKTRACE=1
+Environment=GST_DEBUG="*:2,alsasink:6,alsasrc:6"
 Restart=always
 RestartSec=5
 StandardError=append:/tmp/lesavka-server.stderr

scripts/manual/audio-clip-fetch.sh

@@ -4,7 +4,7 @@
 # Pull & play the most recent 1 s AAC clip from lesavka‑server
 PI_HOST="nikto@192.168.42.253"   # adjust
 REMOTE_DIR="/tmp"
-DEST="$(mktemp -u).aac"
+DEST="$(mktemp -u).wav"
 
 scp "${PI_HOST}:${REMOTE_DIR}/ear-*.aac" "$DEST" 2>/dev/null \
   || { echo "❌ no clip files yet"; exit 1; }

scripts/manual/audio-mic-fetch.sh

@@ -0,0 +1,14 @@
+#!/usr/bin/env bash
+# scripts/manual/audio-clip-fetch.sh
+
+# Pull & play the most recent 1 s AAC clip from lesavka‑server
+PI_HOST="nikto@192.168.42.253"   # adjust
+REMOTE_DIR="/tmp"
+DEST="$(mktemp -u).aac"
+
+scp "${PI_HOST}:${REMOTE_DIR}/ear-*.aac" "$DEST" 2>/dev/null \
+  || { echo "❌ no clip files yet"; exit 1; }
+
+LATEST=$(ls -1t ear-*.aac | head -n1)
+echo "🎧 playing ${LATEST} ..."
+gst-play-1.0 --quiet "${LATEST}"

server/src/main.rs

@@ -184,6 +184,13 @@ impl Relay for Handler {
         tokio::spawn(async move {
             let mut inbound = req.into_inner();
             while let Some(pkt) = inbound.next().await.transpose()? {
+                static CNT: std::sync::atomic::AtomicU64 =
+                    std::sync::atomic::AtomicU64::new(0);
+                let n = CNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                if n < 10 || n % 300 == 0 {
+                    trace!("🎤⬇ srv  pkt#{n}  {} bytes", pkt.data.len());
+                }
+
                 let mut buf = gst::Buffer::from_slice(pkt.data);
                 buf.get_mut().unwrap()
                     .set_pts(Some(gst::ClockTime::from_useconds(pkt.pts)));