들어가며
지난 글에서 우리는 이미 녹음되어 컴퓨터에 저장된 오디오 파일(WAV, MP3)을 불러와 그 위에 이펙터를 입히거나 분석하는 오프라인(Offline) 처리를 다루었습니다. 이러한 방식은 전체 신호를 한 번에 조망할 수 있어 연산이 쉽지만, 마이크로 말하는 순간 곧바로 메아리가 들리는 노래방 기기나 라이브 방송용 노이즈 캔슬링 소프트웨어에는 사용할 수 없습니다.
우리가 만든 디지털 신호 처리(DSP) 알고리즘이 현실 세계에서 진정한 가치를 발휘하려면, 소리가 입력되는 그 순간 즉각적으로 연산을 수행하고 다시 스피커로 내보내는 실시간(Real-time) 환경으로 넘어가야 합니다.
하지만 실시간 처리는 단순히 코드를 무한 루프에 가두는 것으로 해결되지 않습니다. 컴퓨터가 연산하는 속도와 사운드카드가 소리를 뿜어내는 속도의 아슬아슬한 줄타기를 맞춰야 하기 때문입니다. 이 균형이 조금이라도 무너지면 소리가 무참히 끊기거나, 치명적인 지연(Latency)이 발생하여 라이브의 생명력을 잃게 됩니다.
여기에는 오프라인 처리에는 없던 두 가지 제약이 새로 등장합니다. 첫째, 우리는 신호의 아주 작은 조각, 즉 버퍼(buffer) 하나만을 매 순간 손에 쥘 수 있을 뿐이며, 미래의 샘플은 아직 존재하지 않습니다. 둘째, 다음 버퍼가 도착하기 전에 현재 버퍼의 처리를 반드시 끝내야 합니다. 마감 시간을 놓치면 소리가 끊기거나 잡음이 섞입니다.
이 두 제약은 곧 두 가지 핵심 질문으로 이어집니다. 버퍼 단위로 신호를 쪼개면서도 이펙트의 연속성을 어떻게 유지할 것인가, 그리고 끊김 없는 재생을 위한 지연(latency)을 어떻게 관리할 것인가입니다. 오늘 우리는 이 질문에 답하면서, PyAudio의 콜백(callback) 구조 위에 실시간 분석기를 세우고, 지난 시간에 만든 이펙트를 라이브 스트림 위에 얹어 보겠습니다. 우리가 만든 효과가 마침내 '살아 움직이는' 모습을 함께 확인할 차례입니다.
1. 블로킹 방식과 콜백 방식
PyAudio로 오디오 스트림을 다루는 방법은 크게 두 가지입니다.
첫째는 블로킹(blocking) 방식입니다. stream.read()로 입력 버퍼를 한 덩어리 읽고, 처리한 뒤, stream.write()로 출력하는 과정을 우리가 직접 짠 반복문 안에서 순서대로 수행합니다. 흐름이 직관적이라는 장점이 있지만, 오디오 입출력을 우리 코드가 직접 운전해야 하므로, 처리 도중 다른 작업(예: 화면 갱신)을 하느라 다음 read() 호출이 늦어지면 곧바로 끊김으로 이어집니다.
둘째는 콜백(Callback) 방식입니다. 우리는 "버퍼가 준비될 때마다 이 함수를 호출해 달라"고 함수 하나를 등록해 둘 뿐입니다. 그러면 오디오 드라이버(PortAudio)가 별도의 고우선순위 스레드에서, 새 입력 버퍼가 들어올 때마다 또는 새 출력 버퍼가 필요할 때마다 그 함수를 자동으로 불러 줍니다. 오디오 처리가 우리의 메인 스레드와 독립적으로 돌아가므로 지연이 낮고, 메인 스레드는 그동안 사용자 인터페이스나 시각화 같은 다른 일에 전념할 수 있습니다.
실시간 처리에서는 콜백 방식이 표준입니다. 다만 한 가지 계약(Contract)을 지켜야 합니다. 콜백 함수는 빠르게 반환되어야 하고, 그 안에서 시간이 오래 걸리는 작업이나 블로킹을 일으키는 작업을 해서는 안 됩니다. 이 함수는 사실상 마감 시간이 있는 소프트 실시간(Soft Real-time) 코드이기 때문입니다.
2. 콜백 함수의 구조: 가장 단순한 통과(passthrough)
Python에서 실시간 오디오 스트림을 다루기 위해 가장 널리 쓰이는 라이브러리는 PyAudio입니다. PyAudio는 C 언어로 작성된 강력한 PortAudio 라이브러리를 파이썬에서 쉽게 쓸 수 있게 매핑해 놓은 도구입니다.
# PyAudio 설치
pip install pyaudio먼저 입력을 아무 처리 없이 그대로 출력으로 내보내는 가장 단순한 콜백을 보겠습니다. 골격을 익히는 것이 목적입니다.
import pyaudio
import numpy as np
import time
# --- 오디오 스트림 파라미터 ---
RATE = 44100 # 샘플링 주파수 (Hz)
CHUNK = 1024 # 버퍼(블록) 크기 (frames_per_buffer)
CHANNELS = 1 # 모노
FORMAT = pyaudio.paFloat32 # 32비트 부동소수점, 값 범위 [-1, 1]
def callback(in_data, frame_count, time_info, status):
# 1) 바이트 스트림 -> NumPy 배열
x = np.frombuffer(in_data, dtype=np.float32)
# 2) 신호 처리 (지금은 입력을 그대로 통과시킵니다)
y = x
# 3) NumPy 배열 -> 바이트 스트림으로 반환
return (y.tobytes(), pyaudio.paContinue)
p = pyaudio.PyAudio()
stream = p.open(
format=FORMAT, channels=CHANNELS, rate=RATE,
input=True, output=True, # 마이크 입력과 스피커 출력을 동시에
frames_per_buffer=CHUNK,
stream_callback=callback, # 콜백 등록
)
print("스트림을 시작합니다. Ctrl+C 로 종료합니다.")
stream.start_stream()
try:
while stream.is_active():
time.sleep(0.1)
except KeyboardInterrupt:
pass
finally:
stream.stop_stream()
stream.close()
p.terminate()
print("스트림을 종료합니다.")(참고) 한 가지 안전 수칙이 있습니다. 마이크 입력과 스피커 출력을 동시에 켜면, 스피커 소리가 마이크로 되돌아 들어가 하울링(howling)을 일으킬 수 있습니다. 실시간 입출력 실습은 반드시 헤드폰을 착용한 상태에서 진행하시기 바랍니다.
3. 실시간 오디오 처리의 핵심: 버퍼(Buffer)와 지연(Latency)
3.1 샘플 단위 처리 vs 블록 단위 처리
마이크는 아날로그 소리를 초당 44,100번(44.1kHz 기준) 쪼개어 디지털 숫자(샘플)로 바꿉니다. 이상적으로는 이 샘플이 하나 들어올 때마다 DSP 연산을 거쳐 바로 스피커로 내보내면 지연 시간이 0에 가까워질 것입니다. 하지만 파이썬과 같은 고수준 언어나 운영체제는 샘플 하나마다 연산을 지시하고 결과를 기다리기엔 너무 무겁고 느립니다.
그래서 도입된 개념이 버퍼(Buffer, 또는 Block/Frame)입니다. 샘플이 하나씩 들어올 때마다 처리하지 않고, 일정한 바구니(버퍼)에 담아두었다가 바구니가 다 차면 한 번에 뭉텅이로 처리하는 방식입니다.
3.2 버퍼 크기와 지연(Latency)의 딜레마
버퍼 크기(Chunk Size)의 설정은 실시간 오디오 프로그래밍에서 가장 중요한 튜닝 포인트이며, 본질적으로 트레이드오프(Trade-off) 관계에 놓여 있습니다. 한쪽을 얻으면 다른 쪽을 내어 주어야 합니다.
버퍼가 작으면(예: 64, 128 샘플) 바구니가 금방 차기 때문에 소리가 마이크에서 스피커로 나가기까지의 지연이 매우 짧아집니다. 라이브 공연이나 악기 모니터링처럼 즉각적인 반응이 생명인 상황에 적합합니다. 그러나 그만큼 컴퓨터는 아주 짧은 시간 안에 쉴 새 없이 연산을 끝내야 하므로 부담이 커지고, 연산이 조금이라도 밀리면 출력이 비어 소리가 뚝뚝 끊기는 언더런(Under-run)이 발생합니다.
반대로 버퍼가 크면(예: 2048, 4096 샘플) 컴퓨터가 한 번에 많은 데이터를 여유 있게 처리할 수 있어 끊길 위험이 거의 없습니다. 대신 바구니가 다 차기를 오래 기다려야 하므로, 마이크에 대고 말한 소리가 한참 뒤에야 스피커로 흘러나옵니다.
결국 이것은 반응성(낮은 지연)과 안정성(끊김 없음) 사이의 영원한 줄다리기입니다. 그렇다면 버퍼 크기는 지연을 정확히 얼마나 만들어 내며, 우리는 어떤 값에서 출발해야 할까요. 이 물음은 잠시 뒤 3.4절에서 수식과 함께 정량적으로 풀어 보겠습니다.
3.3 버퍼 경계와 상태 유지: 실시간 처리의 핵심
이제 이번 글에서 가장 중요한 개념을 다룰 차례입니다. 오프라인 처리와 실시간 처리를 가르는 결정적인 차이는 바로 상태(State)입니다.
문제의 본질은 이렇습니다. 대부분의 이펙트는 한 버퍼보다 긴 기억(Memory)을 필요로 합니다. 예를 들어 샘플링 주파수 44,100 Hz에서 200 ms 길이의 에코를 만들려면 지연선(Delay Line)의 길이가 $D = 0.2 \times 44100 = 8820$ 샘플이어야 합니다. 그런데 버퍼 하나는 고작 1,024 샘플입니다. 즉, 어떤 출력 샘플을 계산하려면 8~9개 버퍼쯤 전에 들어온 샘플이 필요한데, 그 샘플은 이미 콜백 함수를 떠나 사라진 뒤입니다.
만약 우리가 매 버퍼를 서로 독립적으로 처리한다면, 다시 말해 버퍼 경계마다 이펙트의 기억을 0으로 초기화한다면 어떻게 될까요. 에코가 각 버퍼의 시작점에서 매번 다시 시작되면서, 버퍼와 버퍼가 만나는 이음매마다 신호가 불연속적으로 튀게 됩니다. 이 튐은 귀에 '톡톡' 하는 클릭음으로 들립니다.
아래 그림은 동일한 신호를 두 방식으로 처리한 결과입니다. 오른쪽 확대 그림에서, 상태를 유지한 올바른 처리(짙은 남색)는 버퍼 경계(n=3000)를 매끄럽게 통과하는 반면, 버퍼마다 상태를 초기화한 잘못된 처리(분홍색)는 경계에서 신호가 급격히 튀어 오릅니다. 검증해 보면 상태를 초기화한 경우 경계에서의 샘플 간 점프가 내부 구간의 일반적인 변화량보다 9배에서 30배까지 커졌습니다. 이 차이가 바로 클릭음의 정체입니다.
해법은 명확합니다. 콜백 호출과 호출 사이에 이펙트의 내부 상태를 지속적으로 보존하면 됩니다. 이를 구현하는 방법으로는 직접 순환 버퍼(Ring Buffer)를 만들어 과거 샘플을 들고 다니는 방식도 있지만, 우리는 더 깔끔한 길을 택하겠습니다. 18화에서 다룬 IIR 필터의 도구를 그대로 가져오는 것입니다.
되먹임 에코는 사실 하나의 IIR 필터입니다.
이 식의 전달함수는 다음과 같습니다.
따라서 분자 계수는 $b = [1]$, 분모 계수는 $a = [1, 0, \dots, 0, -g]$ (길이 $D + 1$, 마지막 자리에 $-g$)가 됩니다. 그런데 scipy.signal.lfilter에는 필터의 내부 지연선 상태를 주고받는 zi 인자가 있습니다. 이 zi를 매 버퍼마다 다음 버퍼로 넘겨주면, 지연선이 콜백 호출을 가로질러 자동으로 유지됩니다.
아래 코드는 이 원리가 실제로 옳은지 검증하는 짧은 실험입니다. 전체를 한 번에 처리한 결과(오프라인)와, 버퍼 단위로 쪼개되 상태를 넘기며 처리한 결과를 비교합니다.
import numpy as np
from scipy.signal import lfilter
RATE = 44100
delay_sec, g = 0.20, 0.6
D = int(delay_sec * RATE) # 지연 샘플 수
b = np.array([1.0])
a = np.zeros(D + 1); a[0] = 1.0; a[D] = -g
x = np.random.randn(RATE).astype(np.float64) # 1초 길이 테스트 신호
# (A) 전체를 한 번에 처리 (오프라인 기준)
ref, _ = lfilter(b, a, x, zi=np.zeros(D))
# (B) 버퍼 단위로 쪼개 처리하되, 상태(zi)를 다음 버퍼로 넘김
CHUNK = 1024
zi = np.zeros(D) # 초기 상태(무음)
out = np.empty_like(x)
for s in range(0, len(x), CHUNK):
e = min(s + CHUNK, len(x))
out[s:e], zi = lfilter(b, a, x[s:e], zi=zi)
print("최대 오차:", np.max(np.abs(out - ref))) # -> 0.0실제로 실행하면 최대 오차는 정확히 0.0입니다. 버퍼 크기를 128부터 4096까지 바꿔 가며 확인해도 결과는 동일했습니다. 버퍼 단위로 쪼개 처리한 결과가 전체를 한 번에 처리한 결과와 부동소수점 정밀도 한계까지 일치한다는 사실은, 우리의 상태 유지 방식이 수학적으로 정확함을 보여 줍니다. 여기서 상태 벡터 zi의 길이는 정확히 $D$이며, 이것이 곧 지연선 그 자체입니다.
3.4 지연(Latency)의 정량화와 관리
이제 3.2절에서 미뤄 둔 물음, 즉 버퍼 크기가 지연을 얼마나 만들어 내는가에 수치로 답할 차례입니다. 입력 지연은 단순한 식으로 결정됩니다.
버퍼 하나가 채워지기를 기다려야 처리가 시작되므로, 버퍼가 클수록 기다리는 시간도 비례해서 길어집니다. 샘플링 주파수 44,100 Hz에서 버퍼 크기별 입력 지연을 계산하면 다음과 같습니다.
| 버퍼 크기 (표본) | 한 방향 지연 (ms) | 왕복 지연(개략, ms) |
|---|---|---|
| 64 | 1.45 | 2.90 |
| 128 | 2.90 | 5.80 |
| 256 | 5.80 | 11.61 |
| 512 | 11.61 | 23.22 |
| 1024 | 23.22 | 46.44 |
| 2048 | 46.44 | 92.88 |
| 4096 | 92.88 | 185.76 |
표에서 보듯 같은 샘플링 주파수에서도 지연은 64 샘플의 약 1.5ms부터 4096 샘플의 약 93ms까지 60배 넘게 벌어집니다. 3.2절에서 살펴본 반응성과 안정성의 줄다리기가 바로 이 수치 위에서 벌어지는 셈입니다. 입력과 출력을 모두 거치는 왕복 지연은 대략 한 방향 지연의 두 배로 잡으면 되며, 실제 값은 운영체제의 오디오 드라이버와 버퍼링 정책에 따라 더 늘어날 수 있습니다.
실무적인 출발점은 이렇습니다. 처음에는 512 또는 1,024 샘플에서 시작합니다. 1,024 샘플이면 약 23ms로, 일반적인 작업에는 큰 무리가 없습니다. 더 빠른 반응이 필요하고 이펙트 연산이 가볍다면 256이나 128로 줄여 보되, 끊김이 생기기 시작하는 지점을 한계로 삼습니다. 반대로 끊김이 잦다면 버퍼를 키우는 것이 가장 확실한 해결책입니다. 콜백의 status 인자를 검사하면 이런 언더런 상황을 실시간으로 감지할 수 있습니다.
4. 실시간 이펙터를 스트림에 얹기
이제 3.3절에서 검증한 상태 유지 에코를 콜백 안으로 옮겨, 라이브 마이크 입력에 실시간으로 적용해 보겠습니다. 재사용을 위해 이펙트를 작은 클래스로 감쌉니다.
import pyaudio
import numpy as np
import time
from scipy.signal import lfilter
RATE, CHUNK, CHANNELS, FORMAT = 44100, 1024, 1, pyaudio.paFloat32
class RealtimeEcho:
"""블록 간 지연선 상태(zi)를 유지하는 실시간 되먹임 에코."""
def __init__(self, delay_sec, feedback, rate):
D = int(delay_sec * rate)
self.b = np.array([1.0])
self.a = np.zeros(D + 1); self.a[0] = 1.0; self.a[D] = -feedback
self.zi = np.zeros(D) # 콜백 호출 사이에 유지되는 상태
def process(self, block):
y, self.zi = lfilter(self.b, self.a, block, zi=self.zi)
return y
echo = RealtimeEcho(delay_sec=0.25, feedback=0.5, rate=RATE)
def callback(in_data, frame_count, time_info, status):
if status: # 언더런/오버플로 경고 감지
print("status:", status)
x = np.frombuffer(in_data, dtype=np.float32).astype(np.float64)
y = echo.process(x) # 상태를 갱신하며 블록 처리
y = np.clip(y, -1.0, 1.0) # 출력 클리핑 방지
return (y.astype(np.float32).tobytes(), pyaudio.paContinue)
p = pyaudio.PyAudio()
stream = p.open(format=FORMAT, channels=CHANNELS, rate=RATE,
input=True, output=True, frames_per_buffer=CHUNK,
stream_callback=callback)
print("실시간 에코를 시작합니다 (헤드폰 권장). Ctrl+C 로 종료합니다.")
stream.start_stream()
try:
while stream.is_active():
time.sleep(0.1)
except KeyboardInterrupt:
pass
finally:
stream.stop_stream(); stream.close(); p.terminate()콜백 안에서 지켜야 할 성능 원칙이 몇 가지 있습니다. 연산은 NumPy와 lfilter처럼 C로 최적화된 벡터 연산으로 처리하고, 파이썬의 샘플 단위 반복문은 피해야 합니다. 콜백 안에서 큰 메모리를 새로 할당하거나, 파일을 열거나, print로 출력하는 행위도 가급적 삼가야 합니다. 모두 콜백의 마감 시간을 위협하는 요소이기 때문입니다. 위 코드에서 status 검사 시의 print는 이상 상황에서만 드물게 실행되므로 예외로 두었습니다.
스테레오를 다룰 때는 한 가지만 주의하면 됩니다. CHANNELS=2인 경우 샘플이 좌·우 채널 순으로 번갈아 저장되므로, frombuffer로 푼 뒤 reshape(-1, 2)로 채널을 분리해 처리하고 다시 합쳐야 합니다.
5. 콜백 기반 분석기 만들기: 측정과 표시의 분리
마지막으로, 이번 글의 제목이 가리키는 분석기를 완성하겠습니다. 실시간 분석기란 흘러가는 오디오에서 매 버퍼마다 특징을 측정하여 보여 주는 도구입니다. 가장 기본이 되는 측정값은 신호의 세기, 즉 RMS 레벨입니다.
이 정의가 옳게 동작하는지는 간단히 확인됩니다. 진폭이 1인 풀스케일 정현파의 RMS 레벨은 이론값 -3.01 dBFS이며, 진폭이 0.5인 정현파는 -9.03 dBFS입니다. 실제로 계산해 보면 각각 -3.00, -9.02 dBFS로, 이론값과 정확히 일치합니다.
여기서 dBFS(Decibels Relative to Full Scale)는 디지털 시스템이 데이터 손실 없이 표현할 수 있는 최대 신호 크기, 즉 풀스케일을 0의 기준으로 삼는 진폭 단위입니다. 따라서 0 dBFS가 디지털이 담아낼 수 있는 최댓값이며, 이 선을 넘어 레벨을 올리면 클리핑(Clipping)이 발생해 치명적인 소리 왜곡을 유발합니다.
아래 그림은 진폭이 시간에 따라 변하는 신호를 1,024 샘플 버퍼 단위로 처리하며 RMS 레벨을 추적한 결과입니다. 매 버퍼마다 계산된 레벨이 신호의 음량 변화를 충실히 따라가는 것을 볼 수 있습니다.
여기서 새로운 설계 문제가 등장합니다. 콜백은 오디오 스레드에서 실행되는데, 측정 결과를 화면에 표시하는 일은 메인 스레드에서 이루어집니다. 콜백 안에서 직접 화면을 갱신하면 마감 시간을 위협하고 스레드 충돌의 위험도 생깁니다. 해법은 측정과 표시를 분리하는 것입니다. 콜백(생산자)은 가벼운 측정값만 계산해 스레드 안전한 큐(Queue)에 넣고, 메인 스레드(소비자)가 큐에서 꺼내 표시합니다. 이것이 생산자-소비자(Producer-Consumer) 패턴입니다.
import pyaudio
import numpy as np
import queue
RATE, CHUNK, CHANNELS, FORMAT = 44100, 1024, 1, pyaudio.paFloat32
level_q = queue.Queue() # 오디오 스레드 -> 메인 스레드 전달 통로
def callback(in_data, frame_count, time_info, status):
x = np.frombuffer(in_data, dtype=np.float32)
rms = np.sqrt(np.mean(x.astype(np.float64) ** 2))
db = 20 * np.log10(rms + 1e-12) # 무음에서의 log(0) 방지
level_q.put(db) # 가벼운 결과만 큐에 전달
return (None, pyaudio.paContinue) # 분석만 하므로 출력은 없음
p = pyaudio.PyAudio()
stream = p.open(format=FORMAT, channels=CHANNELS, rate=RATE,
input=True, output=False, # 입력 전용 스트림
frames_per_buffer=CHUNK, stream_callback=callback)
stream.start_stream()
# 메인 스레드: 큐에서 꺼내 콘솔 레벨 미터로 표시
print("실시간 레벨 미터를 시작합니다. Ctrl+C 로 종료합니다.")
try:
while stream.is_active():
db = level_q.get()
bars = int(np.clip((db + 60) / 60 * 40, 0, 40)) # -60~0 dBFS -> 0~40칸
print(f"\r[{'#' * bars:<40}] {db:6.1f} dBFS", end="")
except KeyboardInterrupt:
pass
finally:
stream.stop_stream(); stream.close(); p.terminate()이 프로그램은 마이크로 들어오는 소리의 세기를 콘솔 막대그래프로 실시간 표시합니다. 콜백은 RMS 계산이라는 가벼운 연산만 수행하고 결과를 큐에 넣을 뿐이며, 무거운 표시 작업은 전적으로 메인 스레드가 담당합니다. 입력 전용 스트림이므로 output=False로 두었고, 출력이 없으니 콜백은 (None, paContinue)를 반환합니다.
여기서 한 걸음 더 나아갈 수 있습니다. 콜백 안에서 버퍼에 FFT를 적용해 주파수 스펙트럼을 큐로 넘기면, 메인 스레드에서 Matplotlib 애니메이션으로 실시간 스펙트럼 분석기를 그릴 수 있습니다. 또한 4절의 RealtimeEcho를 이 구조에 결합하면, 에코를 적용해 출력하면서 동시에 입력 레벨을 감시하는 통합 모니터가 됩니다. 측정과 효과, 그리고 표시를 각각의 책임으로 깔끔하게 나누는 것이 견고한 실시간 시스템의 골격입니다.
[Insight]
오늘 우리가 넘어온 강의 폭은 생각보다 넓습니다. 표면적으로는 PyAudio라는 라이브러리의 사용법을 익힌 것처럼 보이지만, 본질적으로는 사고의 틀 자체가 바뀌었습니다. 신호 전체를 손에 쥐고 내려다보던 일괄(Batch) 처리에서, 끝없이 흘러가는 조각만을 마주하며 상태를 미래로 넘겨 가는 스트리밍(Streaming) 처리로 옮겨 온 것입니다.
오늘 다룬 버퍼링과 콜백은 비단 파이썬과 오디오에만 국한된 이야기가 아닙니다. 우리가 시청하는 넷플릭스 스트리밍, 온라인 게임의 네트워크 패킷 전송, 실시간 주식 거래 시스템 등 연속적인 데이터를 끊김 없이 처리해야 하는 모든 소프트웨어 아키텍처의 근간을 관통하는 철학입니다.
지연을 둘러싼 상충 관계 또한 단순한 파라미터 튜닝 이상의 의미를 담고 있습니다. 작은 버퍼는 빠르지만 위태롭고, 큰 버퍼는 느긋하지만 둔합니다. 최소의 지연과 최대의 견고함을 동시에 가질 수는 없으며, 버퍼링이란 결국 시간을 내어 주고 안정성을 사 오는 거래입니다. 네트워크의 지터 버퍼, 영상 스트리밍의 사전 버퍼링 모두 같은 원리의 다른 얼굴입니다.
오랜 시간 신호 처리를 해 온 입장에서 돌아보면, 이런 실시간 처리는 한때 전용 DSP 칩과 어셈블리 수준의 저수준 코드, 그리고 인터럽트와 직접 씨름해야 가능한 일이었습니다. 마감 시간을 놓치지 않으려고 사이클 단위로 코드를 다듬던 시절이 있었습니다. 그 모든 본질, 즉 마감을 지키고 상태를 보존하며 지연을 다스리는 일은 지금도 그대로지만, 우리는 이제 그것을 몇 줄의 파이썬으로 표현합니다. 도구가 추상화되었다고 해서 원리가 사라진 것은 아닙니다. 오히려 도구가 단순해질수록, 그 아래에 깔린 원리를 정확히 아는 사람과 그렇지 못한 사람의 격차는 더 또렷하게 드러납니다. 콜백 안에 무엇을 넣어도 되고 무엇을 넣으면 안 되는지를 아는 것, 그 판단의 근거를 이해하는 것이 결국 실력입니다.
한 가지 구체적인 사례를 들어 보겠습니다. 우리는 에코를 scipy의 lfilter로 우아하게 구현했습니다. 상태가 자동으로 보존되고 코드도 짧다는 분명한 장점이 있습니다. 그러나 lfilter는 분모 계수 a가 거의 0이고 단 하나의 탭(a[D])만 살아 있다는 희소성을 활용하지 못한 채, 샘플마다 지연선 전체를 훑는 O(N·D) 연산을 수행합니다. 실제로 측정해 보면 0.25초 지연(D=11,025)에서 1,024 샘플 한 블록을 처리하는 데 약 9.7ms가 걸려, 콜백에 허락된 23ms 예산의 40%가 넘는 시간을 한 번에 소진합니다. 같은 일을 순환 버퍼(Ring Buffer)로 구현하면 바로 그 하나의 과거 샘플만 더하면 되므로 연산량이 O(N)으로 줄어, 약 0.01ms 만에 끝나 수백 배 빨라집니다. 학습 단계에서는 lfilter의 명료함이 옳은 선택이지만, 지연이 길어지거나 여러 이펙트를 동시에 얹어야 하는 실전에서는 이 비용 구조를 이해하고 순환 버퍼로 갈아타야 합니다. 어떤 도구가 어떤 상황에서 비싼지를 아는 것, 그것이 바로 앞서 말한 '판단의 근거'입니다.
파이썬의 PyAudio는 실시간 처리의 개념을 익히기엔 매우 훌륭한 도구입니다. 하지만 파이썬 자체의 한계(GIL - Global Interpreter Lock)와 운영체제 스케줄러의 한계 때문에, 프로 오디오 인터페이스처럼 완벽한 1~2ms 수준의 초저지연(Ultra-low Latency) 시스템을 상용 수준으로 파이썬만으로 구축하는 것은 무리가 있습니다. 극단의 최적화가 필요한 상용 VST 플러그인이나 DAW(디지털 오디오 워크스테이션)는 여전히 C/C++로 작성되며 메모리 할당조차 실시간 스레드 안에서는 엄격하게 금지됩니다.
그러나 이 콜백 아키텍처 위에서 필터의 상태(State)를 보존하며 데이터를 흘려보내는 감각을 익혔다면, 여러분은 이미 단순한 분석가를 넘어 시스템 엔지니어의 시각을 갖추게 된 것입니다.
다음 글 예고
이제 오디오 신호를 분석하고, 창작하고, 실시간으로 통제하는 방법까지 모두 익혔습니다. 청각의 세계를 정복했으니 이제 시각의 세계로 우리의 신호 처리 도구를 확장해 볼 차례입니다.
다음 시간에는 [24. 이미지 신호 처리: 2D FFT와 공간 영역 필터링의 이해] 편으로 새로운 Part를 엽니다. 한 줄로 늘어선 샘플 대신, 가로와 세로로 펼쳐진 화소(Pixel)의 격자를 신호로 바라보는 순간, 우리가 익힌 개념들이 그대로 2차원으로 확장됩니다. FFT는 2D FFT가 되어 이미지의 공간 주파수를 드러내고, 컨볼루션은 공간 영역 필터링이 되어 흐림과 윤곽 강조를 만들어 냅니다. 소리의 세계에서 갈고닦은 직관이 이미지의 세계에서 어떻게 되살아나는지, 함께 확인해 보겠습니다.
📌 이전 연재 글 보기
- [20] 칼만 필터(Kalman): 센서 데이터 예측 및 필터링 기초
- [21] 오디오 특징 추출: Librosa를 활용한 MFCC, Chroma 추출 (AI 전처리)
- [22] 오디오 이펙터 제작: 리버브, 딜레이, 피치 시프팅 기법
