Create local audio loopback utility script to test microphone and speaker hardware locally using physical buttons
This commit is contained in:
@@ -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()
|
||||||
@@ -69,6 +69,7 @@ def main():
|
|||||||
("lib/audio_util.py", "lib/audio_util.py"),
|
("lib/audio_util.py", "lib/audio_util.py"),
|
||||||
("lib/rtc_util.py", "lib/rtc_util.py"),
|
("lib/rtc_util.py", "lib/rtc_util.py"),
|
||||||
("lib/shtc3_util.py", "lib/shtc3_util.py"),
|
("lib/shtc3_util.py", "lib/shtc3_util.py"),
|
||||||
|
("demo_audio_loopback.py", "demo_audio_loopback.py"),
|
||||||
("video_stream.py", "video_stream.py"),
|
("video_stream.py", "video_stream.py"),
|
||||||
("mcp_server.py", "mcp_server.py"),
|
("mcp_server.py", "mcp_server.py"),
|
||||||
("boot.py", "boot.py"),
|
("boot.py", "boot.py"),
|
||||||
|
|||||||
Reference in New Issue
Block a user