ROS 2 Humble과 TurtleBot3 Burger에서 Google Cloud Text-to-Speech로 문자열 음성 출력하기 #2

31) 볼륨 증폭 예제

경고 메시지를 크게 출력하려면 다음과 같이 실행합니다.

python3 say.py 경고합니다 장애물이 감지되었습니다 --volume-gain-db 4.0

조용하게 출력하려면 다음과 같이 실행합니다.

python3 say.py 시스템 대기 상태입니다 --volume-gain-db -3.0

지나치게 큰 값을 사용하면 음질이 저하되거나 왜곡될 수 있으므로 실제 스피커 환경에서 조정해야 합니다.

32) 언어와 음성 모델 변경 예제

영어 음성을 사용하려면 다음처럼 언어 코드와 음성 이름을 함께 변경합니다.

python3 say.py System startup complete --language-code en-US \
    --voice-name en-US-Standard-A

일본어 음성을 사용하려면 다음과 같이 실행할 수 있습니다.

python3 say.py システムを開始します --language-code ja-JP \
    --voice-name ja-JP-Standard-A

음성 이름은 실제 Google Cloud에서 지원하는 값인지 확인해야 합니다.

33) 모든 옵션을 함께 사용하는 예제

python3 say.py 드론 배송 임무를 시작합니다 \
    --language-code ko-KR \
    --voice-name ko-KR-Standard-A \
    --speaking-rate 0.95 \
    --pitch -1.0 \
    --volume-gain-db 2.0 \
    --audio-device plughw:2,0

내부적으로 다음과 비슷한 객체가 생성됩니다.

engine = GoogleTtsEngine(
    language_code="ko-KR",
    voice_name="ko-KR-Standard-A",
    speaking_rate=0.95,
    pitch=-1.0,
    volume_gain_db=2.0,
    audio_device="plughw:2,0",
)

음성으로 출력되는 문자열은 다음과 같습니다.

드론 배송 임무를 시작합니다

34) 긴 문장 입력 예제

쉘에서 긴 문장을 입력할 때는 따옴표를 사용하는 것이 안전합니다.

python3 say.py "현재 위치에 도착했습니다. 잠시 후 착륙을 시작합니다."

따옴표 없이 입력해도 현재 코드에서는 정상 동작합니다.

python3 say.py 현재 위치에 도착했습니다 잠시 후 착륙을 시작합니다

다만 쉘에서 특별한 의미를 갖는 문자가 포함되면 따옴표가 필요합니다.

예를 들어 다음 기호가 포함된 경우 주의해야 합니다.

&
|
>
<
;
$
*
괄호

안전한 방식은 전체 문장을 큰따옴표로 묶는 것입니다.

python3 say.py "배터리 잔량은 30%입니다. 충전이 필요합니다."

10. ROS 2 토픽 구독 노드 작성

google_tts_ros2/tts_node.py 파일을 작성합니다.

import queue
import threading
from typing import Optional

import rclpy
from google.api_core.exceptions import (
    GoogleAPICallError,
    RetryError,
)
from rclpy.node import Node
from std_msgs.msg import String

from google_tts_ros2.tts_engine import GoogleTtsEngine


class GoogleTtsNode(Node):
    def __init__(self) -> None:
        super().__init__("google_tts_node")

        self.declare_parameter(
            "topic_name",
            "/tts",
        )
        self.declare_parameter(
            "language_code",
            "ko-KR",
        )
        self.declare_parameter(
            "voice_name",
            "ko-KR-Standard-A",
        )
        self.declare_parameter(
            "speaking_rate",
            1.0,
        )
        self.declare_parameter(
            "pitch",
            0.0,
        )
        self.declare_parameter(
            "volume_gain_db",
            0.0,
        )
        self.declare_parameter(
            "audio_device",
            "default",
        )
        self.declare_parameter(
            "speech_queue_size",
            10,
        )
        self.declare_parameter(
            "startup_text",
            "",
        )

        topic_name = str(
            self.get_parameter("topic_name").value
        )

        queue_size = max(
            1,
            int(
                self.get_parameter(
                    "speech_queue_size"
                ).value
            ),
        )

        self._engine = GoogleTtsEngine(
            language_code=str(
                self.get_parameter(
                    "language_code"
                ).value
            ),
            voice_name=str(
                self.get_parameter(
                    "voice_name"
                ).value
            ),
            speaking_rate=float(
                self.get_parameter(
                    "speaking_rate"
                ).value
            ),
            pitch=float(
                self.get_parameter(
                    "pitch"
                ).value
            ),
            volume_gain_db=float(
                self.get_parameter(
                    "volume_gain_db"
                ).value
            ),
            audio_device=str(
                self.get_parameter(
                    "audio_device"
                ).value
            ),
        )

        self._speech_queue: queue.Queue[
            Optional[str]
        ] = queue.Queue(
            maxsize=queue_size
        )

        self._stop_event = threading.Event()

        self._worker = threading.Thread(
            target=self._worker_loop,
            name="google_tts_worker",
            daemon=True,
        )

        self._worker.start()

        self._subscription = self.create_subscription(
            String,
            topic_name,
            self._topic_callback,
            10,
        )

        self.get_logger().info(
            "Google Cloud TTS 준비 완료: "
            f"topic={topic_name}"
        )

        startup_text = str(
            self.get_parameter(
                "startup_text"
            ).value
        ).strip()

        if startup_text:
            self._enqueue(startup_text)

    def _topic_callback(
        self,
        message: String,
    ) -> None:
        text = message.data.strip()

        if not text:
            self.get_logger().warning(
                "빈 문자열을 수신하여 무시했습니다."
            )
            return

        self._enqueue(text)

    def _enqueue(
        self,
        text: str,
    ) -> None:
        try:
            self._speech_queue.put_nowait(text)

            self.get_logger().info(
                f"TTS 대기열 추가: {text}"
            )

        except queue.Full:
            self.get_logger().warning(
                "TTS 대기열이 가득 차서 "
                "새 문자열을 버렸습니다."
            )

    def _worker_loop(self) -> None:
        while not self._stop_event.is_set():
            try:
                text = self._speech_queue.get(
                    timeout=0.2
                )

            except queue.Empty:
                continue

            if text is None:
                self._speech_queue.task_done()
                break

            try:
                self.get_logger().info(
                    "음성 합성 및 재생 시작: "
                    f"{text}"
                )

                self._engine.speak(text)

            except (
                GoogleAPICallError,
                RetryError,
            ) as error:
                self.get_logger().error(
                    "Google Cloud TTS API 오류: "
                    f"{error}"
                )

            except FileNotFoundError:
                self.get_logger().error(
                    "aplay를 찾을 수 없습니다. "
                    "alsa-utils를 설치하세요."
                )

            except Exception as error:
                self.get_logger().error(
                    f"TTS 처리 실패: {error}"
                )

            finally:
                self._speech_queue.task_done()

    def destroy_node(self) -> bool:
        self._stop_event.set()

        try:
            self._speech_queue.put_nowait(None)

        except queue.Full:
            pass

        if self._worker.is_alive():
            self._worker.join(timeout=1.0)

        return super().destroy_node()


def main(args=None) -> None:
    rclpy.init(args=args)

    node = None

    try:
        node = GoogleTtsNode()
        rclpy.spin(node)

    except KeyboardInterrupt:
        pass

    finally:
        if node is not None:
            node.destroy_node()

        if rclpy.ok():
            rclpy.shutdown()


if __name__ == "__main__":
    main()

1) 필요한 Python 모듈

소스 상단에서는 다음 모듈을 가져옵니다.

import queue
import threading
from typing import Optional

import rclpy
from google.api_core.exceptions import (
    GoogleAPICallError,
    RetryError,
)
from rclpy.node import Node
from std_msgs.msg import String

from google_tts_ros2.tts_engine import GoogleTtsEngine

각 모듈의 역할은 다음과 같습니다.

queue 모듈

import queue

queue 모듈은 여러 스레드 사이에서 데이터를 안전하게 전달하기 위해 사용합니다.

이 노드에서는 ROS 2 토픽 콜백에서 수신한 문자열을 음성 출력 작업 스레드로 전달하는 용도로 사용합니다.

threading 모듈

import threading

Google Cloud TTS 음성 합성과 오디오 재생은 시간이 걸리는 작업입니다.

이 작업을 ROS 2 토픽 콜백 안에서 직접 수행하면 음성 재생이 끝날 때까지 콜백 함수가 반환되지 않습니다.

그 결과 다음 메시지 처리가 지연될 수 있습니다.

이를 방지하기 위해 별도의 스레드에서 음성 합성과 재생을 수행합니다.

Optional 타입

from typing import Optional

음성 대기열에는 일반 문자열과 종료 신호인 None이 들어갈 수 있습니다.

따라서 대기열의 타입을 다음과 같이 선언합니다.

queue.Queue[Optional[str]]

이는 대기열에 str 또는 None이 저장될 수 있다는 의미입니다.

rclpy

import rclpy
from rclpy.node import Node

rclpy는 ROS 2 Python 클라이언트 라이브러리입니다.

Node 클래스를 상속해 사용자 정의 ROS 2 노드를 만들고, rclpy.spin()을 통해 콜백을 처리합니다.

Google Cloud 예외 클래스

from google.api_core.exceptions import (
    GoogleAPICallError,
    RetryError,
)

Google Cloud TTS API 호출 중 발생할 수 있는 오류를 구분해 처리하기 위해 사용합니다.

GoogleAPICallError는 Google API 호출 과정에서 발생한 일반적인 오류를 나타냅니다.

RetryError는 Google API 클라이언트가 요청을 재시도했지만 최종적으로 성공하지 못했을 때 발생합니다.

std_msgs 메시지

from std_msgs.msg import String

음성으로 변환할 문자열을 ROS 2 토픽으로 수신하기 위해 std_msgs/msg/String 메시지 타입을 사용합니다.

GoogleTtsEngine

from google_tts_ros2.tts_engine import GoogleTtsEngine

GoogleTtsEngine은 실제 Google Cloud TTS API 호출과 오디오 재생을 담당하는 사용자 정의 클래스입니다.

현재 노드는 ROS 2 통신과 작업 관리에 집중하고, 실제 TTS 처리는 GoogleTtsEngine에 위임하는 구조입니다.

이러한 구조는 역할을 분리하기 때문에 유지보수에 유리합니다.

2) GoogleTtsNode 클래스

class GoogleTtsNode(Node):

GoogleTtsNode는 ROS 2의 Node 클래스를 상속합니다.

이 클래스는 다음 기능을 담당합니다.

ROS 2 파라미터 관리
문자열 토픽 구독
음성 출력 요청 대기열 관리
TTS 작업 스레드 관리
예외 처리
노드 종료 처리

3) 노드 초기화

def __init__(self) -> None:
    super().__init__("google_tts_node")

부모 클래스인 Node의 생성자를 호출하면서 노드 이름을 google_tts_node로 지정합니다.

4) topic_name 파라미터

self.declare_parameter(
    "topic_name",
    "/tts",
)

음성 출력 문자열을 수신할 ROS 2 토픽 이름입니다.

기본값은 다음과 같습니다.

/tts

따라서 기본 설정에서는 다음 명령으로 음성 출력을 요청할 수 있습니다.

ros2 topic pub --once /tts std_msgs/msg/String "{data: '안녕하세요. 로봇 시스템이 시작되었습니다.'}"

5) language_code 파라미터

self.declare_parameter(
    "language_code",
    "ko-KR",
)

Google Cloud TTS에서 사용할 언어 코드를 지정합니다.

기본값은 한국어를 의미하는 ko-KR입니다.

영어 음성을 사용할 경우 다음과 같은 값을 사용할 수 있습니다.

en-US

일본어 음성은 다음과 같이 지정할 수 있습니다.

ja-JP

언어 코드와 음성 이름은 서로 호환되어야 합니다.

예를 들어 언어 코드는 ko-KR인데 음성 이름을 영어 음성으로 지정하면 API 오류가 발생할 수 있습니다.

6) voice_name 파라미터

self.declare_parameter(
    "voice_name",
    "ko-KR-Standard-A",
)

Google Cloud TTS에서 사용할 음성 모델의 이름입니다.

기본값은 다음과 같습니다.

ko-KR-Standard-A

Google Cloud TTS는 언어에 따라 여러 음성을 제공합니다.

음성 모델에 따라 성별, 음색, 자연스러움, 비용 등이 달라질 수 있습니다.

사용 가능한 음성은 Google Cloud TTS의 음성 목록 또는 API를 통해 확인해야 합니다.

7) speaking_rate 파라미터

self.declare_parameter(
    "speaking_rate",
    1.0,
)

음성 재생 속도를 지정합니다.

기본값 1.0은 원래 속도를 의미합니다.

예를 들면 다음과 같습니다.

0.8은 기본보다 느린 속도
1.0은 기본 속도
1.2는 기본보다 빠른 속도
1.5는 빠른 안내 방송에 적합한 속도

속도를 지나치게 높이면 발음이 부자연스러워질 수 있습니다.

로봇 상태 안내에서는 보통 0.9에서 1.2 정도가 적절합니다.

8) pitch 파라미터

self.declare_parameter(
    "pitch",
    0.0,
)

합성 음성의 음높이를 조정합니다.

기본값 0.0은 원래 음높이를 사용한다는 의미입니다.

양수 값은 음높이를 높이고 음수 값은 음높이를 낮춥니다.

경고 안내에는 낮고 안정적인 음높이를 사용하고, 친근한 서비스 로봇에는 조금 높은 음높이를 적용할 수 있습니다.

9) volume_gain_db 파라미터

self.declare_parameter(
    "volume_gain_db",
    0.0,
)

합성된 음성의 볼륨 증폭 값을 데시벨 단위로 설정합니다.

기본값 0.0은 추가 증폭을 하지 않는다는 의미입니다.

양수 값을 사용하면 볼륨이 커지고 음수 값을 사용하면 볼륨이 작아집니다.

과도한 증폭은 오디오 왜곡이나 클리핑을 발생시킬 수 있으므로 주의해야 합니다.

10) audio_device 파라미터

self.declare_parameter(
    "audio_device",
    "default",
)

음성을 출력할 오디오 장치를 지정합니다.

기본값은 default입니다.

Linux에서 오디오 장치 목록은 다음 명령으로 확인할 수 있습니다.

aplay -L

USB 스피커나 특정 ALSA 장치를 사용할 경우 audio_device 값을 해당 장치 이름으로 변경할 수 있습니다.

11) speech_queue_size 파라미터

self.declare_parameter(
    "speech_queue_size",
    10,
)

음성 출력 대기열에 저장할 수 있는 최대 문자열 개수입니다.

기본값은 10입니다.

예를 들어 음성 재생 중 새로운 문자열 메시지가 여러 개 들어오면 즉시 재생하지 않고 대기열에 저장합니다.

대기열이 가득 찬 상태에서 새 메시지가 들어오면 새 메시지는 버려집니다.

이 제한은 메모리 사용량이 계속 증가하는 것을 방지합니다.

12) startup_text 파라미터

self.declare_parameter(
    "startup_text",
    "",
)

노드가 시작될 때 자동으로 출력할 음성입니다.

기본값은 빈 문자열이므로 아무 음성도 출력하지 않습니다.

다음과 같이 설정하면 노드 시작 시 안내 음성이 재생됩니다.

startup_text: "음성 안내 시스템이 시작되었습니다."

13) 토픽 이름 읽기

topic_name = str(
    self.get_parameter("topic_name").value
)

앞에서 선언한 topic_name 파라미터의 현재 값을 가져옵니다.

get_parameter()는 ROS 2 파라미터 객체를 반환하며, 실제 값은 .value 속성으로 읽습니다.

str()을 사용해 값을 명시적으로 문자열로 변환합니다.

14) 대기열 크기 검증

queue_size = max(
    1,
    int(
        self.get_parameter(
            "speech_queue_size"
        ).value
    ),
)

speech_queue_size 파라미터를 정수로 변환하고 최소값을 1로 제한합니다.

사용자가 실수로 다음과 같은 값을 입력하더라도 대기열 크기는 1이 됩니다.

또는

-5

queue.Queue의 maxsize에 잘못된 값이 들어가는 것을 방지하기 위한 방어적 코드입니다.

15) GoogleTtsEngine 객체 생성

self._engine = GoogleTtsEngine(
    language_code=str(
        self.get_parameter(
            "language_code"
        ).value
    ),
    voice_name=str(
        self.get_parameter(
            "voice_name"
        ).value
    ),
    speaking_rate=float(
        self.get_parameter(
            "speaking_rate"
        ).value
    ),
    pitch=float(
        self.get_parameter(
            "pitch"
        ).value
    ),
    volume_gain_db=float(
        self.get_parameter(
            "volume_gain_db"
        ).value
    ),
    audio_device=str(
        self.get_parameter(
            "audio_device"
        ).value
    ),
)

ROS 2 파라미터 값을 읽어 GoogleTtsEngine 객체에 전달합니다.

각 값은 예상되는 타입으로 명시적으로 변환합니다.

language_code는 문자열
voice_name은 문자열
speaking_rate는 실수
pitch는 실수
volume_gain_db는 실수
audio_device는 문자열

GoogleTtsEngine 내부에서는 일반적으로 다음 작업을 수행하게 됩니다.

Google Cloud TTS 클라이언트 생성
입력 문자열 설정
언어와 음성 모델 설정
음성 속도와 음높이 설정
API를 통한 음성 합성 요청
합성된 오디오 데이터 저장 또는 스트리밍
aplay 등을 이용한 음성 재생

현재 노드에서는 이러한 세부 구현을 알 필요 없이 다음 메서드만 호출합니다.

self._engine.speak(text)

16) 음성 대기열 생성

self._speech_queue: queue.Queue[
    Optional[str]
] = queue.Queue(
    maxsize=queue_size
)

음성 출력 요청을 저장하는 대기열입니다.

대기열에는 다음 두 종류의 데이터가 들어갑니다.

음성으로 변환할 문자열
작업 스레드 종료를 의미하는 None

예를 들면 대기열 상태는 다음과 같을 수 있습니다.

[
    "로봇이 출발합니다.",
    "장애물이 감지되었습니다.",
    "목적지에 도착했습니다."
]

작업 스레드는 첫 번째 문자열부터 순서대로 처리합니다.

queue.Queue는 기본적으로 FIFO 방식입니다.

FIFO는 먼저 들어온 데이터가 먼저 나가는 구조를 의미합니다.

17) 종료 이벤트 생성

self._stop_event = threading.Event()

작업 스레드에 종료 요청을 전달하기 위한 이벤트 객체입니다.

초기 상태에서는 이벤트가 설정되지 않습니다.

self._stop_event.is_set()

위 호출은 처음에는 False를 반환합니다.

노드가 종료될 때 다음 코드가 실행됩니다.

self._stop_event.set()

그 이후에는 is_set()이 True를 반환하므로 작업 스레드의 반복문이 종료됩니다.

18) 작업 스레드 생성

self._worker = threading.Thread(
    target=self._worker_loop,
    name="google_tts_worker",
    daemon=True,
)

음성 합성과 재생을 담당할 별도의 스레드를 생성합니다.

각 인자의 의미는 다음과 같습니다.

target=self._worker_loop작업 스레드가 실행할 함수입니다.
name="google_tts_worker"스레드 이름입니다.디버깅이나 로그 분석 시 스레드를 식별하기 쉽습니다.
daemon=True데몬 스레드로 설정합니다.메인 프로그램이 완전히 종료될 때 이 스레드가 프로세스 종료를 막지 않도록 합니다.

다만 데몬 스레드라고 해서 종료 처리를 생략해도 된다는 뜻은 아닙니다.

현재 코드는 destroy_node()에서 명시적으로 종료 요청을 보내고 스레드가 끝날 때까지 일정 시간 기다립니다.

19) 작업 스레드 시작

self._worker.start()

스레드를 실제로 실행합니다.

이 시점부터 _worker_loop() 메서드가 별도로 동작하며 대기열에 음성 출력 요청이 들어오기를 기다립니다.

20) ROS 2 구독자 생성

self._subscription = self.create_subscription(
    String,
    topic_name,
    self._topic_callback,
    10,
)

문자열 메시지를 수신하기 위한 ROS 2 구독자를 생성합니다.

각 인자의 의미는 다음과 같습니다.

String구독할 메시지 타입입니다.
topic_name구독할 토픽 이름입니다.
self._topic_callback메시지를 수신했을 때 호출할 함수입니다.
10ROS 2 구독자의 QoS 큐 깊이입니다.

여기서 구독자의 QoS 큐와 음성 출력 대기열은 서로 다른 구조입니다.

ROS 2 구독 큐는 아직 콜백에서 처리하지 못한 메시지를 임시 보관합니다.

self._speech_queue는 콜백 처리가 완료된 후 실제 음성 재생 순서를 관리합니다.

21) 노드 준비 완료 로그

self.get_logger().info(
    "Google Cloud TTS 준비 완료: "
    f"topic={topic_name}"
)

노드 초기화가 완료되면 구독 중인 토픽 이름을 로그로 출력합니다.

실행 시 다음과 비슷한 메시지가 표시됩니다.

[INFO] [google_tts_node]: Google Cloud TTS 준비 완료: topic=/tts

22) 시작 안내 음성 처리

startup_text = str(
    self.get_parameter(
        "startup_text"
    ).value
).strip()

startup_text 파라미터 값을 읽고 앞뒤 공백을 제거합니다.

이후 문자열이 비어 있지 않으면 대기열에 추가합니다.

if startup_text:
    self._enqueue(startup_text)

중요한 점은 시작 안내 음성도 직접 재생하지 않고 일반 토픽 메시지와 동일하게 대기열에 추가한다는 것입니다.

따라서 음성 재생 처리 흐름이 하나로 통일됩니다.

23) 토픽 콜백 함수

def _topic_callback(
    self,
    message: String,
) -> None:

/tts 토픽으로 std_msgs/msg/String 메시지가 들어오면 호출됩니다.

메시지의 실제 문자열은 message.data에 저장되어 있습니다.

text = message.data.strip()

strip()을 사용해 문자열 앞뒤 공백과 줄바꿈 문자를 제거합니다.

예를 들어 다음 문자열을 수신했다고 가정하겠습니다.

"   로봇이 출발합니다.   "

처리 후에는 다음 문자열이 됩니다.

"로봇이 출발합니다."

24) 빈 문자열 검사

if not text:
    self.get_logger().warning(
        "빈 문자열을 수신하여 무시했습니다."
    )
    return

문자열이 비어 있으면 Google Cloud TTS API를 호출하지 않습니다.

빈 문자열을 API에 전달하면 불필요한 요청이 발생하거나 API 오류가 발생할 수 있습니다.

따라서 경고 로그만 출력하고 콜백을 종료합니다.

다음과 같은 메시지는 무시됩니다.

ros2 topic pub --once /tts std_msgs/msg/String "{data: ''}"

공백만 포함된 문자열도 strip() 이후 빈 문자열이 되므로 무시됩니다.

25) 대기열에 문자열 추가

self._enqueue(text)

유효한 문자열은 _enqueue() 메서드로 전달됩니다.

콜백 함수에서는 음성 합성이나 재생을 직접 수행하지 않습니다.

이 구조 덕분에 콜백 함수는 빠르게 반환됩니다.

26) _enqueue 메서드

def _enqueue(
    self,
    text: str,
) -> None:

음성으로 출력할 문자열을 대기열에 추가하는 메서드입니다.

self._speech_queue.put_nowait(text)

put_nowait()는 대기열에 빈 공간이 있으면 즉시 데이터를 추가합니다.

대기열이 가득 찼더라도 기다리지 않고 바로 queue.Full 예외를 발생시킵니다.

이 방식은 ROS 2 콜백이 대기열 공간이 생길 때까지 멈추는 문제를 방지합니다.

27) 대기열 추가 성공 로그

self.get_logger().info(
    f"TTS 대기열 추가: {text}"
)

문자열이 정상적으로 대기열에 추가되면 로그를 출력합니다.

예시는 다음과 같습니다.

[INFO] [google_tts_node]: TTS 대기열 추가: 로봇이 출발합니다.

개발 단계에서는 유용하지만, 실제 서비스에서는 로그가 너무 많이 발생할 수 있습니다.

운영 환경에서는 info 대신 debug 레벨을 사용하는 것도 좋은 방법입니다.

28) 대기열이 가득 찬 경우

except queue.Full:
    self.get_logger().warning(
        "TTS 대기열이 가득 차서 "
        "새 문자열을 버렸습니다."
    )

음성 출력 속도보다 메시지 입력 속도가 빠르면 대기열이 가득 찰 수 있습니다.

예를 들어 대기열 크기가 10이고 10개의 문장이 재생을 기다리는 상태라면 11번째 문장은 버려집니다.

이 정책은 오래된 메시지를 유지하고 새로운 메시지를 버리는 방식입니다.

상태 안내 시스템에서는 이 정책이 항상 최선은 아닐 수 있습니다.

긴급 경고 메시지가 들어왔는데 대기열이 가득 차 있다면 긴급 메시지가 버려질 수 있기 때문입니다.

실제 로봇 시스템에서는 메시지 중요도에 따라 다음 정책을 고려할 수 있습니다.

가장 오래된 메시지를 제거하고 새 메시지 추가
긴급 메시지 전용 대기열 사용
우선순위 대기열 사용
현재 음성 재생 중단 후 긴급 음성 재생
같은 내용의 중복 메시지 제거

29) 작업 스레드 반복문

def _worker_loop(self) -> None:
    while not self._stop_event.is_set():

작업 스레드는 종료 이벤트가 설정되지 않은 동안 계속 실행됩니다.

이 스레드는 음성 출력 대기열을 감시하고 문자열이 들어오면 하나씩 처리합니다.

30) 대기열 데이터 가져오기

text = self._speech_queue.get(
    timeout=0.2
)

대기열에서 데이터를 하나 가져옵니다.

timeout=0.2는 최대 0.2초 동안 데이터가 들어오기를 기다린다는 의미입니다.

대기열에 데이터가 없으면 0.2초 후 queue.Empty 예외가 발생합니다.

무한정 대기하지 않고 타임아웃을 사용하는 이유는 종료 이벤트를 주기적으로 확인하기 위해서입니다.

만약 다음처럼 타임아웃 없이 호출한다면 스레드는 데이터가 들어올 때까지 계속 멈춰 있을 수 있습니다.

self._speech_queue.get()

현재 구조에서는 최대 0.2초마다 종료 여부를 확인할 수 있습니다.

31) 빈 대기열 처리

except queue.Empty:
    continue

0.2초 동안 데이터가 들어오지 않으면 반복문의 처음으로 돌아갑니다.

이후 _stop_event 상태를 다시 확인하고 종료 요청이 없으면 다시 대기합니다.

32) 종료 신호 처리

if text is None:
    self._speech_queue.task_done()
    break

대기열에서 가져온 값이 None이면 작업 스레드 종료 신호로 판단합니다.

None은 실제 음성 문자열로 사용되지 않으므로 종료용 센티널 값으로 사용할 수 있습니다.

센티널은 데이터 흐름에 특수한 값을 넣어 종료나 상태 변경을 전달하는 방식입니다.

task_done()을 호출한 후 반복문을 종료합니다.

33) 음성 합성 및 재생 시작 로그

self.get_logger().info(
    "음성 합성 및 재생 시작: "
    f"{text}"
)

실제 TTS 처리를 시작하기 전에 현재 처리 중인 문자열을 로그로 출력합니다.

이를 통해 대기열에 추가된 시점과 실제 재생이 시작된 시점을 구분할 수 있습니다.

34) 실제 음성 출력

self._engine.speak(text)

현재 노드의 핵심 실행 코드입니다.

GoogleTtsEngine의 speak() 메서드가 다음 작업을 수행한다고 볼 수 있습니다.

문자열을 Google Cloud TTS API에 전달
음성 데이터 생성
생성된 오디오 데이터 수신
임시 파일 또는 메모리에 저장
지정된 ALSA 오디오 장치로 재생
재생 완료 후 반환

speak() 메서드가 음성 재생이 끝날 때까지 반환하지 않는 동기 방식이어도 문제가 없습니다.

별도의 작업 스레드에서 실행되기 때문에 ROS 2 토픽 콜백 처리는 계속 진행될 수 있습니다.

35) Google Cloud API 오류 처리

except (
    GoogleAPICallError,
    RetryError,
) as error:
    self.get_logger().error(
        "Google Cloud TTS API 오류: "
        f"{error}"
    )

Google Cloud TTS 요청 중 발생한 오류를 처리합니다.

대표적인 원인은 다음과 같습니다.

인터넷 연결 실패
Google Cloud 인증 실패
서비스 계정 키 오류
TTS API 비활성화
잘못된 음성 이름
잘못된 언어 코드
API 사용량 제한 초과
결제 계정 문제
요청 타임아웃
Google Cloud 서비스 일시 장애

예외를 처리하지 않으면 작업 스레드가 종료될 수 있습니다.

현재 코드는 오류 로그를 출력한 후 다음 대기열 항목을 계속 처리합니다.

36) aplay 실행 파일 오류 처리

except FileNotFoundError:
    self.get_logger().error(
        "aplay를 찾을 수 없습니다. "
        "alsa-utils를 설치하세요."
    )

GoogleTtsEngine이 Linux의 aplay 명령으로 음성을 재생하는 경우 aplay가 설치되어 있지 않으면 FileNotFoundError가 발생할 수 있습니다.

Ubuntu에서는 다음 명령으로 설치할 수 있습니다.

sudo apt update
sudo apt install alsa-utils

설치 후 다음 명령으로 정상 동작을 확인할 수 있습니다.

aplay --version

37) 기타 예외 처리

except Exception as error:
    self.get_logger().error(
        f"TTS 처리 실패: {error}"
    )

예상하지 못한 모든 예외를 마지막에 처리합니다.

발생 가능한 오류는 다음과 같습니다.

임시 오디오 파일 생성 실패
파일 접근 권한 오류
오디오 장치 열기 실패
잘못된 파라미터 값
내부 라이브러리 오류
오디오 재생 프로세스 오류

광범위한 Exception 처리는 작업 스레드가 예상하지 못한 오류로 완전히 종료되는 것을 방지합니다.

다만 실제 운영 환경에서는 예외 타입과 스택 정보를 함께 기록하도록 개선하는 것이 좋습니다.

38) task_done 호출

finally:
    self._speech_queue.task_done()

음성 출력이 성공하든 실패하든 대기열 작업이 끝났음을 알립니다.

task_done()은 queue.Queue 내부의 미완료 작업 개수를 감소시킵니다.

이 값은 다른 코드에서 다음 메서드를 사용할 때 필요합니다.

self._speech_queue.join()

현재 소스에서는 join()을 사용하지 않지만, 대기열의 올바른 사용 규칙을 지키기 위해 task_done()을 호출하고 있습니다.

finally 블록에 있기 때문에 예외가 발생해도 반드시 실행됩니다.

39) destroy_node 메서드 재정의

def destroy_node(self) -> bool:

ROS 2 노드가 종료될 때 작업 스레드와 대기열을 안전하게 정리하기 위해 부모 클래스의 destroy_node() 메서드를 재정의합니다.

단순히 ROS 2 노드만 삭제하면 작업 스레드가 실행 중인 상태로 남거나 음성 재생 프로세스가 비정상 종료될 수 있습니다.

a. 종료 이벤트 설정

self._stop_event.set()

작업 스레드에 종료 요청을 전달합니다.

이후 작업 스레드의 반복 조건은 거짓이 됩니다.

while not self._stop_event.is_set():

b. 대기열에 종료 신호 추가

try:
    self._speech_queue.put_nowait(None)

except queue.Full:
    pass

작업 스레드가 queue.get()에서 대기 중일 수 있으므로 대기열에 None을 추가해 즉시 깨우려고 시도합니다.

대기열에 공간이 있으면 None이 들어가고 작업 스레드는 이를 종료 신호로 처리합니다.

대기열이 가득 차 있다면 queue.Full 예외가 발생합니다.

이 경우 이미 _stop_event가 설정되어 있으므로 별도의 처리를 하지 않습니다.

다만 현재 구현에는 한 가지 주의점이 있습니다.

작업 스레드가 이미 대기열에서 문자열을 꺼내 음성을 재생 중이라면 _engine.speak()가 반환될 때까지 즉시 종료되지 않을 수 있습니다.

40) 작업 스레드 종료 대기

if self._worker.is_alive():
    self._worker.join(timeout=1.0)

작업 스레드가 아직 실행 중이면 최대 1초 동안 종료를 기다립니다.

join()을 사용하지 않으면 메인 스레드가 먼저 종료되면서 작업 스레드가 정리되지 않을 수 있습니다.

다만 음성 재생이 1초 이상 남아 있으면 작업 스레드는 완전히 종료되지 않은 상태에서 join() 타임아웃이 끝날 수 있습니다.

현재 작업 스레드는 데몬 스레드이므로 최종적으로 프로세스가 종료될 때 함께 종료됩니다.

더 엄격한 종료 처리가 필요하다면 GoogleTtsEngine에 현재 오디오 재생을 중단하는 기능을 추가하는 것이 좋습니다.

41) 부모 destroy_node 호출

return super().destroy_node()

작업 스레드 종료 처리를 마친 후 부모 클래스의 destroy_node()를 호출합니다.

이 과정에서 ROS 2 구독자, 타이머, 서비스, 노드 핸들 등의 자원이 정리됩니다.

42) main 함수

def main(args=None) -> None:

ROS 2 노드 실행 진입점입니다.

Python 패키지의 setup.py 또는 setup.cfg에서 콘솔 스크립트로 등록하면 ros2 run 명령으로 실행할 수 있습니다.

a. ROS 2 초기화

rclpy.init(args=args)

ROS 2 Python 클라이언트 라이브러리를 초기화합니다.

명령행으로 전달된 ROS 2 인자와 파라미터 재정의 옵션도 여기서 처리됩니다.

b. node 변수 초기화

node = None

노드 생성 중 예외가 발생하더라도 finally 블록에서 안전하게 상태를 확인할 수 있도록 초기값을 None으로 설정합니다.

이 코드가 없다면 노드 생성 전에 예외가 발생했을 때 node 변수를 참조하면서 또 다른 오류가 발생할 수 있습니다.

c. 노드 생성과 spin 실행

try:
    node = GoogleTtsNode()
    rclpy.spin(node)

GoogleTtsNode 객체를 생성한 후 rclpy.spin()으로 콜백 처리를 시작합니다.

rclpy.spin()은 노드가 종료될 때까지 실행되며 다음 이벤트를 처리합니다.

토픽 메시지 수신
서비스 요청
타이머 콜백
기타 ROS 2 이벤트

현재 노드는 주로 /tts 토픽 메시지를 처리합니다.

d. Ctrl+C 처리

except KeyboardInterrupt:
    pass

터미널에서 Ctrl+C를 누르면 KeyboardInterrupt 예외가 발생합니다.

이를 정상 종료 상황으로 처리하기 위해 별도의 오류 로그 없이 넘어갑니다.

e. 노드 정리

finally:
    if node is not None:
        node.destroy_node()

정상 종료, 키보드 인터럽트 또는 실행 중 예외가 발생한 경우에도 destroy_node()를 호출합니다.

이를 통해 작업 스레드 종료와 ROS 2 노드 자원 정리가 수행됩니다.

f. ROS 2 종료

if rclpy.ok():
    rclpy.shutdown()

ROS 2가 아직 활성 상태라면 rclpy.shutdown()을 호출합니다.

rclpy.ok()를 먼저 확인하므로 이미 종료된 상태에서 다시 종료 함수를 호출하는 문제를 방지합니다.

43) Python 직접 실행 처리

if __name__ == "__main__":
    main()

해당 Python 파일을 직접 실행했을 때 main() 함수를 호출합니다.

예를 들면 다음과 같습니다.

python3 google_tts_node.py

ROS 2 패키지에 등록되어 있다면 일반적으로 다음과 같이 실행합니다.

ros2 run google_tts_ros2 google_tts_node

11. 재생 작업을 별도 스레드로 처리하는 이유

Google Cloud API 호출과 음성 재생은 즉시 끝나는 작업이 아닙니다.

토픽 콜백 안에서 직접 다음 작업을 실행하면 문제가 생길 수 있습니다.

def callback(self, msg):
    self.engine.speak(msg.data)

이 방식에서는 음성이 끝날 때까지 ROS 2 콜백이 점유됩니다. 긴 문장을 읽는 동안 다른 콜백 처리가 지연될 수 있습니다.

이번 코드에서는 토픽 콜백이 문자열을 대기열에 넣고 즉시 반환합니다.

토픽 콜백
→ Queue에 문자열 추가
→ 즉시 반환

별도의 Worker Thread가 다음 작업을 처리합니다.

Queue에서 문자열 꺼내기
→ Google Cloud TTS 요청
→ 음성 재생
→ 다음 문자열 처리

따라서 여러 문자열이 연속으로 들어오더라도 순서대로 재생할 수 있습니다.

대기열이 무한히 증가하면 메모리와 API 사용량이 증가할 수 있으므로 최대 크기를 제한했습니다.

speech_queue_size = 10

대기열이 가득 차면 새로운 문자열은 버리고 경고 로그를 출력합니다.

12. Launch 파일 작성

launch/tts.launch.py 파일을 작성합니다.

from launch import LaunchDescription
from launch_ros.actions import Node


def generate_launch_description():
    return LaunchDescription([
        Node(
            package="google_tts_ros2",
            executable="tts_node",
            name="google_tts_node",
            output="screen",
            parameters=[
                {
                    "topic_name": "/tts",
                    "language_code": "ko-KR",
                    "voice_name": "ko-KR-Standard-A",
                    "speaking_rate": 1.0,
                    "pitch": 0.0,
                    "volume_gain_db": 0.0,
                    "audio_device": "default",
                    "speech_queue_size": 10,
                    "startup_text": "",
                }
            ],
        )
    ])

USB 오디오 장치를 직접 지정하려면 다음과 같이 변경합니다.

"audio_device": "plughw:CARD=Device,DEV=0",

노드가 시작될 때 안내 음성을 출력하려면 startup_text를 설정합니다.

"startup_text": "터틀봇 음성 안내 시스템을 시작합니다.",

13. package.xml 작성

package.xml 파일을 다음과 같이 작성합니다.

<?xml version="1.0"?>
<?xml-model
  href="http://download.ros.org/schema/package_format3.xsd"
  schematypens="http://www.w3.org/2001/XMLSchema"?>

<package format="3">
  <name>google_tts_ros2</name>
  <version>0.0.1</version>

  <description>
    Google Cloud Text-to-Speech node for ROS 2 Humble
  </description>

  <maintainer email="your_email@example.com">
    Your Name
  </maintainer>

  <license>Apache-2.0</license>

  <buildtool_depend>ament_python</buildtool_depend>

  <exec_depend>rclpy</exec_depend>
  <exec_depend>std_msgs</exec_depend>
  <exec_depend>launch</exec_depend>
  <exec_depend>launch_ros</exec_depend>

  <export>
    <build_type>ament_python</build_type>
  </export>
</package>

export 영역의 다음 설정을 빠뜨리지 않아야 합니다.

<export>
  <build_type>ament_python</build_type>
</export>

이 값은 해당 패키지가 ament_python 빌드 형식임을 ROS 2 도구에 알려 줍니다.

14. setup.py 작성

setup.py 파일을 작성합니다.

import os
from glob import glob

from setuptools import find_packages
from setuptools import setup


package_name = "google_tts_ros2"


setup(
    name=package_name,
    version="0.0.1",
    packages=find_packages(
        exclude=["test"]
    ),
    data_files=[
        (
            "share/ament_index/resource_index/packages",
            [
                "resource/" + package_name
            ],
        ),
        (
            "share/" + package_name,
            [
                "package.xml"
            ],
        ),
        (
            os.path.join(
                "share",
                package_name,
                "launch",
            ),
            glob("launch/*.launch.py"),
        ),
    ],
    install_requires=[
        "setuptools",
        "google-cloud-texttospeech",
    ],
    zip_safe=True,
    maintainer="Your Name",
    maintainer_email="your_email@example.com",
    description=(
        "Google Cloud Text-to-Speech "
        "node for ROS 2"
    ),
    license="Apache-2.0",
    entry_points={
        "console_scripts": [
            (
                "tts_node = "
                "google_tts_ros2.tts_node:main"
            ),
            (
                "tts_say = "
                "google_tts_ros2.say:main"
            ),
        ],
    },
)

console_scripts에는 두 개의 실행 파일을 등록했습니다.

tts_node : /tts 토픽 구독 노드
tts_say  : 문자열 즉시 재생 명령

15. 패키지 빌드

작업 공간으로 이동합니다.

cd ~/tutlebot3_ws

ROS 2 의존성을 확인합니다.

rosdep install --from-paths src --ignore-src --rosdistro humble -r -y

패키지를 빌드합니다.

colcon build --packages-select google_tts_ros2

환경을 적용합니다.

source /opt/ros/humble/setup.bash
source ~/turtlebot3_ws/install/setup.bash

패키지 실행 파일을 확인합니다.

ros2 pkg executables google_tts_ros2

다음과 같이 표시되어야 합니다.

google_tts_ros2 tts_node
google_tts_ros2 tts_say

16. 문자열을 직접 음성으로 출력하기

먼저 Google Cloud 인증 환경변수를 확인합니다.

echo $GOOGLE_APPLICATION_CREDENTIALS

스피커의 기본 재생 상태도 확인합니다.

aplay /usr/share/sounds/alsa/Front_Center.wav

문자열을 음성으로 출력합니다.

ros2 run google_tts_ros2 tts_say "안녕하세요. 휴로 입니다."

USB 오디오 장치를 직접 지정하려면 다음과 같이 실행합니다.

ros2 run google_tts_ros2 tts_say "USB 스피커 출력 테스트입니다." \
  --audio-device "plughw:CARD=UACDemo10,DEV=0"

남성 음성으로 변경하려면 다음과 같이 실행할 수 있습니다.

ros2 run google_tts_ros2 tts_say "남성 음성 출력 테스트입니다." \
  --voice-name "ko-KR-Standard-C"

음성 속도를 느리게 설정합니다.

ros2 run google_tts_ros2 tts_say "천천히 안내 음성을 출력합니다." \
  --speaking-rate 0.85

음성 높이를 낮춥니다.

ros2 run google_tts_ros2 tts_say "음성 높이 조정 테스트입니다." --pitch -2.0

17. ROS 2 토픽으로 문자열 전달하기

TTS 노드를 실행합니다.

ros2 run google_tts_ros2 tts_node

정상적으로 실행되면 다음과 같은 로그가 출력됩니다.

[INFO] [google_tts_node]:
Google Cloud TTS 준비 완료: topic=/tts

새로운 터미널을 열고 ROS 2 환경을 적용합니다.

source /opt/ros/humble/setup.bash
source ~/turtlebot3_ws/install/setup.bash

/tts 토픽으로 문자열을 발행합니다.

ros2 topic pub --once /tts \
  std_msgs/msg/String "{data: '안녕하세요. 토픽으로 전달된 문자열입니다.'}"

TTS 노드에서 다음과 같은 로그가 출력됩니다.

[INFO] [google_tts_node]:
TTS 대기열 추가: 안녕하세요. 토픽으로 전달된 문자열입니다.

[INFO] [google_tts_node]:
음성 합성 및 재생 시작: 안녕하세요. 토픽으로 전달된 문자열입니다.

스피커에서는 입력한 문자열이 한국어 음성으로 재생됩니다.

18. Launch 파일로 실행하기

다음 명령으로 실행합니다.

ros2 launch google_tts_ros2 tts.launch.py

토픽을 발행합니다.

ros2 topic pub --once /tts \
  std_msgs/msg/String  "{data: '터틀봇 음성 안내 시스템이 정상적으로 동작합니다.'}"

19. ROS 2 파라미터로 설정 변경하기

실행할 때 음성을 변경할 수 있습니다.

ros2 run google_tts_ros2 tts_node \
  --ros-args \
  -p voice_name:="ko-KR-Neural2-C" \
  -p speaking_rate:=0.95 \
  -p pitch:=-1.0 \
  -p volume_gain_db:=2.0

USB 오디오 장치를 지정합니다.

ros2 run google_tts_ros2 tts_node \
  --ros-args \
  -p audio_device:="plughw:CARD=UACDemo10,DEV=0"

토픽 이름을 변경합니다.

ros2 run google_tts_ros2 tts_node \
  --ros-args -p topic_name:="/robot/speech"

변경된 토픽으로 문자열을 발행합니다.

ros2 topic pub --once /robot/speech \
  std_msgs/msg/String "{data: '로봇 음성 토픽 테스트입니다.'}"

시작 음성을 지정합니다.

ros2 run google_tts_ros2 tts_node \
  --ros-args -p startup_text:="휴로 시스템을 시작합니다."

현재 파라미터를 확인합니다.

ros2 param list /google_tts_node

특정 파라미터 값을 확인합니다.

ros2 param get /google_tts_node voice_name

20. 사용할 수 있는 한국어 음성

Google Cloud Text-to-Speech는 한국어 ko-KR에 대해 Standard, WaveNet, Neural2, Chirp3-HD 계열 음성을 제공합니다. 사용 가능한 음성은 Google Cloud 서비스 업데이트에 따라 변경될 수 있습니다.

대표적인 한국어 음성은 다음과 같습니다.

ko-KR-Standard-A   여성
ko-KR-Standard-B   여성
ko-KR-Standard-C   남성
ko-KR-Standard-D   남성

ko-KR-Wavenet-A    여성
ko-KR-Wavenet-B    여성
ko-KR-Wavenet-C    남성
ko-KR-Wavenet-D    남성

ko-KR-Neural2-A    여성
ko-KR-Neural2-B    여성
ko-KR-Neural2-C    남성

기본 테스트에서는 다음 음성을 권장합니다.

ko-KR-Standard-A

자연스러운 안내 음성이 필요한 경우 다음과 같은 음성을 테스트할 수 있습니다.

ko-KR-Neural2-A
ko-KR-Neural2-C

음성 계열에 따라 비용이 다를 수 있으므로 실제 로봇에 적용하기 전에 Google Cloud의 현재 가격 정책과 사용량 제한을 확인해야 합니다.

21. 다른 ROS 2 노드에서 음성 출력 요청하기

다른 Python ROS 2 노드에서 /tts Publisher를 생성하면 간단하게 음성 출력을 요청할 수 있습니다.

from std_msgs.msg import String


self.tts_publisher = self.create_publisher(
    String,
    "/tts",
    10,
)

음성 안내를 출력하는 함수를 작성합니다.

def speak(self, text: str) -> None:
    message = String()
    message.data = text

    self.tts_publisher.publish(message)

    self.get_logger().info(
        f"TTS 요청: {text}"
    )

사용 예시는 다음과 같습니다.

self.speak("목적지에 도착했습니다.")

장애물을 감지했을 때 사용할 수 있습니다.

if obstacle_detected:
    self.speak(
        "전방에 장애물이 있습니다."
    )

배터리가 부족할 때 사용할 수 있습니다.

if battery_percentage < 20.0:
    self.speak(
        "배터리가 부족합니다. "
        "충전이 필요합니다."
    )

자율주행이 완료되었을 때 사용할 수 있습니다.

if navigation_completed:
    self.speak(
        "목적지 이동을 완료했습니다."
    )

22. TurtleBot3 Navigation2와 연동하는 예

Navigation2 액션 결과가 성공했을 때 다음과 같이 TTS 메시지를 발행할 수 있습니다.

from action_msgs.msg import GoalStatus
from std_msgs.msg import String


def navigation_result_callback(
    self,
    future,
) -> None:
    result = future.result()

    message = String()

    if (
        result.status
        == GoalStatus.STATUS_SUCCEEDED
    ):
        message.data = (
            "목적지에 도착했습니다."
        )

    elif (
        result.status
        == GoalStatus.STATUS_CANCELED
    ):
        message.data = (
            "이동 명령이 취소되었습니다."
        )

    else:
        message.data = (
            "목적지 이동에 실패했습니다."
        )

    self.tts_publisher.publish(message)

이 구조를 사용하면 Navigation2 노드가 Google Cloud API를 직접 알 필요가 없습니다.

Navigation2 노드는 /tts 토픽에 문자열만 발행하고, 음성 합성과 스피커 출력은 TTS 노드가 담당합니다.

이처럼 기능을 분리하면 다음과 같은 장점이 있습니다.

자율주행 코드와 음성 코드를 분리할 수 있음
음성 엔진을 다른 서비스로 교체하기 쉬움
여러 노드가 하나의 TTS 노드를 공유할 수 있음
음성 출력 순서를 중앙에서 관리할 수 있음
테스트할 때 ros2 topic pub 명령을 사용할 수 있음

23. 자주 발생하는 오류와 해결 방법

1) DefaultCredentialsError

오류 예시는 다음과 같습니다.

google.auth.exceptions.DefaultCredentialsError:
Could not automatically determine credentials

환경변수를 확인합니다.

echo $GOOGLE_APPLICATION_CREDENTIALS

파일 존재 여부를 확인합니다.

ls -l $GOOGLE_APPLICATION_CREDENTIALS

현재 터미널에 다시 적용합니다.

export GOOGLE_APPLICATION_CREDENTIALS="$HOME/.config/google/turtlebot3-tts.json"

ROS 2 노드를 sudo로 실행하면 일반 사용자 환경변수가 전달되지 않을 수 있습니다.

다음과 같이 실행하지 않는 것이 좋습니다.

sudo ros2 run google_tts_ros2 tts_node

일반 사용자 권한으로 실행합니다.

ros2 run google_tts_ros2 tts_node

2) 403 Permission Denied

오류 예시는 다음과 같습니다.

403 PermissionDenied

다음 항목을 확인합니다.

Cloud Text-to-Speech API가 활성화되어 있는지
결제 계정이 연결되어 있는지
JSON 키가 올바른 프로젝트에서 생성되었는지
서비스 계정이 삭제되거나 비활성화되지 않았는지
Service Usage Consumer 권한이 있는지
API 사용 한도를 초과하지 않았는지

3) aplay를 찾을 수 없음

오류 예시는 다음과 같습니다.

aplay를 찾을 수 없습니다

다음 패키지를 설치합니다.

sudo apt install -y alsa-utils

4) 소리가 출력되지 않음

오디오 장치를 확인합니다.

aplay -l
aplay -L

직접 재생 테스트를 합니다.

aplay -D plughw:1,0 \
  /usr/share/sounds/alsa/Front_Center.wav

ROS 2 노드에도 같은 장치를 설정합니다.

ros2 run google_tts_ros2 tts_node \
  --ros-args \
  -p audio_device:="plughw:1,0"

5) Device or resource busy

오류 예시는 다음과 같습니다.

audio open error:
Device or resource busy

다른 프로그램이 오디오 장치를 사용하고 있는지 확인합니다.

fuser -v /dev/snd/*

필요하다면 해당 프로세스를 종료합니다.

kill <PID>

여러 프로세스가 하나의 장치를 공유해야 한다면 hw 대신 default, plughw, dmix 장치를 사용합니다.

6) 음성이 너무 빠르거나 느림

속도를 조절합니다.

ros2 run google_tts_ros2 tts_node \
  --ros-args \
  -p speaking_rate:=0.9

대표적인 값은 다음과 같습니다.

0.8 : 느린 안내
1.0 : 기본 속도
1.2 : 빠른 안내

7) 토픽을 발행했지만 재생되지 않음

토픽 이름을 확인합니다.

ros2 topic list

토픽 형식을 확인합니다.

ros2 topic type /tts

정상 출력은 다음과 같습니다.

std_msgs/msg/String

TTS 노드가 실제로 구독하고 있는지 확인합니다.

ros2 topic info /tts

메시지가 발행되는지 확인합니다.

ros2 topic echo /tts

ROS_DOMAIN_ID도 확인합니다.

echo $ROS_DOMAIN_ID

Publisher와 Subscriber의 ROS_DOMAIN_ID가 다르면 서로 통신하지 못합니다.

8) Python 모듈을 찾을 수 없음

오류 예시는 다음과 같습니다.

ModuleNotFoundError:
No module named 'google.cloud'

현재 사용자의 Python 환경에 다시 설치합니다.

python3 -m pip install \
  --user \
  --upgrade \
  google-cloud-texttospeech

설치 위치를 확인합니다.

python3 -m pip show \
  google-cloud-texttospeech

24. 실제 로봇에 적용할 때 고려할 점

1) 인터넷 연결이 필요함

Google Cloud Text-to-Speech는 클라우드 API이므로 인터넷 연결이 필요합니다.

네트워크가 끊기면 새로운 문자열을 음성으로 합성할 수 없습니다.

다음과 같은 안전 관련 안내를 클라우드 TTS에만 의존하면 안 됩니다.

긴급 정지
모터 과전류
배터리 위험
화재 또는 충돌 경고

긴급 경고는 로컬 비프음, LED, 저장된 WAV 파일과 함께 사용하는 것이 안전합니다.

2) 자주 사용하는 문장은 캐시하는 것이 좋음

다음 문장은 로봇에서 반복적으로 사용될 가능성이 높습니다.

목적지에 도착했습니다.
배터리가 부족합니다.
충전을 시작합니다.
전방에 장애물이 있습니다.
작업을 완료했습니다.

이 문장들을 매번 API로 합성하면 네트워크 지연과 API 사용량이 증가합니다.

실제 운영에서는 한 번 생성한 음성 데이터를 파일로 저장하고, 같은 문장이 들어오면 저장된 파일을 재생하는 캐시 구조가 효율적입니다.

3) 긴급 안내와 일반 안내를 분리해야 함

현재 구현은 입력된 순서대로 모든 문자열을 재생합니다.

긴 문장을 재생하는 동안 긴급 메시지가 들어오면 기존 문장이 끝날 때까지 기다려야 합니다.

실제 서비스 로봇에서는 다음 기능을 추가할 수 있습니다.

메시지 우선순위
현재 음성 재생 중단
긴급 메시지 즉시 재생
중복 메시지 제거
일정 시간 내 동일 메시지 재생 제한

예를 들어 메시지를 다음과 같이 확장할 수 있습니다.

text
priority
interrupt
repeat_count

이 경우 std_msgs/msg/String 대신 사용자 정의 메시지를 만드는 것이 좋습니다.

4) 서비스 계정 키 관리

서비스 계정 JSON 파일은 로봇 저장장치에 평문으로 존재합니다.

다음 보안 대책이 필요합니다.

파일 권한을 600으로 제한
ROS 2 패키지와 별도 위치에 저장
Git 저장소에 추가하지 않기
로봇 분실 시 즉시 키 폐기
개발용과 운영용 서비스 계정 분리
주기적으로 키 교체
불필요한 IAM 권한 제거

5) 비용과 사용량 제한

Cloud Text-to-Speech는 음성 모델과 처리한 문자 수에 따라 비용이 발생할 수 있습니다.

무한 반복이나 센서 오류로 같은 메시지가 계속 발행되면 예상보다 많은 API 요청이 발생할 수 있습니다.

다음 제한을 추가하는 것이 좋습니다.

동일 문장 재생 간격 제한
분당 최대 요청 수 제한
최대 문자열 길이 제한
대기열 최대 크기 제한
Google Cloud 예산 알림 설정

현재 코드는 대기열 최대 크기를 제한하지만 동일 문자열의 반복은 제한하지 않습니다.

25. 최종 테스트 순서

전체 시스템은 다음 순서로 확인하면 됩니다.

오디오 장치를 확인합니다.

aplay -l

스피커를 테스트합니다.

aplay /usr/share/sounds/alsa/Front_Center.wav

Google Cloud 인증 정보를 확인합니다.

echo $GOOGLE_APPLICATION_CREDENTIALS

Python 라이브러리를 확인합니다.

python3 -c \
  "from google.cloud import texttospeech; print('OK')"

문자열 직접 재생을 테스트합니다.

ros2 run google_tts_ros2 tts_say \
  "직접 음성 출력 테스트입니다."

토픽 구독 노드를 실행합니다.

ros2 launch google_tts_ros2 tts.launch.py

토픽으로 문자열을 전달합니다.

ros2 topic pub \
  --once \
  /tts \
  std_msgs/msg/String \
  "{data: '토픽 음성 출력 테스트입니다.'}"

모든 단계가 정상이라면 TurtleBot3 Burger에서 Google Cloud Text-to-Speech를 이용한 ROS 2 음성 안내 시스템이 완성된 것입니다.