From 95ee2fc036696f8ff734736ef25bc46b22bdd69d Mon Sep 17 00:00:00 2001 From: Adolfo Reyna Date: Wed, 17 Jun 2026 23:00:32 -0400 Subject: [PATCH] Create local audio loopback utility script to test microphone and speaker hardware locally using physical buttons --- demo_audio_loopback.py | 214 +++++++++++++++++++++++++++++++++++++++++ upload_files.py | 1 + 2 files changed, 215 insertions(+) create mode 100644 demo_audio_loopback.py diff --git a/demo_audio_loopback.py b/demo_audio_loopback.py new file mode 100644 index 0000000..416614c --- /dev/null +++ b/demo_audio_loopback.py @@ -0,0 +1,214 @@ +import time +import machine +from machine import Pin, I2S +import board_config +from button_util import BoardButtons +import audio_util + +def play_raw_pcm(filename, channels=2, rate=16000, bits=16, volume=90): + from audio_util import ES8311 + + # 1. Start MCLK PWM using board config parameters + mclk_pwm = None + if board_config.audio_mclk_pin is not None: + mclk_pin = Pin(board_config.audio_mclk_pin, Pin.OUT) + mclk_pwm = machine.PWM(mclk_pin) + mclk_pwm.freq(board_config.audio_mclk_freq) + mclk_pwm.duty_u16(32768) + + # 2. Wake up and configure ES8311 DAC + dac = ES8311(board_config.i2c_bus) + dac.init(sample_rate=rate) + dac.set_volume(volume) + + # 3. Configure I2S TX + i2s_format = I2S.MONO if channels == 1 else I2S.STEREO + i2s = I2S(1, + sck=Pin(board_config.audio_i2s_sck), + ws=Pin(board_config.audio_i2s_ws), + sd=Pin(board_config.audio_i2s_tx_sd), + mode=I2S.TX, + ibuf=4096, + rate=rate, + bits=bits, + format=i2s_format) + + # 4. Enable Speaker Amplifier + on_val = 0 if board_config.audio_amp_active_level == 0 else 1 + off_val = 1 if board_config.audio_amp_active_level == 0 else 0 + amp_pin = Pin(board_config.audio_amp_pin, Pin.OUT, value=on_val) + + try: + print(f"Streaming raw audio to speaker from '{filename}'...") + with open(filename, 'rb') as f: + buf = bytearray(2048) + while True: + bytes_read = f.readinto(buf) + if bytes_read == 0: + break + i2s.write(buf[:bytes_read]) + except Exception as e: + print("Error during raw playback:", e) + finally: + time.sleep_ms(100) # Let buffer play out + amp_pin.value(off_val) # Disable amp + i2s.deinit() + if mclk_pwm: + mclk_pwm.deinit() + print("Playback complete.") + +def main(): + display = board_config.display_instance + print("=== Dynamic Audio Loopback Utility Script ===") + print(f"Board Detected: {board_config.BOARD_TYPE}") + + # Initialize buttons + buttons = BoardButtons() + + # Helper to check if trigger is active + def is_talk_trigger_active(): + if board_config.touch: + return board_config.touch.is_touched() + elif buttons and buttons.key: + return buttons.key.is_pressed() + return False + + filename = "local_audio_test.pcm" + + while True: + if display: + display.clear(0) + if board_config.BOARD_TYPE == 'WAVESHARE_RLCD': + display.text("RLCD Audio Loopback Test", 10, 10, 1) + display.line(10, 20, 390, 20, 1) + display.text("1. Press & HOLD KEY button to record", 15, 60, 1) + display.text("2. Release key to play back", 15, 80, 1) + display.text("Ready...", 15, 120, 1) + else: + display.text("Touch Audio Loopback Test", 10, 10, 1) + display.line(10, 20, 310, 20, 1) + display.text("1. Press & HOLD screen/KEY to record", 10, 50, 1) + display.text("2. Release to play back", 10, 70, 1) + display.text("Ready...", 10, 100, 1) + display.show() + + print("Ready: Press and hold key/screen to record...") + while not is_talk_trigger_active(): + time.sleep_ms(30) + + print("Recording started! Speak into the microphone...") + if display: + display.clear(0) + if board_config.BOARD_TYPE == 'WAVESHARE_RLCD': + display.text("RLCD Audio Loopback Test", 10, 10, 1) + display.line(10, 20, 390, 20, 1) + display.text("RECORDING NOW!", 15, 80, 1) + display.text("Keep holding key...", 15, 100, 1) + else: + display.text("Touch Audio Loopback Test", 10, 10, 1) + display.line(10, 20, 310, 20, 1) + display.text("RECORDING NOW!", 10, 60, 1) + display.text("Keep holding trigger...", 10, 80, 1) + display.show() + + # We record as long as the button is pressed (or up to 10 seconds max) + + # 1. Start MCLK PWM using board config parameters + mclk_pwm = None + if board_config.audio_mclk_pin is not None: + mclk_pin = Pin(board_config.audio_mclk_pin, Pin.OUT) + mclk_pwm = machine.PWM(mclk_pin) + mclk_pwm.freq(board_config.audio_mclk_freq) + mclk_pwm.duty_u16(32768) + + # 2. Configure I2S RX (Stereo 16kHz) + i2s_rx = I2S(1, + sck=Pin(board_config.audio_i2s_sck), + ws=Pin(board_config.audio_i2s_ws), + sd=Pin(board_config.audio_i2s_rx_sd), + mode=I2S.RX, + ibuf=8000, + rate=16000, + bits=16, + format=I2S.STEREO) + + # 3. Wake up and configure the microphone chip (ES7210 vs ES8311) + if board_config.audio_mic_codec == "ES7210": + from audio_util import ES7210 + mic_adc = ES7210(board_config.i2c_bus) + mic_adc.init(sample_rate=16000, bit_width=16) + else: + from audio_util import ES8311 + mic_adc = ES8311(board_config.i2c_bus) + if mic_adc.init(sample_rate=16000): + mic_adc.set_volume(80) + try: + mic_adc._write(0x14, 0x1A) # Enable analog mic input & PGA + mic_adc._write(0x16, 0x01) # Enable +6dB gain boost + mic_adc._write(0x17, 0xC8) # Set ADC digital volume + except: + pass + + # Record loop + buffer = bytearray(2048) + total_bytes = 0 + start_time = time.ticks_ms() + max_duration_ms = 10000 # 10 seconds limit + + try: + with open(filename, 'wb') as f: + while is_talk_trigger_active() and time.ticks_diff(time.ticks_ms(), start_time) < max_duration_ms: + bytes_read = i2s_rx.readinto(buffer) + if bytes_read > 0: + f.write(buffer[:bytes_read]) + total_bytes += bytes_read + except Exception as e: + print("Recording failed:", e) + finally: + i2s_rx.deinit() + if mclk_pwm: + mclk_pwm.deinit() + + print(f"Trigger released! Recorded {total_bytes} bytes to '{filename}'.") + + # Debounce release + time.sleep_ms(150) + + # 4. Playback + if display: + display.clear(0) + if board_config.BOARD_TYPE == 'WAVESHARE_RLCD': + display.text("RLCD Audio Loopback Test", 10, 10, 1) + display.line(10, 20, 390, 20, 1) + display.text("PLAYING BACK RESPONSE...", 15, 80, 1) + display.text(f"Bytes: {total_bytes}", 15, 100, 1) + else: + display.text("Touch Audio Loopback Test", 10, 10, 1) + display.line(10, 20, 310, 20, 1) + display.text("PLAYING BACK RESPONSE...", 10, 60, 1) + display.text(f"Bytes: {total_bytes}", 10, 80, 1) + display.show() + + play_raw_pcm(filename, channels=2, rate=16000, bits=16, volume=95) + + if display: + display.clear(0) + if board_config.BOARD_TYPE == 'WAVESHARE_RLCD': + display.text("RLCD Audio Loopback Test", 10, 10, 1) + display.line(10, 20, 390, 20, 1) + display.text("Playback Finished!", 15, 70, 1) + display.text("Release trigger to restart...", 15, 95, 1) + else: + display.text("Touch Audio Loopback Test", 10, 10, 1) + display.line(10, 20, 310, 20, 1) + display.text("Playback Finished!", 10, 50, 1) + display.text("Release trigger to restart...", 10, 75, 1) + display.show() + + # Wait for release of button/touch to debounce before ready again + while is_talk_trigger_active(): + time.sleep_ms(30) + time.sleep_ms(500) + +if __name__ == "__main__": + main() diff --git a/upload_files.py b/upload_files.py index 0d5af67..68a34f8 100644 --- a/upload_files.py +++ b/upload_files.py @@ -69,6 +69,7 @@ def main(): ("lib/audio_util.py", "lib/audio_util.py"), ("lib/rtc_util.py", "lib/rtc_util.py"), ("lib/shtc3_util.py", "lib/shtc3_util.py"), + ("demo_audio_loopback.py", "demo_audio_loopback.py"), ("video_stream.py", "video_stream.py"), ("mcp_server.py", "mcp_server.py"), ("boot.py", "boot.py"),