[Voice] pyVoIP - Python으로 작성된 VoIP 라이브러리

 

0. pyVoIP

pyVoIP는 순수 Python으로 작성된 VoIP(Voice over IP)/SIP(Session Initiation Protocol)/RTP(Real-time Transport Protocol) 라이브러리입니다. 음성 서비스를 개발하려고 하는 파이썬 개발자가 파이썬을 활용해 간단한 서비스를 만드는데 유용합니다.

---

1. 특징 및 기능

주요 특징

 

- 순수 Python 구현 : 외부 종속성 없이 순수 Python으로 구현되어 있어 설치와 사용이 간편합니다[2][4].

- 지원 코덱 : PCMA, PCMU

- 독립적인 오디오 처리 : 특정 오디오 라이브러리에 의존하지 않아, pyaudio나 wave 등 선형 사운드 데이터를 처리할 수 있는 다양한 라이브러리와 함께 사용할 수 있습니다.

 

주요 기능

 

- VoIP 전화 구현 : VoIPPhone 클래스를 사용하여 간단하게 가상 전화를 구현할 수 있습니다.

- 자동 응답 및 통화 처리 : 콜백 함수를 통해 수신 전화에 대한 자동 응답 및 통화 처리가 가능합니다.

- SIP 및 RTP 프로토콜 지원 : SIP를 통한 세션 설정 및 RTP를 통한 실시간 오디오 전송을 지원합니다.

주의사항

 

- PCMU/PCMA는 8000Hz, 1채널, 8비트 오디오만 지원합니다.

- 실제 프로덕션 환경에서는 Twilio 같은 서비스를 사용하는 것이 권장됩니다. pyVoIP는 오디오 품질이 최적화되어 있지 않을 수 있습니다.

pyVoIP는 간단한 VoIP 시스템을 구현하거나 학습 목적으로 사용하기에 적합한 라이브러리입니다. 하지만 고품질의 오디오나 복잡한 기능이 필요한 실제 서비스에는 제한이 있을 수 있다는 단점이 있습니다.

---

2.  예시

아래의 예시 코드를 하나씩 확인해보도록 하겠습니다.

 

전체 코드

from pyVoIP.VoIP import VoIPPhone, CallState
import uuid
from openai import OpenAI
import os
import pywav

def convert_to_wav(audio, tmpFileName):
    data_bytes = b"".join(audio)
    wave_write = pywav.WavWrite(tmpFileName, 1, 8000, 8, 7)
    wave_write.write(data_bytes)
    wave_write.close()

    return open(tmpFileName, "rb")

def transcribe_to_text(audio_file):
    tmpFileName = f"/tmp/audio/_audio_buffer_{uuid.uuid4()}.wav"
    client = OpenAI()

    transcription = client.audio.transcriptions.create(
        model="whisper-1",
        file=convert_to_wav(audio_file)
    )

    try:
        return transcription.text
    except Exception as ex:
       print(ex)
    return ""

def answer(call):
    try:
        call.answer()
        buffer = []
        buff_length = 0

        while call.state == CallState.ANSWERED:
            audio = call.read_audio()
            buff_length += len(audio) / 8

            if buff_length <= 1000:
                buffer.append(audio)
            else:
                print(transcribe_to_text(buffer))
                buffer = []
                buff_length = 0

    except Exception as e:
        print(e)
    finally:
       call.hangup()

vp = VoIPPhone('xxx', 5060, 'xxx', 'xxx', callCallback=answer)
vp.start()
print(vp._status)
input("Press any key to exist")
vp.stop()

 

 

Convert_to_wav 함수

def convert_to_wav(audio, tmpFileName):
    data_bytes = b"".join(audio)
    wave_write = pywav.WavWrite(tmpFileName, 1, 8000, 8, 7)
    wave_write.write(data_bytes)
    wave_write.close()

    return open(tmpFileName, "rb")

 

pywav를 사용해 바이트 단위의 음성 신호를 wav파일로 저장하는 코드입니다. 

여기서 pywav.WavWrite의 파라미터는 다음과 같습니다.

 

- tmpFileName = 파일명

- 1 = 채널 

- 8000 = 샘플링레이트

- 8 = 비트 수

- 7 = 코덱

 

여기서 지원하는 오디오 코덱의 포맷은 다음과 같습니다.

# Audio format 1 = PCM (without compression)

# Audio format 6 = PCMA (with A-law compression)

# Audio format 7 = PCMU (with mu-law compression)

 

 

transcribe_to_text 함수

def transcribe_to_text(audio_file):
    tmpFileName = f"/tmp/audio/_audio_buffer_{uuid.uuid4()}.wav"
    client = OpenAI()

    transcription = client.audio.transcriptions.create(
        model="whisper-1",
        file=convert_to_wav(audio_file)
    )

    try:
        return transcription.text
    except Exception as ex:
       print(ex)
    return ""

 

 

이 함수는 OpenAI의 Whisper 모델을 사용하여 오디오 파일을 텍스트로 변환합니다.

파일을 WAV 형식으로 변환한 후, Whisper-1 모델을 호출하여 텍스트 전사를 생성하고, 그 결과를 반환합니다.

• client = OpenAI(): OpenAI API 클라이언트를 생성합니다. 이를 통해 OpenAI 모델을 호출하여 작업을 수행할 수 있습니다.
• transcription = client.audio.transcriptions.create(): Whisper 모델을 사용하여 오디오 파일의 텍스트를 생성합니다.
  • model="whisper-1": OpenAI에서 제공하는 Whisper-1 모델을 사용합니다. Whisper는 음성 인식 및 전사에 특화된 모델입니다.
  • file=convert_to_wav(audio_file): convert_to_wav라는 함수(코드에서는 정의되지 않았지만, 파일을 WAV 형식으로 변환하는 기능으로 추정됨)를 호출하여 입력된 오디오 파일을 WAV 형식으로 변환합니다. Whisper 모델은 일반적으로 WAV 파일 형식을 요구합니다.
• return transcription.text: Whisper 모델이 생성한 텍스트를 반환합니다.

call 함수

def answer(call):
    try:
        call.answer()
        buffer = []
        buff_length = 0

        while call.state == CallState.ANSWERED:
            audio = call.read_audio()
            buff_length += len(audio) / 8

            if buff_length <= 1000:
                buffer.append(audio)
            else:
                print(transcribe_to_text(buffer))
                buffer = []
                buff_length = 0

    except Exception as e:
        print(e)
    finally:
       call.hangup()

 

 

이 코드는 전화 통화를 받으면, 통화가 연결된 상태에서 오디오 데이터를 읽어와 버퍼에 일정 크기만큼 모은 후, 그 데이터를 텍스트로 변환하고 출력합니다.

오디오 데이터를 계속 읽어와 버퍼에 모으다가, 1000 바이트 단위로 텍스트 변환을 시도하며, 모든 처리가 끝나면 통화를 종료합니다.

 

call.answer() :

이 부분은 통화를 받는 동작을 의미합니다. call 객체의 answer() 메소드를 호출하여 통화가 연결되었음을 의미합니다.

 

buffer = [] & buff_length = 0 : 

• buffer: 오디오 데이터를 저장할 리스트입니다. 오디오 데이터를 일정 크기만큼 모아서 처리하기 위한 버퍼 역할을 합니다.
• buff_length: 현재까지 모인 오디오 데이터의 길이를 추적합니다. 오디오 데이터가 어느 정도 모이면 이를 텍스트로 변환하기 위해 버퍼를 사용합니다.

while call.state == CallState.ANSWERED:

통화 상태가 "ANSWERED"인 동안에만 오디오 데이터를 처리합니다. 즉, 통화가 연결된 상태에서만 오디오 데이터를 계속 수신하고 처리하는 루프입니다.

audio = call.read_audio()

통화에서 오디오 데이터를 읽어옵니다. call.read_audio() 메소드가 현재 통화의 오디오 데이터를 가져오는 역할을 합니다.

buff_length += len(audio) / 8

• 오디오 데이터의 길이를 8로 나누어 buff_length에 더합니다. 이는 오디오 데이터를 바이트 단위로 처리하고 있다는 것을 나타냅니다.
• 오디오 데이터의 길이를 누적해서 추적하고, 일정 크기에 도달했을 때 텍스트로 변환을 시도합니다.

if buff_length <= 1000:

• 누적된 오디오 데이터의 길이가 1000보다 작거나 같을 경우, 오디오 데이터를 buffer 리스트에 추가합니다.
• 이 조건은 일정 크기 이하의 오디오 데이터를 계속해서 버퍼에 저장하는 역할을 합니다.

else:

• 만약 buff_length가 1000을 초과하면, 버퍼에 모인 오디오 데이터를 transcribe_to_text(buffer) 함수로 넘겨 텍스트로 변환합니다.
• transcribe_to_text(buffer): 이전에 설명한 오디오 전사 함수가 실행되며, 버퍼에 저장된 오디오 데이터를 텍스트로 변환합니다.
• 그 결과를 print()를 통해 출력합니다.

finally call.hangup() :

call.hangup()을 호출하여 통화를 종료합니다.

 

이러한 과정들을 통해 파이썬에서 VOIP 시스템을 사용해 음성 서비스를 간단하게 구현할 수 있습니다.

---

3.  Reserach

해당 라이브러리로 AICC 서비스를 구현하기 위해 read_audio를 통해 넘어오는 음성의 Format, 음성 변환 방법 리서치 등 다양한 연구를 진행했습니다.

 

가장 먼저, pyVOIP의 read_audio 함수를 자세하게 살펴봤습니다.

def read_audio(self, length=160, blocking=True) -> bytes:
    if len(self.RTPClients) == 1:
        return self.RTPClients[0].read(length, blocking)
    data = []
    for x in self.RTPClients:
        data.append(x.read(length))
    # Mix audio from different sources before returning
    nd = audioop.add(data.pop(0), data.pop(0), 1)
    for d in data:
        nd = audioop.add(nd, d, 1)
    return nd

 

 

Reads linear/raw audio data from the received buffer. Returns length amount of bytes. Default length is 160 as that is the amount of bytes sent per PCMU/PCMA packet. When blocking is set to true, this function will not return until data is available. When blocking is set to false and data is not available, this function will return b"\x80" * length

 

→ SIP에서 넘어오는 오디오 신호(바이트를) 버퍼(Default 160)로 받아서 반환함. 이 때, audio data는 linear/raw data로 압축되지 않은 오디오 신호임을 의미함.
→ 8000K는 1초에 8000샘플(8000바이트)을 생성하는데, 각 샘플은 1바이트(8비트)로 인코딩됨.

→ read_audio를 통해 160바이트 단위로 넘어오는 패킷을 1초 단위의 wav 파일로 변환하기 위해 필요한 갯수는 50개가 필요함. 8000샘플 / 160바이트 = 50개

→ 단, 위의 코드에서 buff_length += len(audio) / 8로 해준 이유는, 밀리초 단위로 간단하게 표현하기 위함. 1초를 1000ms로 나누면 8000샘플 / 1000ms = 8샘플/ms이 됨.

→ 그러면 160샘플 / 8 = 20ms이 되므로, buffer_length가 1000ms가 됐을 때 wav파일로 변환또는 텍스트 전사를 진행함.

 

Gather buffer

- 0.32 sec을 쌓으려면? -? 160 sample * 16 =-> 2560 sample 

- 3.2 sec을 쌓으려면? --> 25600 samples

 

Audio Format

 

160 바이트 단위로 넘어오는 것은 오디오 바이트의 포맷은 8K 8BIT PCM 형태입니다.

 

이 부분에 대해서는 정보가 부족하여 정확히 확인은 불가능 했습니다. 원래 아날로그 신호의 포멧은 8K 8BIT U-law 형태인데 pyVOIP의 함수를 거치면서 PCM 형태로 변환되는 것이라 생각하고 있습니다.

 

이렇게 생각한 이유는 read_audio를 통해 들어온 byte를 그대로 waveWrite를 통해 wav파일로 만들고 재생했을 때, 정상적으로 재생이 된다면 PCM 형태일 것이기 때문입니다. (실제로 정상적으로 재생됨.)

---

4. 마치며

pyVOIP를 사용하여 주로 자바로 구현하는 VOIP 시스템을 간단하게 구현할 수 있었습니다.

 

이를 통해 간단하게 AICC 서비스를 구현해 STT Model, Answer Model, TTS Model의 인퍼런스 속도 및 성능을 확인할 수 있었습니다.

 

실제 음성 서비스를 구현하기전에 파이썬을 주로 이용하는 AI 개발자가 사용하기에 적절한 라이브러리라고 생각합니다.