라즈베리파이를 이용한 AI 스피커 만들기 프로젝트 - 중간점검

🍓🫐 오랜만에 돌아온 라파 프로젝트 ! 

정말 오랜만에 기록해보는 라파 프로젝트. 오늘은 지금까지 진행된 사항을 공유해보도록 하겠다.

뿅

---

 

 

AI Speaker with Raspberry Pi

두둥..

 

 

일단 우리 여섯 명은 하드웨어 셋, 소프트웨어 셋 이렇게 나눠서 활동하고 있다.

나는 소프트웨어 팀에 속해서 열심히 이런저런 것들을 하고 있다.

 

🖨️ HARDWARE

 

먼저 하드웨어 팀이 한 것들에 대해서 간단히 소개를 해보겠다. (하드웨어 팀의 발표 스피치를 참고로 적음.)

 

일단 어떻게 라즈베리 파이로 음성을 녹음할 수 있는지에 대해 알아보자.

• 먼저, 기본적인 컴퓨터와 달리 라즈베리파이에는 마이크 단자가 없어서 어댑터를 구매해서 어댑터와 마이크를 연결한뒤에 라파 보드와 연결
• 스위치는 GPIO라는 외부 소자와 연결을 컨트롤 할 수 있는 핀으로 스위치의 신호를 받아오고 스위치를 누르는 동안 녹음하도록 설계
• 녹음이 완료된 후에는 데이터가 wav음성 파일로 변환되고, 스피커와 라즈베리파이를 블루투스 혹은 유선연결로 녹음된 파일을 재생

 

 

음성 녹음을 하기 위해서는 라즈베리파이에 연결되어있는 장치 목록을 알아야 하는데, audiolist.py 코드를 통해서 장치 목록 리스트를 받아왔다.

 

리스트를 받아온 결과 세번째에 USB Audio Device를 찾았고, Index는 0부터 시작하기 때문에 deviceIndex는 2가 된다. 이 deviceIndex가 뒤에 코드에 쓰인다.

 

 

record.py는 음성녹음을 본격적으로 하는 코드다.

 

코드 일부를 보자. 아까 deviceIndex가 2였으므로 dev_index를 2로 설정,

스위치 GPIO 포트넘버는 6으로 설정했고, 녹음파일 저장형식은 wav파일로 저장했다.

스위치를 눌렀을 때, 누르지 않았을 때 녹음이 시작됨과 종료됨을 보이게 했다. 기본적인 로직 자체만 간략히 요약하자면, 마이크로 청크 단위로 음성 입력을 받고 이를 디지털화 하여 저장해두었다가 종료 버튼을 누르면 음성 파일로 인코딩하게 됨.

 

👩🏻‍💻 SOFTWARE

# 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 io
import os
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "[키 파일의 경로]"

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

import io
import os
from pygame import mixer
from google.cloud import texttospeech
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "[키 파일의 경로]"

def synthesize_text(text):
    client = texttospeech.TextToSpeechClient()

    max_length = 200    
    words = text.split('. ')
    sentences = []
    current_sentence = ''

    for word in words:
        if len(current_sentence + word) <= max_length:
            current_sentence += word + ' '
        else:
            sentences.append(current_sentence.strip() + '.')
            current_sentence = word + ' '
    if current_sentence:
        sentences.append(current_sentence.strip() + '.')

    audio_data = []

    for sentence in sentences:
        input_text = texttospeech.SynthesisInput(text=sentence)

        voice = texttospeech.VoiceSelectionParams(
            language_code="ko-KR",
            name="ko-KR-Neural2-C",
            ssml_gender=texttospeech.SsmlVoiceGender.MALE,
        )

        audio_config = texttospeech.AudioConfig(
            audio_encoding=texttospeech.AudioEncoding.MP3
        )

        response = client.synthesize_speech(
            request={"input": input_text, "voice": voice, "audio_config": audio_config}
        )

        audio_data.append(response.audio_content)
  
    audio_data = b"".join(audio_data)
    
    # Play the audio directly
    mixer.init()
    mixer.music.load(io.BytesIO(audio_data))
    mixer.music.play()

    # Add a delay to allow time for audio playback
    while mixer.music.get_busy():
        pass

    print('오디오 실행 완료')


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 get_last_line(file_path):
    with open(file_path, 'r', encoding='utf-8') as file:
        lines = file.readlines()
        if lines:
            return lines[-1].strip()
        else:
            return None

def main() -> None:
    """Transcribe speech from audio file."""
    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
    )

    transcriptions = []  # List to store transcriptions

    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

            result = response.results[0]
            if not result.alternatives:
                continue

            transcript = result.alternatives[0].transcript

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

            transcriptions.append(transcript)

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

    # Save transcriptions to a file
    with open("transcriptions.txt", "w") as file:
        for transcription in transcriptions:
            file.write(transcription + "\n")
    # Read the last line from the file
    last_line = get_last_line('transcriptions.txt')

    # If there's a last line, synthesize the text
    if last_line:
        synthesize_text(last_line)
        print(f"Synthesizing the last line: {last_line}")
    else:
        print("No text found in the file.")

    print("Transcriptions saved to 'transcriptions.txt'.")

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

 

** 나의 영혼의 친구 챗지피티랑 함께 만든 거라서 코드가 깔끔하지 않을 수 있음...

 

 

GCP를 이용해서 STT + TTS 코드를 만들었다. 음성을 입력 받고 텍스트로 변환한 후, 이 텍스트를 다시 mp3 음성으로 내보내는 것이다. 

"종료" 나 "끝내기" 라고 외치면 프로그램이 종료되는데, 종료되는 동시에 음성이 나오고 txt 파일도 나오게 했다.

 

 

👩🏻‍💻 SW 팀의 일원으로서 앞으로 해야할 것들. . .

 

위에서 "종료"나 "끝내기"라고 말했을 때 디바이스가 끝나는 것처럼, 반대로 특정 단어를 외쳤을 때 디바이스가 깨어날 수 있는 트리거 코드를 짜야한다. 대표적인 예로 우리가 "시리야~"라고 불렀을 때 시리가 반응하는 것이 있다.

 

 

실시간 통역이나, 추천 시스템 같은 기능이 있는 스피커를 만들 것 같은데 . . . 이 부분은 공부를 좀 해야겠다.

 

---

 

 

 

아무튼 이제 우리에게 남은 것은.. 융합

하드웨어는 음성 입력을 받고 출력할 준비가 되었고, 소프트웨어 팀 역시 음성 입력을 받아서 음성 출력을 내보내는 모델이 구성이 되었다.

그래서 이제 중간에 라즈베리 파이는 모델을 돌릴 능력이 없으니 중간 컴퓨터 서버를 두어서 중간 처리만 해주면 된다. 이 둘을 연결할 때 바로 api라는 것이 등장하고, 이 부분만 해결해주면 저희 인공지능 스피커가 완성될 것 같습니다!