by Troye
이번 튜토리얼에서는 저번의 gRPC 스트리밍 STT API사용법에 이어, 스트리밍 STT API를 활용하여 마이크 입력을 실시간으로 텍스트로 변환하는 프로그램을 만드는 과정을 설명합니다.
개발 환경인 Python에서 마이크 입력을 인터페이스로 사용하기 위해서 PyAudio 라이브러리를 이용합니다.
PyAudio
저번 글에서 설명한 샘플 코드는 로컬에 저장된 음원파일을 직접 분할하여 넘겨주는 방식입니다.
그렇다면 굳이 스트리밍 API를 이용하지 않고, 리턴제로의 일반 STT API를 이용하는 것이 더 효율적이고 간단한 방법입니다.
그렇기에 스트리밍 API의 이점을 활용하고자, 사용자의 마이크 입력을 오디오 인터페이스로 활용하는 튜토리얼을 진행해 보겠습니다.
설치
PyAudio는 파이썬에서 오디오를 기록, 재생할 수 있도록 기능을 제공하는 라이브러리입니다.
크로스 플랫폼 오디오 I/O 라이브러리인 PortAudio를 Python으로 바인딩한 것이기에, PyAudio 설치를 위해선 먼저 PortAudio의 설치가 필요합니다.
⋇Apple MacOS 기준(개발환경마다 설치법이 다를 수 있으니, 공식 홈페이지의 문서를 참조하세요.)
먼저 Homebrew를 이용해서 PortAudio를 설치합니다.
brew install portaudio
그 후 pip를 이용해서 PyAudio를 설치합니다.
pip install pyaudio
마이크 입력
마이크 입력을 위해서 먼저 PyAudio 객체를 생성하고, Stream을 열어줍니다.
import pyaudio
CHUNK = 1024
FORMAT = pyaudio.paInt16
CHANNELS = 1 # Only supports 1-Channel Input
RATE = 8000
ENCODING = pb.DecoderConfig.AudioEncoding.LINEAR16 # Only supports LINEAR16 Encoding using Voice Input
p = pyaudio.PyAudio()
stream = p.open(format=FORMAT,
channels=CHANNELS,
rate=RATE,
input=True) # 입력으로 사용하기 위함.
녹음에 필요한 파라미터로는 저장할 인코딩 정보인 format, 채널 수 channel, 샘플레이트 rate 가 있습니다. 다른 자세한 파라미터에 대한 정보는 생략하겠습니다.
그 후 stream.read(CHUNK) 함수를 실행하면, CHUNK만큼의 음성 정보를 입력받을 수 있습니다.
후술 할 스트리밍 API와 통합 과정에선 단순하게 Read 하는 것이 아닌, 연속적으로 정보를 받을 수 있도록 할 것입니다. 이는 스트림을 여는 open()함수의stream_callback 인자에 연속적으로 데이터를 받아서 버퍼에 저장하도록 Callback 함수를 이용해서 구현할 수 있습니다.
마이크 입력 인터페이스와 스트리밍 STT API의 통합
통합 전에 사전에 오디오 포맷에 대한 이해가 필요합니다.
오디오의 Header를 포함한 포맷은 최적의 SAMPLERATE 값을 넘겨주지 않아도, 서버에서 최적의 값을 결정합니다. 하지만 Header를 포함하지 않은 포맷은, RAW데이터를 서버로 넘겨주어야 합니다.
- Header를 포함한 포맷
WAV: RIFF형식의 파일 포맷으로 오디오 데이터에 대한 헤더와 오디오 데이터를 포함하고 있습니다. 오디오 데이터는 일반LINEAR가 아닌 다른 RAW인코딩이 될 수 있습니다.FLAC: 무손실 압축 인코딩으로, 헤더에 오디오에 대한 정보를 포함하고 있습니다.
- Header를 포함하지 않은 RAW포맷
LINEAR16: 16bit,SAMPLERATE값의 설정이 필요합니다.MULAW: 8bit-PCM EncodingALAW: 8bit-PCM EncodingAMR: 8000HzAMR_WB: 16000HzOPUS: RTP 패킷의 Payload에 부분만 전송해야합니다. 8000Hz, 12000Hz, 16000Hz, 24000Hz, 48000HzOGG_OPUS: OGG 컨테이너에서, Opus 데이터만 전송해야합니다.
PyAudio 의 입력 제한으로 인해 인코딩은 LINEAR16 을 사용해야합니다.위의 마이크 입력 오디오 인터페이스 Stream을 더 편하게 사용하기 위해서 아래와 같이 Class로 만들 수 있습니다.
class MicrophoneStream:
"""
Ref[1]: https://cloud.google.com/speech-to-text/docs/transcribe-streaming-audio
Recording Stream을 생성하고 오디오 청크를 생성하는 제너레이터를 반환하는 클래스.
"""
def __init__(self: object, rate: int = SAMPLE_RATE, chunk: int = CHUNK, channels: int = CHANNELS, format = FORMAT) -> None:
self._rate = rate
self._chunk = chunk
self._channels = channels
self._format = format
# Create a thread-safe buffer of audio data
self._buff = queue.Queue()
self.closed = True
self._audio_interface = pyaudio.PyAudio()
self._audio_stream = self._audio_interface.open(
format=pyaudio.paInt16,
channels=self._channels,
rate=self._rate,
input=True,
frames_per_buffer=self._chunk,
stream_callback=self._fill_buffer,
)
self.closed = False
def terminate(
self: object,
) -> None:
"""
Stream을 닫고, 제너레이터를 종료하는 함수
"""
self._audio_stream.stop_stream()
self._audio_stream.close()
self.closed = True
self._buff.put(None)
self._audio_interface.terminate()
def _fill_buffer(
self: object,
in_data: object,
frame_count: int,
time_info: object,
status_flags: object,
) -> object:
"""
오디오 Stream으로부터 데이터를 수집하고 버퍼에 저장하는 콜백 함수.
Args:
in_data: 바이트 오브젝트로 된 오디오 데이터
frame_count: 프레임 카운트
time_info: 시간 정보
status_flags: 상태 플래그
Returns:
바이트 오브젝트로 된 오디오 데이터
"""
self._buff.put(in_data)
return None, pyaudio.paContinue
def generator(self: object) -> object:
"""
Stream으로부터 오디오 청크를 생성하는 Generator.
Args:
self: The MicrophoneStream object
Returns:
오디오 청크를 생성하는 Generator
"""
while not self.closed:
chunk = self._buff.get()
if chunk is None:
return
data = [chunk]
while True:
try:
chunk = self._buff.get(block=False)
if chunk is None:
return
data.append(chunk)
except queue.Empty:
break
yield b"".join(data)
또 위의 Class를 사용하기 위해서 RTZROpenAPIClient Class를 아래와 같이 변경해 줍니다.
class RTZROpenAPIClient:
def __init__(self, client_id, client_secret):
super().__init__()
self._logger = logging.getLogger(__name__)
self.client_id = client_id
self.client_secret = client_secret
self._sess = Session()
self._token = None
self.stream = MicrophoneStream(SAMPLE_RATE, CHUNK, CHANNELS, FORMAT) # 마이크 입력을 오디오 인터페이스 사용하기 위한 Stream 객체 생성
@property
def token(self):
if self._token is None or self._token["expire_at"] < time.time():
resp = self._sess.post(
API_BASE + "/v1/authenticate",
data={"client_id": self.client_id, "client_secret": self.client_secret},
)
resp.raise_for_status()
self._token = resp.json()
return self._token["access_token"]
def transcribe_streaming_grpc(self, config):
base = GRPC_SERVER_URL
with grpc.secure_channel(
base, credentials=grpc.ssl_channel_credentials()
) as channel:
stub = pb_grpc.OnlineDecoderStub(channel)
cred = grpc.access_token_call_credentials(self.token)
audio_generator = self.stream.generator() # (1). 마이크 스트림 Generator
def req_iterator():
yield pb.DecoderRequest(streaming_config=config)
for chunk in audio_generator: # (2). yield from Stream Generator
yield pb.DecoderRequest(audio_content=chunk) # chunk를 넘겨서, 스트리밍 STT 수행
req_iter = req_iterator()
resp_iter = stub.Decode(req_iter, credentials=cred)
for resp in resp_iter:
resp: pb.DecoderResponse
for res in resp.results:
# 실시간 출력 형태를 위해서 캐리지 리턴 이용
if not res.is_final:
print("\033[K"+"Text: {}".format(res.alternatives[0].text), end="\r", flush=True) # \033[K: clear line Escape Sequence
else:
print("\033[K" + "Text: {}".format(res.alternatives[0].text), end="\n")
def __del__(self):
self.stream.terminate()
또 이들을 실행하기 위해서, 파일 입력은 필요가 없어졌으므로 main은 아래와 같이 변경될 수 있습니다.
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
)
env_config = configparser.ConfigParser()
env_config.read("config.ini")
# parser.add_argument("stream", help="File to stream to the API") # 마이크로 입력을 받기에 파일의 위치는 사용되지 않습니다.
args = parser.parse_args()
config = pb.DecoderConfig(
sample_rate=SAMPLE_RATE,
encoding=ENCODING,
use_itn=True,
use_disfluency_filter=False,
use_profanity_filter=False,
)
client = RTZROpenAPIClient(env_config["DEFAULT"]["CLIENT_ID"], env_config["DEFAULT"]["CLIENT_SECRET"])
try:
client.transcribe_streaming_grpc(config)
except KeyboardInterrupt:
print("Program terminated by user.")
del client
STT를 중단하고 싶다면 Ctrl + c 를 입력해서 종료하면 됩니다.
‼️ 주의사항
- 파일을 이용한 스트리밍 샘플코드를 사용할 때 오디오 인코딩에 따라
ENCODING,SAMPLE_RATE사용을 주의하세요! - 만약
WAV포맷을 가진 음원파일을 넘긴다면, 인코딩으로pb.DecoderConfig.AudioEncoding.WAV를 사용해 주세요.
서버 측에서 음원에 맞는 최적의SAMPLE_RATE를 결정해서 STT를 진행합니다. - 그럼에도
LINEAR16(비압축 인코딩 형태)을 인코딩으로 사용하고 싶다면, 꼭SAMPLE_RATE의 값으로 음원의 샘플레이트값을 사용해 주세요!
음원의 샘플레이트 값을 모른다면soundfile라이브러리를 이용하는 아래 추가 과정을 통해 코드를 수정해서 사용 가능합니다.
Extra
"soundfile" 사용 방법
먼저 soundfile 라이브러리를 설치합니다.soundfile 라이브러리는 추가적으로 numpy 모듈을 필요로 합니다.
pip install numpy
pip install soundfile그다음 샘플 소스에 soundfile 모듈을 추가합니다.
import grpc
import soundfile as sf # soundfile
import vito_stt_client_pb2 as pb
import vito_stt_client_pb2_grpc as pb_grpc
from requests import Session
그리고 프로그램 실행 부분을 아래와 같이 변경함으로써, LINEAR16 인코딩을 사용할 때 SAMPLE_RATE를 잘못 결정하는 실수를 방지할 수 있습니다.
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
)
env_config = configparser.ConfigParser()
env_config.read("config.ini")
parser.add_argument("stream", help="File to stream to the API")
args = parser.parse_args()
# Audio file samplerate check for
if args.stream.endswith(".wav"): # Only for .wav file
_, samplerate = sf.read(args.stream)
if ENCODING == pb.DecoderConfig.AudioEncoding.LINEAR16:
assert (
samplerate == SAMPLE_RATE
), f"SAMPLE_RATE must be same for using LINEAR16 encoding. Your Audio file Sample Rate is {samplerate}"
else:
pass
else:
pass
config = pb.DecoderConfig(
sample_rate=SAMPLE_RATE,
encoding=ENCODING,
use_itn=True,
use_disfluency_filter=False,
use_profanity_filter=False,
)
client = RTZROpenAPIClient(CLIENT_ID, CLIENT_SECRET)
client.transcribe_streaming_grpc(args.stream, config)
이상으로 리턴제로의 스트리밍 STT API를 활용하여, 마이크 입력을 실시간으로 텍스트로 변환하는 튜토리얼 코드의 설명을 마치겠습니다.
키워드 부스팅
"농협은행" -> "넘 예쁘네" 와 같이 발음이 비슷해서 잘못 인식하거나, 전문적인 용어처럼 인식의 어려움을 겪는 단어가 있다면 키워드 부스팅 기능을 사용해 보세요. 키워드 부스팅은 인식의 어려움을 겪는 단어들을 가중치와 함께 제시해 줌으로써, 인식률을 높여주는 기능입니다.
사용방법은 간단합니다. 아래와 같이 DecoderConfig 설정에 추가적으로 keywords 인자를 제시하면 됩니다.
config = pb.DecoderConfig(
sample_rate=SAMPLE_RATE,
encoding=ENCODING,
use_itn=True,
use_disfluency_filter=False,
use_profanity_filter=False,
keywords=[ ] # 키워드 부스팅 단어
)keywords 인자는 str 배열 형태로 제시되어야 합니다.
작성하는 방법은 단어:점수 와 같은 형태로 인식률을 높이고자 하는 단어와 지정할 스코어를 작성해주어야 합니다. 단어만 제시하고, 점수를 제시하지 않으면, 기본값인 2.0의 점수가 할당됩니다.
-5.0 ~ 5.0 으로 점수를 낮추면 반대로 인식률을 떨어트릴 수도 있습니다.예를 들어
- 농협은행 :
5.0 - 넘 예쁘네 :
-5.0
와 같이 키워드를 넘겨주고 싶다면, 아래와 같이 작성할 수 있습니다.
keywords=["농협은행:5.0", "넘 예쁘네:-5.0"]또한 그냥 "리턴제로"라는 단어의 인식률을 높이려고 한다면, 단순히 단어를 제시해 주는 것으로도 가능합니다.
keywords=["농협은행:5.0", "넘 예쁘네:-5.0", "리턴제로"]추가적으로 문장이 끝날 때마다 키워드 부스팅 단어를 변경하고 싶다면, 아래와 같이 코드를 변경하여 사용 가능합니다.
Bi-Directional 스트리밍 방식은 입력 스트리밍이 끝날 때, 응답 스트리밍도 종료됩니다. 수정된 아래의 코드는 반환된 인식 문장이 끝나면 강제로 입력 스트리밍을 닫습니다. 응답 스트리밍이 종료되면 다시 API에 요청을 보내기 전에, get_config()를 이용하여 DecoderConfig 를 원하는 설정으로 변경하고, 음성인식을 요청할 수 있습니다.
예시에서는 키워드를 두 개만 입력했기 때문에 bool 형태로 구현했지만, 응용한다면 더 많은 키워드를 입력할 수 있을 것입니다. 또, get_config()의 인자를 추가해 준다면, 더 많은 설정을 원하는 대로 조작할 수 있습니다.
...
def get_config(keywords=None):
"""
keywords를 받아서, config를 동적으로 생성하는 함수
"""
return pb.DecoderConfig(
sample_rate=SAMPLE_RATE,
encoding=ENCODING,
use_itn=True,
use_disfluency_filter=False,
use_profanity_filter=False,
keywords=keywords,
)
...
class RTZROpenAPIClient:
def __init__(self, client_id, client_secret):
...
def reset_stream(self):
"""
입력 스트림을 초기화 하는 함수.
"""
self.stream = MicrophoneStream(SAMPLE_RATE, CHUNK, CHANNELS, FORMAT)
...
def print_transcript(self, start_time, transcript, is_final=False):
clear_line_escape = "\033[K" # clear line Escape Sequence
if not is_final:
end_time_str = "~"
end_chr = "\r"
else:
end_time_str = f"~ {(start_time + transcript.duration / 1000):.2f}"
end_chr = None
print(f"{clear_line_escape}{start_time:.2f} {end_time_str} : {transcript.alternatives[0].text}", end=end_chr)
def transcribe_streaming_grpc(self):
"""
grpc를 이용해서 스트리밍 STT 수행 하는 메소드
"""
base = GRPC_SERVER_URL
with grpc.secure_channel(base, credentials=grpc.ssl_channel_credentials()) as channel:
stub = pb_grpc.OnlineDecoderStub(channel)
cred = grpc.access_token_call_credentials(self.token)
def req_iterator(keywords):
yield pb.DecoderRequest(streaming_config=get_config(keywords))
self.reset_stream()
for chunk in self.stream.generator():
yield pb.DecoderRequest(audio_content=chunk)
keyword_idx = False
keywords = {
True: ["농협은행:5.0", "넘 예쁘네:-5.0", "리턴제로"],
False: ["넘 예쁘네", "이턴제로"],
}
global_st_time = time.time()
while True:
keyword_idx = not keyword_idx
req_iter = req_iterator(keywords=keywords[keyword_idx])
resp_iter = stub.Decode(req_iter, credentials=cred)
session_st_time = time.time() - global_st_time
for resp in resp_iter:
resp: pb.DecoderResponse
for res in resp.results:
if res.is_final:
self.stream.terminate()
self.print_transcript(session_st_time, res, is_final=res.is_final)
def __del__(self):
self.stream.terminate()
소스 코드
전체 예시를 한번에 실행시킬 수 있는 코드는 아래를 참조하면 됩니다.
https://github.com/vito-ai/python-tutorial/tree/main/python-stt-sample
vito-ai참조
[1] MicrophoneStream: Transcribe audio from streaming input | Google Cloud
