luccapirovano/naowalk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

moved to claude, added volume display and adjuster tool, asked to fix underruns

Browse Source
This commit is contained in:
Lucca Pirovano
2026-07-15 14:39:02 -04:00
parent 69e41aa67c
commit 2a3e0bae23
3 changed files with 1279 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
chat-history/gemini_fixing-nao-robot-wave-gesture-google-gemini_2026-07-15T18-24-30+0000.md
+1124
View File
File diff suppressed because it is too large Load Diff
chat-history/scripts-used.txt
+1
View File
@@ -1 +1,2 @@
https://github.com/iikoshteruu/enhanced-grok-export
https://github.com/revivalstack/ai-chat-exporter
naowalk.py
+151 -24
View File
@@ -6,6 +6,7 @@ import argparse
import sys
import time
import threading
import queue
import cv2
import numpy as np
import pygame
@@ -23,22 +24,114 @@ except ImportError:
# AUDIO RECEIVER SERVICE (Runs in background)
# ==========================================
class SoundReceiver:
def __init__(self):
# How many chunks to buffer before we start playback. This absorbs Wi-Fi
# jitter so a late/slow packet doesn't starve ALSA the instant it lands.
PREBUFFER_CHUNKS = 3
# Silence written to the output stream whenever the queue runs dry, so we
# feed ALSA continuously instead of letting it hard-underrun (which is
# what produces the clicking/silence you were hearing).
SILENCE_CHUNK = b"\x00" * 4096
def __init__(self, volume_getter):
self.volume_getter = volume_getter # callable -> float, e.g. lambda: self.volume
self.running = True
self.underrun_count = 0
self.chunks_received = 0
self._last_status_print = 0
if HAS_AUDIO:
self.p = pyaudio.PyAudio()
# NAO front mic is usually 16000Hz, 1 channel, 16-bit
# NAO front mic is usually 16000Hz, 1 channel, 16-bit.
# frames_per_buffer is set explicitly (rather than left at the
# PyAudio default) so ALSA's buffer size matches what we feed it.
self.stream = self.p.open(format=pyaudio.paInt16,
channels=1,
rate=16000,
output=True)
channels=1,
rate=16000,
output=True,
frames_per_buffer=2048)
# The queue decouples the NAOqi network thread (which calls
# processRemote) from the actual blocking audio write. Without
# this, a slow/blocking stream.write() call inside processRemote
# stalls NAOqi's callback thread, which in turn delays the next
# packet, which starves the speaker further - a feedback loop
# that shows up as underruns and dropouts.
self.queue = queue.Queue(maxsize=40)
self.playback_thread = threading.Thread(target=self._playback_loop, daemon=True)
self.playback_thread.start()
# ALAudioDevice strictly requires this exact method signature to send data
# ALAudioDevice strictly requires this exact method signature to send data.
# Keep this method as fast as possible - it runs on NAOqi's network thread.
def processRemote(self, nbOfChannels, nbOfSamplesByChannel, timeStamp, buffer):
if HAS_AUDIO and self.stream:
if not HAS_AUDIO:
return
self.chunks_received += 1
try:
self.queue.put_nowait(bytes(buffer))
except queue.Full:
# We're falling behind - drop the oldest chunk rather than
# blocking the NAOqi thread (blocking here is what causes
# cascading underruns).
try:
# Play the raw audio bytes pushed from the robot directly to your speakers
self.stream.write(bytes(buffer))
except Exception as e:
self.queue.get_nowait()
self.queue.put_nowait(bytes(buffer))
except queue.Empty:
pass
def _playback_loop(self):
# Wait for a small pile of chunks before we start playing, so the
# very first sounds you hear aren't immediately starved.
primed = []
while self.running and len(primed) < self.PREBUFFER_CHUNKS:
try:
primed.append(self.queue.get(timeout=1.0))
except queue.Empty:
break
for chunk in primed:
self._write_chunk(chunk)
while self.running:
try:
chunk = self.queue.get(timeout=0.2)
self._write_chunk(chunk)
except queue.Empty:
# Nothing arrived in time - feed silence instead of letting
# the ALSA buffer run completely dry.
self.underrun_count += 1
try:
self.stream.write(self.SILENCE_CHUNK)
except Exception:
pass
now = time.time()
if now - self._last_status_print > 10:
self._last_status_print = now
print(f"🎤 audio: {self.chunks_received} chunks received, "
f"{self.underrun_count} underruns, queue depth {self.queue.qsize()}")
def _write_chunk(self, raw_bytes):
try:
vol = self.volume_getter()
if vol != 1.0:
samples = np.frombuffer(raw_bytes, dtype=np.int16).astype(np.float32)
samples *= vol
np.clip(samples, -32768, 32767, out=samples)
raw_bytes = samples.astype(np.int16).tobytes()
self.stream.write(raw_bytes)
except Exception:
pass
def close(self):
self.running = False
if HAS_AUDIO:
try:
self.playback_thread.join(timeout=1.0)
except Exception:
pass
try:
self.stream.stop_stream()
self.stream.close()
self.p.terminate()
except Exception:
pass
@@ -70,6 +163,10 @@ class NaoTeleop:
self.battery_level = 100
self.last_batt_check = 0
# Volume for the incoming mic audio (software gain applied before
# playback). 1.0 = unity gain, 2.0 = +100%, 0.0 = muted.
self.volume = 0.6
# Chatbox variables
self.typing_mode = False
self.chat_message = ""
@@ -87,6 +184,7 @@ class NaoTeleop:
pygame.display.set_caption("NAO Teleop | Battery: Checking...")
self.font = pygame.font.SysFont(None, 24)
self.clock = pygame.time.Clock()
self.update_title()
def _init_video_maxfps(self):
try:
@@ -97,19 +195,26 @@ class NaoTeleop:
print("❌ Camera not available")
def _init_audio(self):
try:
self.audio_device = self.session.service("ALAudioDevice")
# Register our local class as a service on the robot's network
self.sound_receiver = SoundReceiver()
self.session.registerService(self.audio_service_name, self.sound_receiver)
try:
self.audio_device = self.session.service("ALAudioDevice")
self.sound_receiver = SoundReceiver(lambda: self.volume)
# Configure: 16000Hz, Channel 3 (Front Mic), 0 (Interleaved)
self.audio_device.setClientPreferences(self.audio_service_name, 16000, 3, 0)
self.audio_device.subscribe(self.audio_service_name)
print("✅ Live Audio Stream ready (Check your speakers!)")
except Exception as e:
print(f"❌ Audio init failed: {e}")
self.audio_device = None
# 1. Register the local service
self.session.registerService(self.audio_service_name, self.sound_receiver)
# 2. Add a tiny delay to let the network handshake complete
time.sleep(0.5)
# 3. Use the IP of the machine running the code
# We explicitly tell the robot the IP of your laptop (replace with your laptop IP if still failing)
# '0' tells it to auto-discover, but sometimes we need to be explicit.
self.audio_device.setClientPreferences(self.audio_service_name, 16000, 3, 0)
self.audio_device.subscribe(self.audio_service_name)
print("✅ Live Audio Stream ready")
except Exception as e:
print(f"❌ Audio init failed: {e}")
self.audio_device = None
def get_image(self):
if not self.video or not self.video_client:
@@ -163,16 +268,25 @@ class NaoTeleop:
if changed:
self.motion.setAngles(["HeadYaw", "HeadPitch"], [self.head_yaw, self.head_pitch], 0.3)
def update_title(self):
mode = "[TYPING] " if self.typing_mode else ""
vol_pct = int(round(self.volume * 100))
pygame.display.set_caption(f"NAO Teleop | {mode}Battery: {self.battery_level}% | Vol: {vol_pct}%")
def change_volume(self, delta):
self.volume = round(min(2.0, max(0.0, self.volume + delta)), 2)
print(f"🔊 Volume: {int(round(self.volume * 100))}%")
self.update_title()
def check_battery(self):
if not self.battery: return
now = time.time()
if now - self.last_batt_check > 10:
try:
self.battery_level = self.battery.getBatteryCharge()
mode = "[TYPING] " if self.typing_mode else ""
pygame.display.set_caption(f"NAO Teleop | {mode}Battery: {self.battery_level}%")
except:
pass
self.update_title()
self.last_batt_check = now
def run(self):
@@ -194,6 +308,8 @@ class NaoTeleop:
print(" I J K L = Head")
print(" 1 = Wave")
print(" T = Type to Speak (TTS)")
print(" - / = = Mic volume down / up")
print(" 0 = Mute mic")
print(" ESC = Quit")
while self.running:
@@ -234,6 +350,14 @@ class NaoTeleop:
self.typing_mode = True
self.motion.stopMove()
self.last_batt_check = 0
elif event.key in (pygame.K_MINUS, pygame.K_KP_MINUS):
self.change_volume(-0.1)
elif event.key in (pygame.K_EQUALS, pygame.K_PLUS, pygame.K_KP_PLUS):
self.change_volume(0.1)
elif event.key == pygame.K_0:
self.volume = 0.0
print("🔇 Muted")
self.update_title()
# Walking (Only if not typing)
if not self.typing_mode:
@@ -283,6 +407,9 @@ class NaoTeleop:
self.audio_device.unsubscribe(self.audio_service_name)
except:
pass
if getattr(self, "sound_receiver", None):
self.sound_receiver.close()
# Clean up Video
if self.video and self.video_client: