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

clarified to claude i needed it to control the speaker volume and not the mic volume, also asked it to fix mic streaming to the PC

Browse Source
This commit is contained in:
Lucca Pirovano
2026-07-15 14:45:13 -04:00
parent 2a3e0bae23
commit 7fa946f185
1 changed files with 100 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
naowalk.py
+100 -39
View File
@@ -29,36 +29,72 @@ class SoundReceiver:
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).
# what produces clicking/silence).
SILENCE_CHUNK = b"\x00" * 4096
def __init__(self, volume_getter):
self.volume_getter = volume_getter # callable -> float, e.g. lambda: self.volume
def __init__(self, output_device_index=None):
self.running = True
self.underrun_count = 0
self.chunks_received = 0
self._write_errors = 0
self._peak_since_print = 0
self._last_status_print = 0
if HAS_AUDIO:
self.p = pyaudio.PyAudio()
# Print every playback-capable device so you can tell whether
# PortAudio's "default" is actually your laptop speakers (on
# Linux it very often picks an HDMI or dummy device instead).
print("🔈 Available output devices:")
for i in range(self.p.get_device_count()):
info = self.p.get_device_info_by_index(i)
if info.get("maxOutputChannels", 0) > 0:
print(f" [{i}] {info['name']}")
try:
chosen = (self.p.get_device_info_by_index(output_device_index)
if output_device_index is not None
else self.p.get_default_output_device_info())
print(f"🔈 Using output device: [{chosen['index']}] {chosen['name']}"
f"{' (forced via --audio-device-index)' if output_device_index is not None else ' (default)'}")
except Exception as e:
print(f"⚠️ Could not resolve output device: {e}")
# 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,
output_device_index=output_device_index,
frames_per_buffer=2048)
self._play_test_tone()
# 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.
# stalls NAOqi's callback thread, which 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()
def _play_test_tone(self):
# Plays a short beep straight to the output device, independent of
# any data from the robot. If you don't hear this, the problem is
# your laptop's audio routing/volume, not the robot or the network -
# troubleshoot with alsamixer / your system's sound settings first.
try:
duration, freq, rate = 0.3, 440.0, 16000
t = np.linspace(0, duration, int(rate * duration), False)
tone = (np.sin(freq * t * 2 * np.pi) * 12000).astype(np.int16)
self.stream.write(tone.tobytes())
print("🔔 Played a test beep. If you didn't hear it, this is a laptop "
"audio output/device problem, not a robot/network problem.")
except Exception as e:
print(f"⚠️ Test tone failed to play: {e}")
# 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):
@@ -106,19 +142,22 @@ class SoundReceiver:
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()}")
f"{self.underrun_count} underruns, {self._write_errors} write errors, "
f"queue depth {self.queue.qsize()}, peak level {self._peak_since_print}/32767")
self._peak_since_print = 0
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
samples = np.frombuffer(raw_bytes, dtype=np.int16)
if samples.size:
peak = int(np.abs(samples).max())
if peak > self._peak_since_print:
self._peak_since_print = peak
except Exception as e:
self._write_errors += 1
if self._write_errors <= 3:
print(f"⚠️ audio write error: {e}")
def close(self):
self.running = False
@@ -139,7 +178,7 @@ class SoundReceiver:
# MAIN TELEOP CLASS
# ==========================================
class NaoTeleop:
def __init__(self, session):
def __init__(self, session, audio_device_index=None):
self.session = session
self.motion = session.service("ALMotion")
self.posture = session.service("ALRobotPosture")
@@ -156,6 +195,7 @@ class NaoTeleop:
# Audio setup variables
self.audio_device = None
self.audio_service_name = "SoundReceiver"
self.audio_device_index = audio_device_index # which laptop output device to play the mic feed on
self.running = True
self.head_yaw = 0.0
@@ -163,20 +203,25 @@ 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
# Volume of the NAO's own onboard speakers (0-100), controlled via
# ALAudioDevice.setOutputVolume. This is separate from - and has
# nothing to do with - the laptop playback of the robot's mic feed.
self.volume = 50
# Chatbox variables
self.typing_mode = False
self.chat_message = ""
# Get ALAudioDevice up front - this is needed for speaker volume
# control regardless of whether PyAudio/mic-listening is available.
self._init_audio_device()
# Initialize Video
self._init_video_maxfps()
# Initialize Audio
# Initialize mic listening (PyAudio -> laptop speakers)
if HAS_AUDIO:
self._init_audio()
self._init_mic_stream()
# Initialize Pygame
pygame.init()
@@ -186,6 +231,15 @@ class NaoTeleop:
self.clock = pygame.time.Clock()
self.update_title()
def _init_audio_device(self):
try:
self.audio_device = self.session.service("ALAudioDevice")
self.volume = self.audio_device.getOutputVolume()
print(f"🔊 NAO speaker volume: {self.volume}%")
except Exception as e:
print(f"⚠️ Could not reach ALAudioDevice for speaker volume control: {e}")
self.audio_device = None
def _init_video_maxfps(self):
try:
self.video = self.session.service("ALVideoDevice")
@@ -194,10 +248,12 @@ class NaoTeleop:
except:
print("❌ Camera not available")
def _init_audio(self):
def _init_mic_stream(self):
if not self.audio_device:
print("❌ Can't start mic listening - ALAudioDevice unavailable")
return
try:
self.audio_device = self.session.service("ALAudioDevice")
self.sound_receiver = SoundReceiver(lambda: self.volume)
self.sound_receiver = SoundReceiver(output_device_index=self.audio_device_index)
# 1. Register the local service
self.session.registerService(self.audio_service_name, self.sound_receiver)
@@ -214,7 +270,6 @@ class NaoTeleop:
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:
@@ -270,12 +325,16 @@ class NaoTeleop:
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}%")
pygame.display.set_caption(f"NAO Teleop | {mode}Battery: {self.battery_level}% | NAO Vol: {self.volume}%")
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.volume = int(min(100, max(0, self.volume + delta)))
if self.audio_device:
try:
self.audio_device.setOutputVolume(self.volume)
except Exception as e:
print(f"⚠️ Failed to set NAO speaker volume: {e}")
print(f"🔊 NAO speaker volume: {self.volume}%")
self.update_title()
def check_battery(self):
@@ -308,8 +367,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(" - / = = NAO speaker volume down / up")
print(" 0 = Mute NAO speaker")
print(" ESC = Quit")
while self.running:
@@ -351,13 +410,11 @@ class NaoTeleop:
self.motion.stopMove()
self.last_batt_check = 0
elif event.key in (pygame.K_MINUS, pygame.K_KP_MINUS):
self.change_volume(-0.1)
self.change_volume(-5)
elif event.key in (pygame.K_EQUALS, pygame.K_PLUS, pygame.K_KP_PLUS):
self.change_volume(0.1)
self.change_volume(5)
elif event.key == pygame.K_0:
self.volume = 0.0
print("🔇 Muted")
self.update_title()
self.change_volume(-self.volume)
# Walking (Only if not typing)
if not self.typing_mode:
@@ -425,6 +482,10 @@ if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--ip", type=str, default="127.0.0.1")
parser.add_argument("--port", type=int, default=9559)
parser.add_argument("--audio-device-index", type=int, default=None,
help="Force a specific PyAudio output device index for the robot's "
"mic feed (see the '🔈 Available output devices' list printed "
"at startup) if the system default isn't your speakers.")
args = parser.parse_args()
session = qi.Session()
@@ -435,4 +496,4 @@ if __name__ == "__main__":
print(f"❌ Connection failed: {e}")
sys.exit(1)
NaoTeleop(session).run()
NaoTeleop(session, audio_device_index=args.audio_device_index).run()