by Troye
리턴제로는 개발 가이드 Developers 사이트에서 리턴제로의 우수한 음성인식 API를 사용하는 방법을 제공합니다. 리턴제로의 음성인식 API에는 음성파일을 한 번에 텍스트로 변환하는 일반 STT와 음성파일을 쪼개어 실시간으로 변환하는 스트리밍 STT가 있습니다. 스트리밍 STT의 방식에는 gRPC와 WebSocket 두 가지 방식을 제공하고 있으나, 오늘은 gRPC를 이용한 방식을 사용하도록 하겠습니다.
이번 튜토리얼은 2편에 걸쳐서 진행할 예정입니다.
- 1. 기본적인 gRPC STT API 사용법
- 2. 마이크 입력을 통한 STT API 사용예제
gRPC 셋팅
gRPC는 Google에서 개발한 Remote Procedure Call 프레임워크입니다. gRPC는 Protocol Buffer와 HTTP/2 기반의 전송방식을 이용하여 효율적으로 통신합니다. gRPC는 IDL(Interface Definition Language)를 이용해서 서버(Remote)의 자원을 다양한 환경의 클라이언트에서도 쉽게 활용이 가능합니다.
클라이언트에서 gRPC API를 이용하기 위해선, 함수명이나 인자, 반환 값 등에 대한 데이터형이 정의된 Protocol Buffer .proto IDL 파일을 컴파일하여, “Stub”을 생성해야 합니다.
먼저 .proto 파일을 [링크]에서 받아줍니다.
wget https://raw.github.com/vito-ai/openapi-grpc/main/protos/vito-stt-client.proto다음은 .proto 파일을 컴파일하기 위해서, grpcio-tools 라이브러리를 설치해야 합니다.
pip install grpcio-tools
grpcio-tools 의 설치가 완료 됐다면, 아래의 명령어를 vito-stt-client.proto 파일이 있는 디렉터리에서 입력하여 Protocol Buffer의 컴파일을 진행합니다.
python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. ./vito-stt-client.proto
그럼 동일 폴더 내에 아래와 같은 파일 두 개가 생긴 것을 확인할 수 있습니다.
.
├── vito-stt-client.proto
├── vito_stt_client_pb2.py
└── vito_stt_client_pb2_grpc.py
생성된 파일들엔 통신에 필요한 Stub, 메소드 등이 담겨있습니다.
이제 gRPC를 이용하여 통신할 준비는 끝이 났습니다.
gRPC 스트리밍 API 사용해보기
먼저 리턴제로의 API를 사용하기 위해서는 Developers 사이트 회원가입과 애플리케이션 등록이 필요합니다.
회원가입을 완료하셨다면, 개발자 콘솔에 접속하여 애플리케이션 등록을 진행합니다.
애플리케이션 등록이 완료됐다면, 제공되는 CLIENT ID 와 CLIENT SECRET을 저장해두어야만 합니다!
이렇게 API 사용 준비를 마쳤습니다.
리턴제로 Developers 사이트에서는 사용자들의 API 사용을 위해 개발 가이드를 제공합니다.
Python 환경에서 로컬 음성파일을 gRPC 스트리밍 STT API를 이용해서 텍스트로 변환하기 위해 개발 가이드의 샘플 코드를 구동해 보겠습니다.
먼저 API사용을 위해 액세스 토큰을 발급받아야합니다.
그러기 위해선 애플리케이션을 등록하면서 발급받은 CLIENT_ID, CLIENT_SECRET 값이 필요합니다. 자신의 값을 config.ini 파일을 만들고 아래와 같은 형태로 입력해 주세요!
# File Name : config.ini
[DEFAULT]
# 개발자 콘솔에서 발급받은 CLIENT ID, CLIENT SECRET을 입력하세요.
CLIENT_ID = YOUR_CONFIG_ID
CLIENT_SECRET = YOUR_CONFIG_SECRET이제 샘플코드를 하나씩 살펴보겠습니다.
"""
File Name:
vitoopenapi-stt-streaming-sample.py
Example usage:
python sample_stt.py sample/filepath
"""
import argparse
import configparser
import logging
import os
import time
from io import DEFAULT_BUFFER_SIZE
import grpc
import soundfile as sf
import vito_stt_client_pb2 as pb
import vito_stt_client_pb2_grpc as pb_grpc
from requests import Session
API_BASE = "https://openapi.vito.ai"
GRPC_SERVER_URL = "grpc-openapi.vito.ai:443"
SAMPLE_RATE = 8000
ENCODING = pb.DecoderConfig.AudioEncoding.LINEAR16
BYTES_PER_SAMPLE= 2
# 본 예제에서는 스트리밍 입력을 음성파일을 읽어서 시뮬레이션 합니다.
# 실제사용시에는 마이크 입력 등의 실시간 음성 스트림이 들어와야합니다.
class FileStreamer:
def __init__(self,filepath):
self.filepath = filepath
self.file = None
def __enter__(self):
self.file = open(self.filepath,"rb")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.file.close()
os.remove(self.filepath)
def read(self, size):
if size > 1024 * 1024:
size = 1024*1024
time.sleep(size / (SAMPLE_RATE*BYTES_PER_SAMPLE))
content = self.file.read(size)
return content
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
@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, filepath, 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)
def req_iterator():
yield pb.DecoderRequest(streaming_config=config)
with FileStreamer(filepath) as f:
while True:
buff = f.read(size=DEFAULT_BUFFER_SIZE)
if buff is None or len(buff) == 0:
break
yield pb.DecoderRequest(audio_content=buff)
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" + "{:.2f} : {}".format(res.start_at / 1000, res.alternatives[0].text),
end="\r",
flush=True,
)
else:
print(
"\033[K"
+ "{:.2f} - {:.2f} : {}".format(
res.start_at / 1000,
(res.start_at + res.duration) / 1000,
res.alternatives[0].text,
),
end="\n",
)
if __name__ == "__main__":
file_dir = os.path.dirname(os.path.abspath(__file__))
par_dir = os.path.dirname(file_dir)
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
env_config = configparser.ConfigParser()
env_config.read(os.path.join(par_dir, "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(env_config["DEFAULT"]["CLIENT_ID"], env_config["DEFAULT"]["CLIENT_SECRET"])
client.transcribe_streaming_grpc(args.stream, config)샘플코드를 라이브러리 선언을 제외한 구역별로 분해해서 보겠습니다.
필요 변수
가장 먼저 애플리케이션 실행에 사용될 정보들을 선언하는 부분입니다.
API_BASE = "https://openapi.vito.ai"
GRPC_SERVER_URL = "grpc-openapi.vito.ai:443"
SAMPLE_RATE = 8000
ENCODING = pb.DecoderConfig.AudioEncoding.LINEAR16API_BASE: API를 사용하기 위한 기본적인 도메인입니다.GRPC_SERVER_URL: gRPC통신을 하기 위한 기본적인 도메인입니다.SAMPLE_RATE: 음원의 SAMPLE_RATE 정보입니다. 여기선 8,000Hz를 사용합니다.ENCODING: 음원의 인코딩 정보입니다. 여기선 LINEAR16을 사용합니다.
✅ 리턴제로의 스트리밍 API에서는 아래의 인코딩 코덱을 지원합니다.
LINEAR16, WAV, FLAC, MULAW, ALAW, AMR, AMR_WB, OGG_OPUS, OPUS
RTZROpenAPI 클래스
RTZROpenAPI 클래스에는 리턴제로의 스트리밍 API를 사용하기 위해 필요한 함수와 정보들이 담겨있습니다.
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
@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, filepath, 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)
def req_iterator():
yield pb.DecoderRequest(streaming_config=config)
with FileStreamer(filepath) as f:
while True:
buff = f.read(size=DEFAULT_BUFFER_SIZE)
if buff is None or len(buff) == 0:
break
yield pb.DecoderRequest(audio_content=buff)
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" + "{:.2f} : {}".format(res.start_at / 1000, res.alternatives[0].text),
end="\r",
flush=True,
)
else:
print(
"\033[K"
+ "{:.2f} - {:.2f} : {}".format(
res.start_at / 1000,
(res.start_at + res.duration) / 1000,
res.alternatives[0].text,
),
end="\n",
)클래스 내부에 선언된 함수들을 나눠서 설명하겠습니다.
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
클래스의 생성자 부분에는 API사용을 위한 정보들을 제공받습니다.
그중 client_id, client_secret 은 액세스 토큰을 생성하기 위해 위에서 애플리케이션을 생성할 때 발급받은 인자들입니다.
_sess, _token 은 각각 액세스 토큰을 발급받기 위해서 필요한 세션 객체, 내부 저장 변수입니다.
다음은 token 프로퍼티입니다.
@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"]
리턴제로의 액세스 토큰은 만료 기간이 6시간입니다. 그렇기에 주기적인 갱신이 필요합니다.
따라서 토큰 갱신을 위한 로직이 포함된 프로퍼티를 이용합니다.
이 로직은 토큰이 발급되지 않았거나, 만료 시간이 지난 토큰을 위해 다시 발급을 진행합니다.
토큰 발급을 위해 만들어둔 세션을 이용해, 토큰 발급 API인 /v1/authenticate 으로 client_id 와 client_secret을 액세스 토큰을 요청하고, 반환합니다.
마지막은 transcribe_streaming_grpc 함수입니다.
이 함수는 gRPC서버에 음성 정보를 보내고, 변환된 텍스트의 출력을 담당합니다.
def transcribe_streaming_grpc(self, filepath, 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)
def req_iterator():
yield pb.DecoderRequest(streaming_config=config)
with FileStreamer(filepath) as f:
while True:
buff = f.read(size=DEFAULT_BUFFER_SIZE)
if buff is None or len(buff) == 0:
break
yield pb.DecoderRequest(audio_content=buff)
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" + "{:.2f} : {}".format(res.start_at / 1000, res.alternatives[0].text),
end="\r",
flush=True,
)
else:
print(
"\033[K"
+ "{:.2f} - {:.2f} : {}".format(
res.start_at / 1000,
(res.start_at + res.duration) / 1000,
res.alternatives[0].text,
),
end="\n",
)
함수 내에 중요한 부분 위주로 설명을 진행하겠습니다.
pb_grpc.OnlineDecoderStub(object channel): 위에서 설명했던 Stub 객체를 생성하는 과정입니다. 통신에 필요한 서버의 함수나 자원을 로컬에 있는 것처럼 사용할 수 있게 됩니다. Stub 객체 생성을 위해선grpc.Channel오브젝트가 필요합니다. 여기선 with 컨텍스트 안에서,secure_channel함수로 생성된 오브젝트를 이용합니다.grpc.access_token_call_credentials(str token): Access Token을 넘겨서 Authentication이 필요할 때 사용할CallCredentials객체를 반환받습니다. Http Request에서 Authorization을 위해 “authorization: Bearer <Access_Token>”을 헤더에 포함하여 요청하는 것과 상응합니다.req_iterator(): 스트리밍으로 Client 측에서 전송을 진행하므로, 서버에 request를 위해 iterator 객체를 넘깁니다.DecoderRequest()함수는streaming_config,audio_content를 인자로 받습니다. 리턴제로 API는 처음에는 설정 정보인streaming_config만 전송하고, 그 후에 실제 텍스트 변환을 위한 음성정보를audio_content로 넘겨줍니다.- 스트리밍 방식이기에 데이터를 분할하여 넣어줄 수 있도록 구현하는 것을 권장합니다.
stub.Decode(): 요청에 필요한 이터레이터와 엑세스토큰을 포함한 Credential 객체를 넘겨서 서버에 STT를 하도록 요청합니다. 양방향 스트리밍이기에 반환되는 객체 또한 이터레이터 형태입니다.
API를 사용하여 정상적으로 STT가 수행됐다면, 반환된 이터레이터의 출력을 진행합니다.
for resp in resp_iter:
resp: pb.DecoderResponse
for res in resp.results:
if not res.is_final:
print(
"\033[K" + "{:.2f} : {}".format(res.start_at / 1000, res.alternatives[0].text),
end="\r",
flush=True,
)
else:
print(
"\033[K"
+ "{:.2f} - {:.2f} : {}".format(
res.start_at / 1000,
(res.start_at + res.duration) / 1000,
res.alternatives[0].text,
),
end="\n",
)
반환된 객체는 문장의 종료 여부인 is_final, 음성에 맞게 변환된 텍스트들인 alternatives 키 등을 가지고 있습니다. 출력에 필요한 핵심 데이터는 이렇게 필요합니다.
다양한 반환된 키는 Developers 사이트 링크를 참조하세요.
이로써 gRPC를 이용한 스트리밍 STT API의 기본적인 사용법 설명을 마칩니다.
vito-ai참조
스트리밍 STT - GRPC | RTZR Developers
Introduction to gRPC | gRPC
