zyphlar
229 lines
7.8 KiB
Python
229 lines
7.8 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
upload_track.py — serial audio uploader for baby_mobile_v2 POC_INTERNAL_FLASH mode
|
||
|
||
Usage:
|
||
python upload_track.py <port> <track_num> <file>
|
||
|
||
Supported formats: .raw, .wav, .mp3, .ogg, .flac, .aac, .m4a (anything ffmpeg handles)
|
||
|
||
Requires: pyserial (pip install pyserial)
|
||
ffmpeg in PATH for non-WAV/RAW files
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import shutil
|
||
import struct
|
||
import subprocess
|
||
import serial
|
||
import time
|
||
|
||
CHUNK_SIZE = 512
|
||
BAUD_RATE = 115200
|
||
TIMEOUT_S = 10.0
|
||
|
||
RAW_EXTS = {'.raw'}
|
||
WAV_EXTS = {'.wav'}
|
||
|
||
|
||
def convert_to_pcm(path: str) -> bytes:
|
||
ext = os.path.splitext(path)[1].lower()
|
||
|
||
if ext in RAW_EXTS:
|
||
with open(path, "rb") as f:
|
||
return f.read()
|
||
|
||
if ext in WAV_EXTS:
|
||
with open(path, "rb") as f:
|
||
data = f.read()
|
||
if data[:4] == b"RIFF":
|
||
try:
|
||
audio_format = struct.unpack_from("<H", data, 20)[0]
|
||
num_channels = struct.unpack_from("<H", data, 22)[0]
|
||
sample_rate = struct.unpack_from("<I", data, 24)[0]
|
||
bits_per_samp = struct.unpack_from("<H", data, 34)[0]
|
||
needs_convert = (audio_format != 1 or num_channels != 1
|
||
or sample_rate != 8000 or bits_per_samp != 8)
|
||
except Exception:
|
||
needs_convert = True
|
||
|
||
if not needs_convert:
|
||
print("WAV is already 8-bit mono 8kHz — stripping header")
|
||
return data[44:]
|
||
print("WAV needs resampling — converting via ffmpeg")
|
||
# fall through to ffmpeg conversion
|
||
|
||
# Use ffmpeg for everything else (mp3, ogg, flac, aac, non-conformant wav...)
|
||
if not shutil.which('ffmpeg'):
|
||
sys.exit(
|
||
"ERROR: ffmpeg not found. Install it or convert manually:\n"
|
||
f" ffmpeg -i \"{path}\" -ar 8000 -ac 1 -f u8 -acodec pcm_u8 out.raw"
|
||
)
|
||
|
||
print(f"Converting {os.path.basename(path)} via ffmpeg...")
|
||
result = subprocess.run(
|
||
['ffmpeg', '-y', '-i', path,
|
||
'-ar', '8000', '-ac', '1', '-f', 'u8', '-acodec', 'pcm_u8', 'pipe:1'],
|
||
capture_output=True
|
||
)
|
||
if result.returncode != 0:
|
||
sys.exit(f"ERROR: ffmpeg failed:\n{result.stderr.decode(errors='replace')[-400:]}")
|
||
|
||
pcm = result.stdout
|
||
duration = len(pcm) / 8000
|
||
print(f" Converted: {len(pcm):,} bytes ({duration:.1f}s)")
|
||
return pcm
|
||
|
||
|
||
def upload(port: str, track_num: int, pcm: bytes) -> None:
|
||
total = len(pcm)
|
||
print(f"Uploading {total} bytes as track {track_num} via {port} ...")
|
||
|
||
# dsrdtr=False / rtscts=False prevents Windows from toggling DTR/RTS on
|
||
# port open, which would reset the nRF52840 and stall the 5-second boot loop.
|
||
with serial.Serial(port, BAUD_RATE, timeout=2.0,
|
||
dsrdtr=False, rtscts=False) as ser:
|
||
# Drain any boot output; send a bare newline first to flush any
|
||
# partial line sitting in the firmware's line buffer.
|
||
time.sleep(0.5)
|
||
ser.reset_input_buffer()
|
||
ser.write(b"\n")
|
||
ser.flush()
|
||
time.sleep(0.1)
|
||
ser.reset_input_buffer()
|
||
|
||
# Send UPLOAD command
|
||
cmd = f"UPLOAD {track_num} {total}\n"
|
||
ser.write(cmd.encode())
|
||
ser.flush()
|
||
|
||
# Wait for READY — print everything received so failures are diagnosable
|
||
deadline = time.time() + TIMEOUT_S
|
||
while True:
|
||
if time.time() > deadline:
|
||
sys.exit("Timed out waiting for READY")
|
||
line = ser.readline().decode(errors="replace").strip()
|
||
if line == "READY":
|
||
break
|
||
if line:
|
||
print(f" firmware: {line}")
|
||
|
||
# Stream PCM in chunks; stop early if firmware sends ERR
|
||
sent = 0
|
||
err_line = None
|
||
ser.timeout = 0.05 # short timeout so we can poll for ERR while sending
|
||
while sent < total:
|
||
chunk = pcm[sent:sent + CHUNK_SIZE]
|
||
ser.write(chunk)
|
||
sent += len(chunk)
|
||
pct = sent * 100 // total
|
||
print(f"\r {sent}/{total} bytes ({pct}%) ", end="", flush=True)
|
||
# Check for early ERR from firmware (e.g. LittleFS full)
|
||
line = ser.readline().decode(errors="replace").strip()
|
||
if line.startswith("ERR"):
|
||
err_line = line
|
||
break
|
||
print()
|
||
ser.timeout = 2.0
|
||
|
||
if err_line:
|
||
print(f"ERROR from firmware: {err_line}")
|
||
# at= tells us how many bytes were accepted before the failure
|
||
if "at=" in err_line:
|
||
accepted = int(err_line.split("at=")[1].split()[0])
|
||
safe = int(accepted * 0.9) # 10% headroom
|
||
max_s = safe / 8000
|
||
print(f" LittleFS accepted {accepted} bytes before full.")
|
||
print(f" Safe clip length: ~{max_s:.1f}s")
|
||
print(f" Trim with: ffmpeg -i input.mp3 -t {max_s:.0f} -ar 8000 -ac 1 -f u8 out.raw")
|
||
sys.exit(1)
|
||
|
||
# Wait for OK
|
||
deadline = time.time() + TIMEOUT_S
|
||
while True:
|
||
if time.time() > deadline:
|
||
sys.exit("Timed out waiting for OK")
|
||
line = ser.readline().decode(errors="replace").strip()
|
||
if line.startswith("OK"):
|
||
print(f"Upload done: {line}")
|
||
break
|
||
if line.startswith("ERR"):
|
||
print(f"ERROR from firmware: {line}")
|
||
sys.exit(1)
|
||
if line:
|
||
print(f" firmware: {line}")
|
||
|
||
# Verify: request hex dump of first 256 bytes and compare
|
||
print("Verifying...")
|
||
ser.reset_input_buffer()
|
||
ser.write(f"DUMP {track_num}\n".encode())
|
||
ser.flush()
|
||
|
||
deadline = time.time() + TIMEOUT_S
|
||
fw_size = None
|
||
fw_bytes = bytearray()
|
||
while True:
|
||
if time.time() > deadline:
|
||
print("WARNING: timed out waiting for DUMP response")
|
||
break
|
||
line = ser.readline().decode(errors="replace").strip()
|
||
if not line:
|
||
continue
|
||
if line.startswith("SIZE"):
|
||
fw_size = int(line.split()[1])
|
||
elif line == "END":
|
||
break
|
||
elif line == "NO FILE":
|
||
print("ERROR: firmware says file does not exist!")
|
||
break
|
||
else:
|
||
# Parse hex bytes
|
||
for tok in line.split():
|
||
try:
|
||
fw_bytes.append(int(tok, 16))
|
||
except ValueError:
|
||
pass
|
||
|
||
if fw_size is not None:
|
||
print(f" Firmware file size: {fw_size} bytes")
|
||
if fw_size == 0:
|
||
print(" ERROR: file is empty — write failed (LittleFS full?)")
|
||
elif fw_size < total:
|
||
print(f" WARNING: only {fw_size}/{total} bytes written (LittleFS full?)")
|
||
else:
|
||
print(f" Size OK: {fw_size}/{total}")
|
||
|
||
compare_len = min(len(fw_bytes), len(pcm), 256)
|
||
if compare_len > 0:
|
||
mismatches = sum(1 for i in range(compare_len) if fw_bytes[i] != pcm[i])
|
||
if mismatches == 0:
|
||
print(f" Data OK: first {compare_len} bytes match")
|
||
else:
|
||
print(f" ERROR: {mismatches}/{compare_len} byte mismatches in first {compare_len} bytes")
|
||
print(f" Expected: {' '.join(f'{b:02X}' for b in pcm[:16])}")
|
||
print(f" Got: {' '.join(f'{b:02X}' for b in fw_bytes[:16])}")
|
||
|
||
|
||
def main():
|
||
if len(sys.argv) != 4:
|
||
print(__doc__)
|
||
sys.exit(1)
|
||
|
||
port = sys.argv[1]
|
||
track_num = int(sys.argv[2])
|
||
path = sys.argv[3]
|
||
|
||
if not 0 <= track_num < 32:
|
||
sys.exit("track_num must be 0–31")
|
||
|
||
pcm = convert_to_pcm(path)
|
||
if not pcm:
|
||
sys.exit("File is empty after conversion")
|
||
|
||
upload(port, track_num, pcm)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|