실시간 SST - 리턴제로 api 를 통해서 실시간으로 음성을 텍스트로 변환하기

진행하고 있는 프로젝트에서는 실시간 SST 기술이 필요합니다.

 

 

하지만 현재 오픈 소스로 사용하고 있는 SST기술은 실시간이 아닌 파일을 업로드하여 텍스트로 변환하는 프로그램이 대부분입니다.

흔히 쓰는 SST API 기술 중에는 구글과 클로바 speech도 모두 실시간성이 아닌 파일을 업로드하거나

녹음을 끝내야 텍스트로 변환이 시작됩니다.

 

 

이와 다르게 리턴제로 음성인식 API는 실시간 텍스트 변환을 제공합니다. 

음성파일을 약 2초 간격으로 쪼개면서 텍스트를 변환하여 실시간처럼 텍스트를 제공합니다. 

 

 

이번 프로젝트에서 실시간성이라는 조건에 적합한  SST기술을 

리턴제로 음성인식 API 를 사용하여 실시간으로 텍스트로 변환하는 프로그램을 진행해 보려고 합니다. 

 

 

 

스트리밍 STT의 방식에는 gRPC와 WebSocket 두 가지 방식을 제공하고 있으나, gRPC를 이용한 방식을  다뤄볼려고 합니다.

 

 

개발 환경인 Python에서 마이크 입력을 인터페이스로 사용하기 위해서 PyAudio 라이브러리를 이용합니다.

모듈을 설치하기 위해 터미널에 아래 명령어를 통해 PyAudio 라이브러리를 설치합니다.

pip install pyaudio

 

 

그리고 스트리밍 STT의 방식 중 gRPC 을 사용하기 위한 셋팅을 진행합니다.

 

 

아래 깃허브를 통해서 .proto 파일을 받습니다

https://github.com/vito-ai/openapi-grpc/blob/main/protos/vito-stt-client.proto?ref=blog.rtzr.ai

openapi-grpc/protos/vito-stt-client.proto at main · vito-ai/openapi-grpc

 

 

아래 명령어를 통해 .proto 파일을 컴파일하기 위한 grpcio-tools 라이브러리를 설치합니다.

 

pip install grpcio-tools

 

 

아래의 명령어를 vito-stt-client.proto 파일이 있는 터미널창에서 입력하여 Protocol Buffer의 컴파일을 진행합니다.

 

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. ./vito-stt-client.proto

 

지금까지의 순서를 잘 따라왔다면 폴더 내에 2개의 파일이 생성된 점을 확인할수 있다. 

 

 

 

 

먼저 리턴제로의 API를 사용하기 위해서는 아래 링크를 통해서 애플리케이션 등록이 필요합니다.

 

RTZR STT

 

회원가입 > 개발자 콘솔> 어플리케이션 등록 >  CLIENT ID 와 CLIENT SECRET 키 받기

 

그리고 폴더 내에 config.ini 파일을 아래와 같이 생성하고 개발자 콘솔에서 CLIENT ID 와 CLIENT SECRET 키를 입력합니다.

 

[DEFAULT]
# 개발자 콘솔에서 발급받은 CLIENT ID, CLIENT SECRET을 입력하세요.
CLIENT_ID = ''
CLIENT_SECRET = ''

 

 

이렇게 하면 모든 코드를 돌리기 위해서 하는 세팅은 끝이 납니다.

 

이제부터는 본격적인 메인 소스코드에 대해  이야기 하고자 합니다. 

 

 

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)

 

1. __init__ 메서드: 클래스의 초기화 메서드로, 오디오 인터페이스를 초기화하고 오디오 스트림을 엽니다. 설정된 포맷, 채널, 샘플링 속도 등의 매개변수를 사용하여 오디오 스트림을 설정합니다.
2. _fill_buffer 메서드: 오디오 스트림에서 데이터를 수집하고 버퍼에 저장하는 콜백 함수입니다. 이 함수는 입력된 오디오 데이터를 버퍼에 넣어줍니다.
   
   
3. terminate 메서드: 오디오 스트림을 닫고 제너레이터를 종료하는 메서드입니다. 이 메서드는 클래스 인스턴스를 정리하고 종료할 때 호출됩니다.
   
   
4. generator 메서드: 오디오 청크를 생성하는 제너레이터를 반환합니다. 이 메서드는 무한 루프를 실행하여 버퍼에서 오디오 청크를 가져와 반환합니다. 버퍼가 닫힐 때까지 이 루프가 계속됩니다.

이 클래스는 오디오 스트림을 생성하고 버퍼에 오디오 데이터를 채우는 역할을 수행합니다

 

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()

 

또 위의 Class를 사용하기 위해서 리턴제로에서 제공한 RTZROpenAPIClient Class를 아래와 같이 변경해 줍니다.

 

if __name__ == "__main__":
    file_dir = os.path.dirname(os.path.abspath(__file__))
    par_dir = os.path.dirname(file_dir)

    env_config = configparser.ConfigParser()
    env_config.read(os.path.join(par_dir, "config.ini"))
    for section in env_config.sections():
        print(f"[{section}]")
        for key, value in env_config.items(section):
            print(f"{key} = {value}")
        print()

    # 설정 파일 경로
    config_file_path = ""

    # 설정 파일 읽기
    env_config = configparser.ConfigParser()
    env_config.read(config_file_path)
                    
    # 설정 파일 내용 출력
    for section in env_config.sections():
        print(f"[{section}]")
        for key, value in env_config.items(section):
            print(f"{key} = {value}")
        print()

    # 'CLIENT_ID' 키 가져오기
    client_id = env_config["DEFAULT"].get("CLIENT_ID")
    if client_id is None:
        print("CLIENT_ID가 설정 파일에 존재하지 않습니다.")
    else:
        print("CLIENT_ID:", client_id)

    client = RTZROpenAPIClient(env_config["DEFAULT"]["CLIENT_ID"], env_config["DEFAULT"]["CLIENT_SECRET"])
    try:
        client.transcribe_streaming_grpc()

    except KeyboardInterrupt:
        print("Program terminated by user.")
        del client

 

마지막으로 config.ini파일로부터 API키를 받을 수 있도록 메인을 작성합니다.

 

마지막으로 디렉토리는 아래 사진과 같은 모습으로 나타납니다.

 

STT를 중단하고 싶다면 Ctrl + C를 터미널에 입력하여 종료하면 됩니다.

 

실제 코드를 실행한 결과는 아래 영상을 첨부하도록 하겠습니다

 

 

본 블로그는 리턴제로에서 제공하는 튜토리올 블로글을 참고하여 작성하였습니다.

https://blog.rtzr.ai/grpc-tutorial-2-developers-api/#%EC%86%8C%EC%8A%A4-%EC%BD%94%EB%93%9C