전화 음성에서 웹 자막까지 5 — Python Gateway의 UDP 수신과 PCM 출력

전화 음성이 Python에 도착하면 곧바로 사람이 들을 수 있는 파형이 되는 것은 아닙니다. UDP 데이터그램에서 RTP 헤더를 읽고, G.711 payload를 PCM으로 바꾸고, 늦거나 빠진 패킷을 정리한 뒤 일정한 20ms 프레임으로 내보내야 합니다.

코드·JSON 조회에는 별도 원본 프로젝트가 필요합니다. 블로그 저장소에는 pbx_gateway/ 소스와 증거 파일이 포함돼 있지 않습니다. 원본이 없다면 파일 조회를 건너뛰고 본문 예시와 계산을 따라가세요. 실습 준비 조건을 먼저 확인하세요.

이번 글의 목표는 코드를 직접 고치는 것이 아니라 다음 한 줄을 실제 코드에서 추적하는 것입니다.

UDP 데이터그램 → RTP 파싱 → G.711 디코딩 → 지터 버퍼 → 20ms PCM → 출력 큐

현재 PBX 훈련에서는 실제 패킷 관찰을 아직 끝내지 않았습니다. 아래의 숫자와 코드 흐름은 저장소의 구현을 읽어 확인한 내용이며, 실통화에서 관찰한 결과처럼 표현하지 않습니다.

시리즈: 전체 지도 · 이전: RTP 패킷 읽기 · 5/8 Python Gateway · 다음: PCM 전달과 Agora 녹음

1. 먼저 전체 경로를 잡는다

이 프로젝트에서 SIP 처리는 Asterisk가 맡습니다. Python Gateway는 Asterisk가 넘긴 RTP와 ARI 이벤트를 처리합니다.

Linphone
  │ SIP: 통화 설정
  ▼
Asterisk
  │ RTP: G.711 음성
  ▼
Python Gateway의 UDP 수신기
  │ RTP payload 추출·디코딩·정렬
  ▼
PCMFrame: S16LE mono, 20ms
  ├─ LiveOutput → Meter / UDP / Agora 중 하나
  └─ WaveRecorder → 선택적 로컬 WAV

여기서 가장 중요한 경계는 RTP와 PCM입니다.

G.711을 디코딩한다고 마이크 원본을 비트 단위로 복원하는 것은 아닙니다. G.711은 손실이 있는 companding 코덱이고, 디코더는 G.711 값에 대응하는 선형 PCM 값을 만들어 냅니다.

2. 개념 하나: bind와 datagram_received는 수신 입구다

UDP는 연결을 먼저 맺지 않고 데이터그램 단위로 보냅니다. Python asyncio는 UDP 데이터가 도착할 때 datagram_received(data, addr) callback을 호출합니다.

현재 구현의 Receiver는 첫 송신 주소를 기억하고, 같은 통화 중 다른 주소에서 온 데이터그램은 버립니다. 받아들인 바이트는 MediaPipeline.feed()로 넘깁니다.

# 실제 코드의 핵심만 발췌
def datagram_received(self, data, addr):
    if r.peer is None:
        r.peer = addr
    elif addr != r.peer:
        r.source_drops += 1
        return

    r.pipeline.feed(data, time.monotonic())

time.monotonic()은 시스템 시각을 표시하기 위한 값이 아닙니다. 시간이 뒤로 가지 않는 기준으로 도착 간격과 재생 deadline을 계산하기 위해 사용합니다.

코드 읽기 실습 1

실행 위치: Ubuntu VM의 pbx_gateway 저장소. 프로세스를 실행하지 않고 파일만 읽습니다.

cd /workspace/pbx_gateway
sed -n '75,110p' gateway/__main__.py

기대 결과:

• datagram_received가 data와 addr를 받는다.
• 허용한 peer가 아니면 source_drops를 늘린다.
• 정상 데이터는 pipeline.feed()로 전달한다.

해석: UDP 포트를 열었다는 사실은 음성이 정상이라는 뜻이 아닙니다. callback 이후 RTP 파싱과 payload type 검사까지 통과해야 합니다.

3. 개념 하나: RTP 헤더와 payload를 분리한다

RTP의 고정 헤더는 최소 12바이트입니다. 현재 _parse_rtp()는 network byte order인 !BBHII로 다음 값을 읽습니다.

first, second, sequence, timestamp, ssrc = struct.unpack_from("!BBHII", datagram)

현재 설정에서 ulaw는 PT 0, alaw는 PT 8을 기대합니다. 설정과 실제 RTP의 PT가 다르면 packets_rejected가 증가합니다.

코드 읽기 실습 2

실행 위치: Ubuntu VM의 pbx_gateway 저장소.

cd /workspace/pbx_gateway
sed -n '118,142p' gateway/media.py
sed -n '215,245p' gateway/media.py

기대 결과:

• 첫 구간에서 RTP version, CSRC, extension, padding을 처리한다.
• 두 번째 구간에서 예상 PT와 빈 payload, 너무 큰 payload를 거절한다.

해석: packets_received는 UDP callback에 들어온 수이고, packets_accepted는 RTP 구조와 PT 검사를 통과한 수입니다. 두 값이 다르면 네트워크 단절보다 파싱 또는 설정 불일치를 먼저 의심할 수 있습니다.

4. 개념 하나: G.711 1바이트가 PCM 2바이트가 된다

현재 전화 음성의 RTP payload는 샘플 하나를 G.711 1바이트로 표현합니다. Gateway의 decode_g711()은 각 값을 signed 16-bit PCM 샘플로 바꾸고 little-endian 바이트로 묶습니다.

# 실제 코드
def decode_g711(payload: bytes, codec: str) -> bytes:
    decoder = _decode_ulaw if _codec_name(codec) == "ulaw" else _decode_alaw
    samples = (decoder(value) for value in payload)
    return struct.pack(f"<{len(payload)}h", *samples) if payload else b""

<는 little-endian, h는 signed 16-bit 정수입니다. 따라서 160바이트 G.711 payload는 160개의 PCM 샘플, 즉 320바이트가 됩니다.

8,000 samples/s × 0.020s = 160 samples
G.711: 160 samples × 1 byte = 160 bytes
PCM S16LE: 160 samples × 2 bytes = 320 bytes

오프라인 실습 3

실행 위치: Mac 터미널 또는 Ubuntu VM, Python 3가 있는 어느 쪽이든 가능합니다. 네트워크와 통화를 사용하지 않습니다.

python3 - <<'PY'
sample_rate = 8000
duration = 0.020
samples = int(sample_rate * duration)
print("samples:", samples)
print("G.711 bytes:", samples * 1)
print("PCM S16LE bytes:", samples * 2)
PY

기대 출력:

samples: 160
G.711 bytes: 160
PCM S16LE bytes: 320

해석: 바이트 수가 두 배가 된 이유는 음성 길이가 늘어서가 아니라 샘플 표현이 8비트에서 16비트로 바뀌었기 때문입니다.

5. 개념 하나: 지터 버퍼는 늦은 패킷을 잠깐 기다린다

UDP 패킷은 순서대로 도착한다는 보장이 없습니다. MediaPipeline은 sequence와 timestamp를 펼쳐서 저장하고, 기본 40ms를 기다린 뒤 timestamp 순서대로 PCM을 꺼냅니다.

# 실제 코드
deadline = (
    self._base_arrival
    + self.jitter_ms / 1000.0
    + (self._next_timestamp - self._base_timestamp) / RTP_CLOCK_RATE
)

재생할 시점까지 샘플이 없으면 현재 구현은 0을 넣습니다.

entry = self._samples.pop(timestamp, None)
if entry is None:
    values.append(0)
    missing += 1

이것은 정교한 Packet Loss Concealment가 아니라 무음 보충입니다. 끊김 대신 짧은 무음이 들어가지만, 빠진 음성을 복원하지는 못합니다.

버퍼 상한은 G.711 1,600샘플, 즉 200ms입니다. 무한히 기다리면 음성 지연이 계속 커지므로, 이 구현은 큰 timestamp 점프나 과도한 적체를 끊고 새 기준을 잡습니다.

코드 읽기 실습 4

실행 위치: Ubuntu VM의 pbx_gateway 저장소.

cd /workspace/pbx_gateway
sed -n '347,392p' gateway/media.py

확인할 질문:

1. force=False일 때 deadline 전에는 왜 break할까요?
2. force=True는 통화 종료 때 남은 버퍼를 비우는 데 왜 필요할까요?
3. missing > 0이면 어떤 통계가 증가할까요?

6. 개념 하나: 출력 단위는 항상 20ms다

RTP payload가 언제나 정확히 20ms라고 가정하면 위험합니다. 현재 구현은 입력 payload의 샘플을 timestamp 위치에 넣고, 출력할 때 FRAME_SAMPLES = 160 단위로 다시 묶습니다.

FRAME_SAMPLES = 160  # 8kHz 기준 20ms

8kHz 출력이면 160샘플·320바이트입니다. pcm_sample_rate=16000이면 선형 보간으로 320샘플·640바이트를 만듭니다. 샘플 수가 늘어도 원래 전화 음성에 없던 고주파 정보가 새로 생기는 것은 아닙니다.

완성된 프레임은 다음 구조입니다.

@dataclass(frozen=True, slots=True)
class PCMFrame:
    data: bytes
    sample_rate: int
    received_at: float | None = None
    synthesized: bool = False

received_at은 Gateway 내부 지연을 계산할 기준이고, synthesized는 무음 보충 여부입니다.

7. 개념 하나: 실시간 출력과 녹음은 같은 PCM에서 갈라진다

Runtime.emit()은 drain()이 만든 프레임을 LiveOutput과 선택적 WaveRecorder에 각각 전달합니다.

# 실제 코드
for frame in self.pipeline.drain(time.monotonic(), force=force):
    self.output.submit(frame)
    if self.recorder:
        self.recorder.submit(frame)

녹음 파일을 다시 읽어 실시간 전송하는 구조가 아닙니다. 메모리에 있는 같은 PCM 프레임이 두 갈래로 전달됩니다.

LiveOutput 큐의 기본 상한이 200ms라면 20ms 프레임 10개입니다. 큐가 가득 차면 가장 오래된 프레임을 버리고 최신 프레임을 넣습니다. 전화 음성에서는 오래된 데이터를 모두 보존해 지연을 키우는 것보다, 일부를 버리고 현재 시점에 가까이 머무는 정책이 유리할 수 있습니다.

다만 트레이드오프가 있습니다.

arrival_to_output_p95_ms는 RTP가 Gateway에 도착한 시점부터 sink 처리가 끝날 때까지입니다. 전화기→RTC→ASR→Signaling→웹 화면 전체 지연이 아닙니다.

8. 결과를 해석하는 순서

실제 실습을 진행한 뒤에는 한 숫자만 보지 말고 앞에서부터 확인합니다.

call_start 있음?
  ↓
rtp_peer 있음?
  ↓
packets_received 증가?
  ↓
packets_accepted 증가?
  ↓
frames_emitted 증가?
  ↓
output_frames 증가?
  ↓
sink_errors / live_drops는 0인가?

현재 학습 세션에서는 이 패킷 실측을 아직 완료하지 않았습니다. 표는 앞으로 수집할 로그를 읽는 기준입니다.

9. 흔한 오해와 오류

recvfrom()이 성공했으니 RTP도 정상이다

UDP 바이트가 왔다는 뜻일 뿐입니다. RTP version, header 길이, PT와 payload 검사를 통과해야 합니다.

PT 8이면 언제나 PCMA다

현재 구성과 RTP/AVP 정적 payload type 범위에서는 PT 8을 PCMA로 사용합니다. 동적 payload type은 SDP 협상 문맥을 함께 봐야 합니다.

sequence가 하나 빠지면 즉시 손실이다

패킷 순서가 바뀌어 늦게 올 수 있으므로 지터 버퍼의 재생 경계까지 기다린 뒤 판단합니다.

16kHz로 올리면 전화 음질이 두 배 좋아진다

현재 코드는 8kHz 샘플 사이를 보간합니다. 후속 SDK의 입력 형식에는 맞출 수 있지만 원본에 없던 음성 대역을 복원하지는 않습니다.

Gateway p95가 곧 자막 지연이다

이 지표는 Gateway 내부의 RTP 도착→PCM sink 경계만 잽니다. RTC 전송, ASR, Signaling, 브라우저 렌더링은 포함하지 않습니다.

10. 확인 문제

1. 8kHz에서 20ms는 몇 샘플인가요?
2. 같은 160샘플이 G.711에서는 160바이트, S16LE PCM에서는 320바이트인 이유는 무엇인가요?
3. packets_received와 packets_accepted의 차이는 무엇인가요?
4. packets_reordered와 packets_late는 어떤 재생 경계를 기준으로 달라지나요?
5. arrival_to_output_p95_ms에 ASR과 웹 자막 시간이 포함되지 않는 이유는 무엇인가요?

정답을 한 문장씩 말할 수 있으면 다음 글에서 PCM을 UDP와 Agora로 전달하고, 로컬 WAV와 원격 WAV가 각각 무엇을 증명하는지 구분할 준비가 된 것입니다.

구현 근거와 공식 참고 자료

이 글의 실제 구현 근거는 원본 저장소의 다음 파일입니다.

• pbx_gateway/gateway/__main__.py: UDP 수신, Runtime.emit(), 출력·녹음 분기
• pbx_gateway/gateway/media.py: RTP 파싱, G.711 변환, 지터 버퍼, 20ms PCMFrame
• pbx_gateway/gateway/outputs.py: 제한된 실시간 큐와 WAV recorder
• pbx_gateway/README.md: 현재 구현의 계산, 통계, 검증 범위

프로토콜과 Python API는 다음 공식 문서를 함께 참고했습니다.

• RFC 3550: RTP
• RFC 3551: RTP/AVP의 PCMU·PCMA payload type
• Python asyncio Datagram Protocol
• Python struct 모듈