nRF
This commit is contained in:
zyphlar
+123
-158
@@ -1,16 +1,20 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
flash_upload.py — Upload audio files to the Baby Mobile's W25Q128 flash.
|
||||
flash_upload.py — Upload audio to Baby Mobile v2 via serial.
|
||||
|
||||
This is a fallback for uploading audio if USB mass storage isn't working.
|
||||
Normally you'd just drag files onto the USB drive.
|
||||
|
||||
Usage:
|
||||
python3 flash_upload.py /dev/ttyUSB0 song1.raw song2.raw song3.raw ...
|
||||
python3 flash_upload.py /dev/ttyACM0 song1.mp3 song2.wav song3.raw
|
||||
|
||||
Audio files must be raw unsigned 8-bit PCM at 8000 Hz mono.
|
||||
Convert with ffmpeg:
|
||||
ffmpeg -i song.mp3 -ar 8000 -ac 1 -f u8 -acodec pcm_u8 song.raw
|
||||
Audio files will be auto-converted to 8kHz/8-bit/unsigned PCM using ffmpeg.
|
||||
|
||||
The script writes a track table to sector 0, then audio data starting
|
||||
at sector 1 (address 0x1000).
|
||||
Track table format (v2):
|
||||
[0] magic = 0xBB
|
||||
[1] num_tracks
|
||||
[2..3] reserved
|
||||
[4..] 12-byte entries: start(4) + length(4) + rate_khz(1) + bits(1) + pad(2)
|
||||
"""
|
||||
|
||||
import serial
|
||||
@@ -21,211 +25,172 @@ import os
|
||||
import subprocess
|
||||
import shutil
|
||||
|
||||
BAUD = 9600
|
||||
BAUD = 115200
|
||||
PAGE_SIZE = 256
|
||||
SECTOR_SIZE = 4096
|
||||
AUDIO_START = 0x1000
|
||||
MAX_TRACKS = 32
|
||||
FLASH_SIZE = 16 * 1024 * 1024 # 16MB
|
||||
FLASH_SIZE = 16 * 1024 * 1024
|
||||
TABLE_MAGIC = 0xBB
|
||||
ENTRY_SIZE = 12
|
||||
|
||||
|
||||
def open_serial(port):
|
||||
ser = serial.Serial(port, BAUD, timeout=2)
|
||||
time.sleep(2) # wait for Arduino reset
|
||||
# Flush
|
||||
time.sleep(2)
|
||||
ser.read(ser.in_waiting)
|
||||
return ser
|
||||
|
||||
|
||||
def flash_identify(ser):
|
||||
ser.write(b'I')
|
||||
line = ser.readline().decode().strip()
|
||||
print(f" {line}")
|
||||
return line
|
||||
|
||||
|
||||
def flash_erase(ser):
|
||||
print(" Erasing entire flash (takes ~40 seconds)...")
|
||||
ser.write(b'E')
|
||||
line = ser.readline().decode().strip()
|
||||
while "Done" not in line:
|
||||
line = ser.readline().decode().strip()
|
||||
if line:
|
||||
print(f" {line}")
|
||||
print(" Erase complete.")
|
||||
|
||||
|
||||
def flash_write_page(ser, addr, data):
|
||||
"""Write up to 256 bytes to flash."""
|
||||
assert len(data) <= 256
|
||||
cmd = b'W'
|
||||
cmd += struct.pack('>I', addr)[1:] # 3 bytes address
|
||||
cmd += struct.pack('>H', len(data))
|
||||
ser.write(cmd)
|
||||
ser.write(data)
|
||||
resp = ser.readline().decode().strip()
|
||||
return resp
|
||||
|
||||
|
||||
def flash_write_data(ser, start_addr, data, label=""):
|
||||
"""Write arbitrary length data, page by page."""
|
||||
total = len(data)
|
||||
written = 0
|
||||
addr = start_addr
|
||||
|
||||
while written < total:
|
||||
chunk = data[written:written + PAGE_SIZE]
|
||||
# Pad last page
|
||||
if len(chunk) < PAGE_SIZE:
|
||||
chunk = chunk + b'\x80' * (PAGE_SIZE - len(chunk)) # 0x80 = silence
|
||||
|
||||
flash_write_page(ser, addr, chunk)
|
||||
written += PAGE_SIZE
|
||||
addr += PAGE_SIZE
|
||||
|
||||
pct = min(100, written * 100 // total)
|
||||
print(f"\r Writing {label}: {pct}% ({written}/{total} bytes)", end="", flush=True)
|
||||
|
||||
print()
|
||||
return addr
|
||||
|
||||
|
||||
def convert_to_raw(input_path):
|
||||
"""Convert any audio file to 8kHz 8-bit unsigned PCM using ffmpeg."""
|
||||
raw_path = input_path + ".raw"
|
||||
|
||||
"""Convert any audio file to 8kHz 8-bit unsigned PCM."""
|
||||
if input_path.endswith('.raw'):
|
||||
return input_path
|
||||
|
||||
|
||||
raw_path = input_path + ".raw"
|
||||
if not shutil.which('ffmpeg'):
|
||||
print(f"ERROR: ffmpeg not found. Please convert manually:")
|
||||
print(f" ffmpeg -i {input_path} -ar 8000 -ac 1 -f u8 -acodec pcm_u8 {raw_path}")
|
||||
print(f" ERROR: ffmpeg not found. Convert manually:")
|
||||
print(f" ffmpeg -i {input_path} -ar 8000 -ac 1 -f u8 -acodec pcm_u8 {raw_path}")
|
||||
sys.exit(1)
|
||||
|
||||
print(f" Converting {os.path.basename(input_path)} to 8kHz/8bit PCM...")
|
||||
subprocess.run([
|
||||
|
||||
print(f" Converting {os.path.basename(input_path)}...")
|
||||
result = subprocess.run([
|
||||
'ffmpeg', '-y', '-i', input_path,
|
||||
'-ar', '8000', '-ac', '1', '-f', 'u8', '-acodec', 'pcm_u8',
|
||||
raw_path
|
||||
], capture_output=True)
|
||||
|
||||
|
||||
if result.returncode != 0:
|
||||
print(f" ERROR: ffmpeg failed: {result.stderr.decode()[-200:]}")
|
||||
sys.exit(1)
|
||||
|
||||
return raw_path
|
||||
|
||||
|
||||
def build_track_table(tracks):
|
||||
"""Build the track index (stored in sector 0 of flash).
|
||||
|
||||
Format:
|
||||
byte 0: num_tracks (uint8)
|
||||
bytes 1-3: padding
|
||||
bytes 4+: track_start[MAX_TRACKS] as uint32 big-endian
|
||||
then: track_length[MAX_TRACKS] as uint32 big-endian
|
||||
"""
|
||||
num = len(tracks)
|
||||
table = struct.pack('B', num) + b'\x00' * 3
|
||||
|
||||
# Start addresses
|
||||
"""Build v2 track table."""
|
||||
table = bytearray()
|
||||
table.append(TABLE_MAGIC)
|
||||
table.append(len(tracks))
|
||||
table.extend(b'\x00\x00') # reserved
|
||||
|
||||
for start, length in tracks:
|
||||
table += struct.pack('>I', start)
|
||||
for _ in range(MAX_TRACKS - num):
|
||||
table += struct.pack('>I', 0)
|
||||
|
||||
# Lengths
|
||||
for start, length in tracks:
|
||||
table += struct.pack('>I', length)
|
||||
for _ in range(MAX_TRACKS - num):
|
||||
table += struct.pack('>I', 0)
|
||||
|
||||
# Pad to sector size
|
||||
table += b'\xFF' * (SECTOR_SIZE - len(table))
|
||||
return table
|
||||
table.extend(struct.pack('>I', start))
|
||||
table.extend(struct.pack('>I', length))
|
||||
table.append(8) # sample rate kHz
|
||||
table.append(8) # bits per sample
|
||||
table.extend(b'\x00\x00')
|
||||
|
||||
# Pad to page boundary
|
||||
while len(table) % PAGE_SIZE != 0:
|
||||
table.append(0xFF)
|
||||
|
||||
return bytes(table)
|
||||
|
||||
|
||||
def serial_cmd(ser, cmd):
|
||||
"""Send a single-char command and read response line."""
|
||||
ser.write(cmd.encode())
|
||||
return ser.readline().decode().strip()
|
||||
|
||||
|
||||
def serial_write_page(ser, addr, data):
|
||||
"""Write up to 256 bytes via the serial 'W' command."""
|
||||
assert len(data) <= 256
|
||||
ser.write(b'W')
|
||||
ser.write(struct.pack('>I', addr)[1:]) # 3-byte addr
|
||||
ser.write(struct.pack('>H', len(data))) # 2-byte length
|
||||
ser.write(data)
|
||||
return ser.readline().decode().strip()
|
||||
|
||||
|
||||
def serial_write_data(ser, start_addr, data, label=""):
|
||||
"""Write arbitrary-length data, page by page."""
|
||||
total = len(data)
|
||||
written = 0
|
||||
addr = start_addr
|
||||
|
||||
while written < total:
|
||||
chunk = data[written:written + PAGE_SIZE]
|
||||
if len(chunk) < PAGE_SIZE:
|
||||
chunk = chunk + b'\x80' * (PAGE_SIZE - len(chunk))
|
||||
|
||||
serial_write_page(ser, addr, chunk)
|
||||
written += PAGE_SIZE
|
||||
addr += PAGE_SIZE
|
||||
|
||||
pct = min(100, written * 100 // total)
|
||||
print(f"\r Writing {label}: {pct}%", end="", flush=True)
|
||||
|
||||
print()
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 3:
|
||||
print("Usage: python3 flash_upload.py <serial_port> <audio_file> [audio_file ...]")
|
||||
print()
|
||||
print("Supported input formats: .raw (8kHz/8bit/unsigned), .mp3, .wav, .ogg, .flac")
|
||||
print("Non-.raw files will be converted automatically using ffmpeg.")
|
||||
print()
|
||||
print("Example:")
|
||||
print(" python3 flash_upload.py /dev/ttyUSB0 lullaby.mp3 twinkle.wav brahms.mp3")
|
||||
print("Usage: python3 flash_upload.py <port> <audio_file> [...]")
|
||||
print(" Supported: .raw .mp3 .wav .ogg .flac .aac")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
port = sys.argv[1]
|
||||
audio_files = sys.argv[2:]
|
||||
|
||||
if len(audio_files) > MAX_TRACKS:
|
||||
print(f"ERROR: Maximum {MAX_TRACKS} tracks supported.")
|
||||
sys.exit(1)
|
||||
|
||||
# Convert files to raw PCM
|
||||
print("=== Preparing audio files ===")
|
||||
|
||||
# Convert
|
||||
print("=== Preparing audio ===")
|
||||
raw_files = []
|
||||
for f in audio_files:
|
||||
raw = convert_to_raw(f)
|
||||
size = os.path.getsize(raw)
|
||||
duration = size / 8000
|
||||
print(f" {os.path.basename(f)}: {size} bytes ({duration:.1f} seconds)")
|
||||
print(f" {os.path.basename(f)}: {size/1024:.0f} KB ({size/8000:.1f}s)")
|
||||
raw_files.append(raw)
|
||||
|
||||
# Check total size
|
||||
total_audio = sum(os.path.getsize(f) for f in raw_files)
|
||||
available = FLASH_SIZE - AUDIO_START
|
||||
if total_audio > available:
|
||||
print(f"ERROR: Audio ({total_audio} bytes) exceeds flash capacity ({available} bytes)")
|
||||
sys.exit(1)
|
||||
|
||||
total_duration = total_audio / 8000
|
||||
print(f"\n Total: {total_audio} bytes ({total_duration:.1f} seconds / {total_duration/60:.1f} minutes)")
|
||||
print(f" Flash usage: {(total_audio + AUDIO_START) * 100 // FLASH_SIZE}%")
|
||||
|
||||
|
||||
total = sum(os.path.getsize(f) for f in raw_files)
|
||||
print(f" Total: {total/1024:.0f} KB ({total/8000/60:.1f} min)")
|
||||
|
||||
# Connect
|
||||
print(f"\n=== Connecting to {port} ===")
|
||||
ser = open_serial(port)
|
||||
flash_identify(ser)
|
||||
|
||||
print(f" {serial_cmd(ser, 'I')}")
|
||||
|
||||
# Erase
|
||||
print("\n=== Erasing flash ===")
|
||||
flash_erase(ser)
|
||||
|
||||
# Write audio data and build track list
|
||||
ser.write(b'E')
|
||||
while True:
|
||||
line = ser.readline().decode().strip()
|
||||
if line: print(f" {line}")
|
||||
if "Done" in line: break
|
||||
|
||||
# Write tracks
|
||||
print("\n=== Writing audio ===")
|
||||
tracks = []
|
||||
addr = AUDIO_START
|
||||
|
||||
for i, raw_path in enumerate(raw_files):
|
||||
with open(raw_path, 'rb') as f:
|
||||
|
||||
for i, raw in enumerate(raw_files):
|
||||
with open(raw, 'rb') as f:
|
||||
data = f.read()
|
||||
|
||||
start = addr
|
||||
length = len(data)
|
||||
tracks.append((start, length))
|
||||
|
||||
name = os.path.basename(audio_files[i])
|
||||
addr = flash_write_data(ser, start, data, label=f"Track {i+1} ({name})")
|
||||
|
||||
# Align to sector boundary for clean layout
|
||||
if addr % SECTOR_SIZE != 0:
|
||||
|
||||
tracks.append((addr, len(data)))
|
||||
serial_write_data(ser, addr, data, label=f"Track {i+1}")
|
||||
|
||||
addr += len(data)
|
||||
# Align to sector
|
||||
if addr % SECTOR_SIZE:
|
||||
addr = ((addr // SECTOR_SIZE) + 1) * SECTOR_SIZE
|
||||
|
||||
# Write track table to sector 0
|
||||
|
||||
# Write track table
|
||||
print("\n=== Writing track table ===")
|
||||
table = build_track_table(tracks)
|
||||
flash_write_data(ser, 0, table, label="Track table")
|
||||
|
||||
serial_write_data(ser, 0, table, label="Track table")
|
||||
|
||||
# Summary
|
||||
print("\n=== Upload complete! ===")
|
||||
print(f" {len(tracks)} tracks written")
|
||||
for i, (start, length) in enumerate(tracks):
|
||||
print(f" Track {i+1}: addr=0x{start:06X}, {length} bytes ({length/8000:.1f}s)")
|
||||
print(f"\n Reset the board to start playback.")
|
||||
|
||||
# Cleanup temp files
|
||||
print(f"\n=== Done! {len(tracks)} tracks uploaded ===")
|
||||
for i, (s, l) in enumerate(tracks):
|
||||
print(f" Track {i+1}: 0x{s:06X}, {l/8000:.1f}s")
|
||||
|
||||
# Cleanup
|
||||
for raw, orig in zip(raw_files, audio_files):
|
||||
if raw != orig and os.path.exists(raw):
|
||||
os.remove(raw)
|
||||
|
||||
|
||||
ser.close()
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user