54 lines
2.1 KiB
Python
54 lines
2.1 KiB
Python
"""메모리 위에서만 도는 오디오 처리. 파일도 서브프로세스도 없다."""
|
|
import io
|
|
|
|
import av
|
|
import numpy as np
|
|
|
|
SAMPLE_RATE = 44100 # 렌더 믹스 기준
|
|
MP3_BIT_RATE = 192000
|
|
INT16_FULL_SCALE = 32768
|
|
|
|
|
|
def decode_mp3(data: bytes, sample_rate: int = SAMPLE_RATE) -> np.ndarray:
|
|
"""mp3 → 모노 s16 PCM. 소스 샘플레이트가 달라도 리샘플해 맞춘다."""
|
|
container = av.open(io.BytesIO(data))
|
|
resampler = av.audio.resampler.AudioResampler(
|
|
format="s16", layout="mono", rate=sample_rate)
|
|
chunks = []
|
|
for frame in container.decode(audio=0):
|
|
for resampled in resampler.resample(frame):
|
|
chunks.append(resampled.to_ndarray().reshape(-1))
|
|
for resampled in resampler.resample(None): # 남은 것 밀어내기
|
|
chunks.append(resampled.to_ndarray().reshape(-1))
|
|
container.close()
|
|
return np.concatenate(chunks) if chunks else np.zeros(0, np.int16)
|
|
|
|
|
|
def encode_mp3(pcm: np.ndarray, sample_rate: int = SAMPLE_RATE) -> bytes:
|
|
buffer = io.BytesIO()
|
|
container = av.open(buffer, "w", format="mp3")
|
|
stream = container.add_stream("mp3", rate=sample_rate)
|
|
stream.layout = "mono"
|
|
stream.bit_rate = MP3_BIT_RATE
|
|
frame = av.AudioFrame.from_ndarray(pcm.reshape(1, -1), format="s16", layout="mono")
|
|
frame.rate = sample_rate
|
|
for packet in stream.encode(frame):
|
|
container.mux(packet)
|
|
for packet in stream.encode():
|
|
container.mux(packet)
|
|
container.close()
|
|
return buffer.getvalue()
|
|
|
|
|
|
def trim_silence(pcm: np.ndarray, threshold_db: float = -45,
|
|
keep_lead: float = 0.05, keep_trail: float = 0.12,
|
|
sample_rate: int = SAMPLE_RATE) -> np.ndarray:
|
|
"""앞뒤 무음만 깎는다. 말소리 사이의 쉼은 보존한다."""
|
|
threshold = INT16_FULL_SCALE * 10 ** (threshold_db / 20)
|
|
loud = np.flatnonzero(np.abs(pcm) > threshold)
|
|
if len(loud) == 0:
|
|
return pcm[:0]
|
|
start = max(0, loud[0] - round(keep_lead * sample_rate))
|
|
end = min(len(pcm), loud[-1] + 1 + round(keep_trail * sample_rate))
|
|
return pcm[start:end]
|