Adolfo Reyna
237 lines
9.4 KiB
Python
237 lines
9.4 KiB
Python
import time
|
|
import machine
|
|
from machine import Pin, I2S
|
|
import board_config
|
|
|
|
def play_ram_pcm(audio_chunks, 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 RAM ({len(audio_chunks)} chunks)...")
|
|
for chunk in audio_chunks:
|
|
i2s.write(chunk)
|
|
except Exception as e:
|
|
print("Error during RAM 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 RAM-based Audio Loopback Script ===")
|
|
print(f"Board Detected: {board_config.BOARD_TYPE}")
|
|
|
|
# Initialize buttons using polling Pins instead of BoardButtons class
|
|
key_pin = Pin(18, Pin.IN, Pin.PULL_UP)
|
|
boot_pin = Pin(0, Pin.IN, Pin.PULL_UP)
|
|
|
|
# Helper to check if trigger is active
|
|
def is_talk_trigger_active():
|
|
if board_config.touch:
|
|
return board_config.touch.is_touched()
|
|
return key_pin.value() == 0
|
|
|
|
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 KEY button to record 10s", 15, 60, 1)
|
|
display.text("2. Playback will start automatically", 15, 80, 1)
|
|
display.text("Ready (RAM-based)...", 15, 120, 1)
|
|
else:
|
|
display.text("Touch Audio Loopback Test", 10, 10, 1)
|
|
display.line(10, 20, 310, 20, 1)
|
|
display.text("1. Press screen/KEY to record 10s", 10, 50, 1)
|
|
display.text("2. Playback starts automatically", 10, 70, 1)
|
|
display.text("Ready (RAM-based)...", 10, 100, 1)
|
|
display.show()
|
|
|
|
print("Ready: Press and hold key/screen to record...")
|
|
start_wait = time.ticks_ms()
|
|
while not is_talk_trigger_active():
|
|
if boot_pin.value() == 0 or time.ticks_diff(time.ticks_ms(), start_wait) > 120000:
|
|
print("Exit condition met. Deleting flag and resetting...")
|
|
if display:
|
|
display.clear(0)
|
|
if board_config.BOARD_TYPE == 'WAVESHARE_RLCD':
|
|
display.text("Exiting test...", 15, 60, 1)
|
|
display.text("Rebooting to normal...", 15, 80, 1)
|
|
else:
|
|
display.text("Exiting test...", 10, 50, 1)
|
|
display.text("Rebooting to normal...", 10, 70, 1)
|
|
display.show()
|
|
time.sleep(1)
|
|
try:
|
|
import os
|
|
os.remove("run_loopback.txt")
|
|
except:
|
|
pass
|
|
import machine
|
|
machine.reset()
|
|
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: 10 seconds...", 15, 80, 1)
|
|
display.text("Speak into microphone!", 15, 100, 1)
|
|
else:
|
|
display.text("Touch Audio Loopback Test", 10, 10, 1)
|
|
display.line(10, 20, 310, 20, 1)
|
|
display.text("RECORDING: 10 seconds...", 10, 60, 1)
|
|
display.text("Speak now!", 10, 80, 1)
|
|
display.show()
|
|
|
|
# 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) - ibuf set to 16000 for safety
|
|
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=16000,
|
|
rate=16000,
|
|
bits=16,
|
|
format=I2S.STEREO)
|
|
|
|
# 3. Wake up and configure the microphone chip (ES7210 vs ES8311)
|
|
init_ok = False
|
|
if board_config.audio_mic_codec == "ES7210":
|
|
from audio_util import ES7210
|
|
mic_adc = ES7210(board_config.i2c_bus)
|
|
init_ok = 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):
|
|
init_ok = True
|
|
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
|
|
|
|
if not init_ok:
|
|
print("Microphone codec initialization failed! Aborting recording.")
|
|
i2s_rx.deinit()
|
|
if mclk_pwm:
|
|
mclk_pwm.deinit()
|
|
if display:
|
|
display.clear(0)
|
|
display.text("Codec Init Failed!", 15, 80, 1)
|
|
display.show()
|
|
time.sleep(3)
|
|
continue
|
|
|
|
# Record loop to RAM - using ticks_ms for safe timing
|
|
buffer = bytearray(2048)
|
|
audio_chunks = []
|
|
total_bytes = 0
|
|
start_rec_time = time.ticks_ms()
|
|
max_duration_ms = 10000 # 10 seconds
|
|
|
|
try:
|
|
while time.ticks_diff(time.ticks_ms(), start_rec_time) < max_duration_ms:
|
|
bytes_read = i2s_rx.readinto(buffer)
|
|
if bytes_read > 0:
|
|
audio_chunks.append(bytes(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"Recorded {total_bytes} bytes in RAM ({len(audio_chunks)} chunks).")
|
|
|
|
# Wait for release of key/trigger to debounce
|
|
time.sleep_ms(200)
|
|
|
|
# 4. Playback from RAM
|
|
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_ram_pcm(audio_chunks, 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()
|