You've already forked MicroPythonOS
mirror of
https://github.com/m5stack/MicroPythonOS.git
synced 2026-05-20 11:51:27 -07:00
AudioPlayer: upscale if low frequency
This commit is contained in:
@@ -6,16 +6,19 @@ import micropython
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# AudioPlayer – robust, volume-controllable WAV player (MONO + STEREO)
|
||||
# Auto-up-samples any rate < 22050 Hz to 22050 Hz for MAX98357
|
||||
# ----------------------------------------------------------------------
|
||||
class AudioPlayer:
|
||||
# class-level defaults
|
||||
_i2s = None
|
||||
_volume = 50 # 0-100
|
||||
_volume = 50 # 0-100
|
||||
_keep_running = True
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# WAV header parser
|
||||
# ------------------------------------------------------------------
|
||||
@staticmethod
|
||||
def find_data_chunk(f):
|
||||
"""Skip chunks until 'data' is found → (data_start, data_size, sample_rate, channels)"""
|
||||
"""Return (data_start, data_size, sample_rate, channels)"""
|
||||
f.seek(0)
|
||||
if f.read(4) != b'RIFF':
|
||||
raise ValueError("Not a RIFF file")
|
||||
@@ -68,6 +71,34 @@ class AudioPlayer:
|
||||
print("stop_playing()")
|
||||
cls._keep_running = False
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Helper: up-sample a raw PCM buffer (zero-order-hold)
|
||||
# ------------------------------------------------------------------
|
||||
@staticmethod
|
||||
def _upsample_buffer(raw: bytearray, factor: int) -> bytearray:
|
||||
"""
|
||||
Duplicate each 16-bit sample `factor` times.
|
||||
Input: interleaved L,R,L,R... (or mono)
|
||||
Output: same layout, each sample repeated `factor` times.
|
||||
"""
|
||||
if factor == 1:
|
||||
return raw
|
||||
|
||||
upsampled = bytearray(len(raw) * factor)
|
||||
out_idx = 0
|
||||
# each sample = 2 bytes
|
||||
for i in range(0, len(raw), 2):
|
||||
lo = raw[i]
|
||||
hi = raw[i + 1]
|
||||
for _ in range(factor):
|
||||
upsampled[out_idx] = lo
|
||||
upsampled[out_idx + 1] = hi
|
||||
out_idx += 2
|
||||
return upsampled
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Main playback routine
|
||||
# ------------------------------------------------------------------
|
||||
@classmethod
|
||||
def play_wav(cls, filename):
|
||||
cls._keep_running = True
|
||||
@@ -77,15 +108,28 @@ class AudioPlayer:
|
||||
file_size = st[6]
|
||||
print(f"File size: {file_size} bytes")
|
||||
|
||||
data_start, data_size, sample_rate, channels = cls.find_data_chunk(f)
|
||||
print(f"data chunk: {data_size} bytes @ {sample_rate} Hz, {channels}-channel")
|
||||
# ----- parse header ------------------------------------------------
|
||||
data_start, data_size, original_rate, channels = cls.find_data_chunk(f)
|
||||
|
||||
# ----- decide playback rate (force >= 22050 Hz) --------------------
|
||||
target_rate = 22050
|
||||
if original_rate >= target_rate:
|
||||
playback_rate = original_rate
|
||||
upsample_factor = 1
|
||||
else:
|
||||
# find smallest integer factor so original * factor >= target
|
||||
upsample_factor = (target_rate + original_rate - 1) // original_rate
|
||||
playback_rate = original_rate * upsample_factor
|
||||
|
||||
print(f"Original: {original_rate} Hz → Playback: {playback_rate} Hz "
|
||||
f"(factor {upsample_factor}), {channels}-ch")
|
||||
|
||||
if data_size > file_size - data_start:
|
||||
data_size = file_size - data_start
|
||||
|
||||
# ---- I2S init ------------------------------------------------
|
||||
i2s_format = machine.I2S.MONO if channels == 1 else machine.I2S.STEREO
|
||||
# ----- I2S init ----------------------------------------------------
|
||||
try:
|
||||
i2s_format = machine.I2S.MONO if channels == 1 else machine.I2S.STEREO
|
||||
cls._i2s = machine.I2S(
|
||||
0,
|
||||
sck=machine.Pin(2, machine.Pin.OUT),
|
||||
@@ -94,18 +138,18 @@ class AudioPlayer:
|
||||
mode=machine.I2S.TX,
|
||||
bits=16,
|
||||
format=i2s_format,
|
||||
rate=sample_rate,
|
||||
rate=playback_rate,
|
||||
ibuf=32000
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"Warning: simulating playback (I2S init failed): {e}")
|
||||
|
||||
print(f"Playing {data_size} bytes (vol {cls._volume}%) …")
|
||||
print(f"Playing {data_size} original bytes (vol {cls._volume}%) …")
|
||||
f.seek(data_start)
|
||||
|
||||
# ----- Viper volume scaler (works on any buffer) -------------------
|
||||
@micropython.viper
|
||||
def scale_audio(buf: ptr8, num_bytes: int, scale_fixed: int):
|
||||
# Process 16-bit samples (2 bytes each)
|
||||
for i in range(0, num_bytes, 2):
|
||||
lo = int(buf[i])
|
||||
hi = int(buf[i+1])
|
||||
@@ -121,17 +165,17 @@ class AudioPlayer:
|
||||
buf[i+1] = (sample >> 8) & 255
|
||||
|
||||
chunk_size = 4096
|
||||
bytes_per_sample = 2 * channels # 2 bytes per channel
|
||||
total = 0
|
||||
bytes_per_original_sample = 2 * channels # 2 bytes per channel
|
||||
total_original = 0
|
||||
|
||||
while total < data_size:
|
||||
while total_original < data_size:
|
||||
if not cls._keep_running:
|
||||
print("Playback stopped by user.")
|
||||
break
|
||||
|
||||
to_read = min(chunk_size, data_size - total)
|
||||
# Ensure we read full samples
|
||||
to_read -= (to_read % bytes_per_sample)
|
||||
# read a chunk of *original* data
|
||||
to_read = min(chunk_size, data_size - total_original)
|
||||
to_read -= (to_read % bytes_per_original_sample)
|
||||
if to_read <= 0:
|
||||
break
|
||||
|
||||
@@ -139,21 +183,25 @@ class AudioPlayer:
|
||||
if not raw:
|
||||
break
|
||||
|
||||
# Apply volume scaling (in-place, per sample)
|
||||
# ----- up-sample if needed ---------------------------------
|
||||
if upsample_factor > 1:
|
||||
raw = cls._upsample_buffer(raw, upsample_factor)
|
||||
|
||||
# ----- volume scaling ---------------------------------------
|
||||
scale = cls._volume / 100.0
|
||||
if scale < 1.0:
|
||||
scale_fixed = int(scale * 32768)
|
||||
scale_audio(raw, len(raw), scale_fixed)
|
||||
|
||||
# Write to I2S (stereo interleaves L,R,L,R...)
|
||||
# ----- output ------------------------------------------------
|
||||
if cls._i2s:
|
||||
cls._i2s.write(raw)
|
||||
else:
|
||||
# Simulate timing
|
||||
num_samples = len(raw) // bytes_per_sample
|
||||
time.sleep(num_samples / sample_rate)
|
||||
# simulate timing with the *playback* rate
|
||||
num_samples = len(raw) // (2 * channels)
|
||||
time.sleep(num_samples / playback_rate)
|
||||
|
||||
total += len(raw)
|
||||
total_original += to_read # count original bytes only
|
||||
|
||||
print("Playback finished.")
|
||||
except Exception as e:
|
||||
|
||||
Reference in New Issue
Block a user