구글 클라우드 플랫폼(GCP)를 통해 음성, 언어, 텍스트 API 이용해보기 - (2)

📌 할 일

• STT → TTS 샘플로 확인하는 것 GCP API
• GCP TTS 사용해서 다양한 음성 합성하는 방법 확인하기 (Custom data로 학습시켜서 음성 출력)

이번 포스팅은 STT API를 사용 기록!!

+) macOS 쓰는 사람의 포스팅임

---

📟 STT (Speech-to-Text) API 사용

STT란 Speech-to-text의 약자로 말 그대로 말하는 것(음성)을 문자(텍스트)로 바꿔주는 AI기술

 

STT API를 사용해보겠다...!

 

📎 참고 사이트 : https://jvvp.tistory.com/1107

GCP API Speech to text 사용하기 -1

 

☁️ 파이썬 가상환경 생성 및 활성화

python -m venv rapa-venv

→ 나는 이름을 rapa-venv라고 지었다. 라즈베리파이 프로젝트이기 때문에 줄여서 라파,,ㅋㅋㅋㅋ

 

source venv/bin/activate

 

→ 활성화시키기! macOS면 그대로 따라하시면 됩니다! 

 

짠

☁️ gcloud 초기 설정

로컬 환경에서 Cloud Platform 관련된 연동 및 로컬 테스트를 수행하기 위해서는 SDK를 설치해야 한다.

 

Google Cloud SDK는 주로 "gcloud"라고 알려진 Google Cloud CLI의 일부입니다. "gcloud" 명령을 사용하여 Google Cloud Platform (GCP) 서비스를 관리하고, 프로젝트, 인스턴스, 스토리지, 데이터베이스 등을 설정하고 제어할 수 있습니다. Google Cloud SDK는 GCP와 상호 작용하고 GCP 리소스를 관리하는 데 사용되는 도구와 라이브러리 모음입니다. Google Cloud SDK를 설치하면 "gcloud" 명령을 사용할 수 있으며, 이를 통해 클라우드 리소스를 관리할 수 있습니다.

https://manchann.tistory.com/24

[GCP] gcloud 초기 설정

↑ 나는 위 링크 보면서 따라했다.

 

Your Google Cloud SDK is configured and ready to use!

 

☁️  필요한 라이브러리 설치!

 

pip install google-cloud-storage
pip install google-cloud-speech
pip install pyaudio

 

근데 pyaudio 설치하면서 오류가 생겼다. 이유를 찾아보니 아래와 같았다.

 

패키지 종속성 설치: PyAudio의 빌드 오류는 종속성 패키지가 설치되지 않았거나 올바르게 설치되지 않은 경우에 발생할 수 있습니다. 다음 패키지를 설치해보세요.

• PortAudio: PyAudio의 하부 라이브러리인 PortAudio를 설치해야 합니다. macOS에서는 **brew**를 사용하여 설치할 수 있습니다.

brew install portaudio

→ macOS는 이렇게 설치! portaudio 설치하고 pyaudio 설치하면 된다.

 

+) 이거 말고도 더 설치해야 하는 라이브러리가 있을 수도 있다! 

 

☁️  샘플 코드

https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/speech/microphone/transcribe_streaming_mic.py

 

위에 깃헙 링크 들어가서 코드 복사하기.

 

language_code = 'ko-KR'

 

→ 언어를 우리나라로 수정한다.

 

☁️  환경변수 설정

코드를 실행할 때마다 키 파일을 이용해서 인증을 받아야 한다. 환경 변수를 설정하여 애플리케이션에서 인증 정보를 찾을 수 있도록 한다.

export GOOGLE_APPLICATION_CREDENTIALS=path/to/your/credentials.json

 

이렇게 하면 뭔가 되는 것 같다. 근데 실행했더니 맘에 안 드는 부분이 좀 몇 가지 있어서(종료할 때 'exit'말고 '끝내기'라고 말하면 인식되기, 1초 만에 사라지는 터미널 속 텍스트들...) 챗지피티한테 고쳐달라고 부탁했다. 그래서 최종 코드는... !

 

# Copyright 2017 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Google Cloud Speech API sample application using the streaming API.

NOTE: This module requires the additional dependency `pyaudio`. To install
using pip:

    pip install pyaudio

Example usage:
    python transcribe_streaming_mic.py
"""

# [START speech_transcribe_streaming_mic]

import queue
import re
import sys

from google.cloud import speech

import pyaudio

# Audio recording parameters
RATE = 16000
CHUNK = int(RATE / 10)  # 100ms


class MicrophoneStream:
    """Opens a recording stream as a generator yielding the audio chunks."""

    def __init__(self: object, rate: int = RATE, chunk: int = CHUNK) -> None:
        """The audio -- and generator -- is guaranteed to be on the main thread."""
        self._rate = rate
        self._chunk = chunk

        # Create a thread-safe buffer of audio data
        self._buff = queue.Queue()
        self.closed = True

    def __enter__(self: object) -> object:
        self._audio_interface = pyaudio.PyAudio()
        self._audio_stream = self._audio_interface.open(
            format=pyaudio.paInt16,
            # The API currently only supports 1-channel (mono) audio
            # https://goo.gl/z757pE
            channels=1,
            rate=self._rate,
            input=True,
            frames_per_buffer=self._chunk,
            # Run the audio stream asynchronously to fill the buffer object.
            # This is necessary so that the input device's buffer doesn't
            # overflow while the calling thread makes network requests, etc.
            stream_callback=self._fill_buffer,
        )

        self.closed = False

        return self

    def __exit__(
        self: object,
        type: object,
        value: object,
        traceback: object,
    ) -> None:
        """Closes the stream, regardless of whether the connection was lost or not."""
        self._audio_stream.stop_stream()
        self._audio_stream.close()
        self.closed = True
        # Signal the generator to terminate so that the client's
        # streaming_recognize method will not block the process termination.
        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:
        """Continuously collect data from the audio stream, into the buffer.

        Args:
            in_data: The audio data as a bytes object
            frame_count: The number of frames captured
            time_info: The time information
            status_flags: The status flags

        Returns:
            The audio data as a bytes object
        """
        self._buff.put(in_data)
        return None, pyaudio.paContinue

    def generator(self: object) -> object:
        """Generates audio chunks from the stream of audio data in chunks.

        Args:
            self: The MicrophoneStream object

        Returns:
            A generator that outputs audio chunks.
        """
        while not self.closed:
            # Use a blocking get() to ensure there's at least one chunk of
            # data, and stop iteration if the chunk is None, indicating the
            # end of the audio stream.
            chunk = self._buff.get()
            if chunk is None:
                return
            data = [chunk]

            # Now consume whatever other data's still buffered.
            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)


def listen_print_loop(responses: object) -> str:
    """Iterates through server responses and prints them.

    The responses passed is a generator that will block until a response
    is provided by the server.

    Each response may contain multiple results, and each result may contain
    multiple alternatives; for details, see https://goo.gl/tjCPAU.  Here we
    print only the transcription for the top alternative of the top result.

    In this case, responses are provided for interim results as well. If the
    response is an interim one, print a line feed at the end of it, to allow
    the next result to overwrite it, until the response is a final one. For the
    final one, print a newline to preserve the finalized transcription.

    Args:
        responses: List of server responses

    Returns:
        The transcribed text.
    """
    num_chars_printed = 0
    for response in responses:
        if not response.results:
            continue

        # The `results` list is consecutive. For streaming, we only care about
        # the first result being considered, since once it's `is_final`, it
        # moves on to considering the next utterance.
        result = response.results[0]
        if not result.alternatives:
            continue

        # Display the transcription of the top alternative.
        transcript = result.alternatives[0].transcript

        # Display interim results, but with a carriage return at the end of the
        # line, so subsequent lines will overwrite them.
        #
        # If the previous result was longer than this one, we need to print
        # some extra spaces to overwrite the previous result
        overwrite_chars = " " * (num_chars_printed - len(transcript))

        if not result.is_final:
            sys.stdout.write(transcript + overwrite_chars + "\r")
            sys.stdout.flush()

            num_chars_printed = len(transcript)

        else:
            print(transcript + overwrite_chars)

            # Exit recognition if any of the transcribed phrases could be
            # one of our keywords.
            if re.search(r"\b(exit|quit)\b", transcript, re.I):
                print("Exiting..")
                break

            num_chars_printed = 0

        return transcript
    


def main() -> None:
    """Transcribe speech from audio file."""
    # See http://g.co/cloud/speech/docs/languages
    # for a list of supported languages.
    language_code = "ko-KR"  # a BCP-47 language tag

    client = speech.SpeechClient()
    config = speech.RecognitionConfig(
        encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
        sample_rate_hertz=RATE,
        language_code=language_code,
    )

    streaming_config = speech.StreamingRecognitionConfig(
        config=config, interim_results=True
    )

    with MicrophoneStream(RATE, CHUNK) as stream:
        audio_generator = stream.generator()
        requests = (
            speech.StreamingRecognizeRequest(audio_content=content)
            for content in audio_generator
        )

        responses = client.streaming_recognize(streaming_config, requests)

        for response in responses:
            if not response.results:
                continue

            # The `results` list is consecutive. For streaming, we only care about
            # the first result being considered, since once it's `is_final`, it
            # moves on to considering the next utterance.
            result = response.results[0]
            if not result.alternatives:
                continue

            # Display the transcription of the top alternative.
            transcript = result.alternatives[0].transcript

            # Print the recognized text to the terminal
            print("Recognized:", transcript)

            # Check for exit keywords in Korean
            if re.search(r"\b(종료|끝내기)\b", transcript):
                print("프로그램을 종료합니다.")
                break

if __name__ == "__main__":
    main()
# [END speech_transcribe_streaming_mic]

 

 

기여운 우리팀 ㅋㅋㅋㅋ <3

---

세세하게 기록하면서 한 실습이 아니라, 누락되거나 생략된 부분이 있을 수도 있음,,, ㅜㅜ