#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
MRCP 1.0 TTS 전용 실제 통신 클라이언트
실제 MRCP 서버와 통신하여 TTS(Text-to-Speech) 기능만 제공
"""
import socket
import time
import threading
from enum import Enum
from typing import Dict, List, Optional
from dataclasses import dataclass
class RequestState(Enum):
"""MRCP 요청 상태"""
IDLE = "IDLE"
IN_PROGRESS = "IN-PROGRESS"
COMPLETE = "COMPLETE"
FAILED = "FAILED"
class CompletionCause(Enum):
"""완료 원인 코드"""
NORMAL = "000 normal"
TTS_ERROR = "006 tts-error"
NO_INPUT_TIMEOUT = "001 no-input-timeout"
CONNECTION_ERROR = "999 connection-error"
@dataclass
class MRCPRequest:
"""MRCP 요청 구조"""
method: str
url: str
cseq: int
content_type: str
content_length: int
headers: Dict[str, str]
body: str
def __str__(self):
"""RTSP 형식의 요청 패킷 생성"""
lines = [f"{self.method} {self.url} RTSP/1.0"]
lines.append(f"CSeq: {self.cseq}")
lines.append(f"Content-Type: {self.content_type}")
lines.append(f"Content-Length: {self.content_length}")
for key, value in self.headers.items():
lines.append(f"{key}: {value}")
lines.append("") # 빈 줄
lines.append(self.body)
return "\r\n".join(lines)
@dataclass
class MRCPResponse:
"""MRCP 응답 구조"""
status_code: int
status_message: str
cseq: int
content_length: int
request_state: RequestState
completion_cause: Optional[CompletionCause] = None
headers: Optional[Dict[str, str]] = None
body: str = ""
def __str__(self):
"""RTSP 형식의 응답 패킷 생성"""
lines = [f"RTSP/1.0 {self.status_code} {self.status_message}"]
lines.append(f"CSeq: {self.cseq}")
lines.append(f"Content-Length: {self.content_length}")
lines.append(f"Request-State: {self.request_state.value}")
if self.completion_cause:
lines.append(f"Completion-Cause: {self.completion_cause.value}")
if self.headers:
for key, value in self.headers.items():
lines.append(f"{key}: {value}")
lines.append("") # 빈 줄
if self.body:
lines.append(self.body)
return "\r\n".join(lines)
@classmethod
def parse(cls, response_text: str) -> 'MRCPResponse':
"""RTSP 응답 텍스트를 파싱"""
lines = response_text.strip().split('\r\n')
if not lines:
raise ValueError("빈 응답")
# 첫 번째 줄: RTSP/1.0 200 OK
status_line = lines[0]
status_parts = status_line.split(' ', 2)
if len(status_parts) < 2:
raise ValueError(f"잘못된 상태 라인: {status_line}")
status_code = int(status_parts[1])
status_message = status_parts[2] if len(status_parts) > 2 else ""
# 헤더 파싱
headers = {}
body_start = 0
cseq = 0
content_length = 0
request_state = RequestState.IDLE
completion_cause = None
for i, line in enumerate(lines[1:], 1):
if line == '': # 빈 줄이면 헤더 끝
body_start = i + 1
break
if ':' in line:
key, value = line.split(':', 1)
key = key.strip()
value = value.strip()
headers[key] = value
# 중요 헤더들 추출
if key.lower() == 'cseq':
cseq = int(value)
elif key.lower() == 'content-length':
content_length = int(value)
elif key.lower() == 'request-state':
try:
request_state = RequestState(value)
except ValueError:
request_state = RequestState.IDLE
elif key.lower() == 'completion-cause':
try:
completion_cause = CompletionCause(value)
except ValueError:
completion_cause = None
# 바디 추출
body = '\r\n'.join(lines[body_start:]) if body_start < len(lines) else ""
return cls(
status_code=status_code,
status_message=status_message,
cseq=cseq,
content_length=content_length,
request_state=request_state,
completion_cause=completion_cause,
headers=headers,
body=body
)
class MRCPTTSClient:
"""MRCP 1.0 TTS 전용 실제 클라이언트"""
def __init__(self, server_host: str = "127.0.0.1", server_port: int = 554, timeout: int = 30):
self.server_host = server_host
self.server_port = server_port
self.timeout = timeout
self.session_id = f"session_{int(time.time())}"
self.cseq = 0
self.socket: Optional[socket.socket] = None
self.is_connected = False
def connect(self) -> bool:
"""실제 MRCP 서버에 연결"""
try:
print(f"[CLIENT] MRCP 서버 연결 시도: {self.server_host}:{self.server_port}")
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.settimeout(self.timeout)
self.socket.connect((self.server_host, self.server_port))
self.is_connected = True
print(f"[CLIENT] 서버 연결 성공! 세션 ID: {self.session_id}")
return True
except socket.timeout:
print(f"[CLIENT] 연결 타임아웃: {self.server_host}:{self.server_port}")
return False
except ConnectionRefusedError:
print(f"[CLIENT] 연결 거부됨: {self.server_host}:{self.server_port} (서버가 실행 중인지 확인)")
return False
except socket.gaierror:
print(f"[CLIENT] 호스트명 해석 실패: {self.server_host}")
return False
except Exception as e:
print(f"[CLIENT] 연결 실패: {e}")
return False
def disconnect(self):
"""서버 연결 해제"""
if self.socket:
try:
self.socket.close()
except:
pass
finally:
self.socket = None
self.is_connected = False
print(f"[CLIENT] 서버 연결 해제")
def _get_next_cseq(self) -> int:
"""다음 시퀀스 번호 생성"""
self.cseq += 1
return self.cseq
def _send_request(self, request: str) -> str:
"""RTSP 요청 전송 및 응답 수신"""
if not self.is_connected or not self.socket:
raise ConnectionError("서버에 연결되지 않았습니다.")
try:
# 요청 전송
request_bytes = request.encode('utf-8')
self.socket.sendall(request_bytes)
print(f"\n[CLIENT] TTS 요청 전송:")
print("=" * 50)
print(request)
print("=" * 50)
# 응답 수신
response_data = b""
while True:
try:
chunk = self.socket.recv(4096)
if not chunk:
break
response_data += chunk
# RTSP 응답 끝 확인
response_text = response_data.decode('utf-8', errors='ignore')
if '\r\n\r\n' in response_text:
# Content-Length 체크
if 'Content-Length:' in response_text:
try:
headers_end = response_text.find('\r\n\r\n')
headers_part = response_text[:headers_end]
content_length = 0
for line in headers_part.split('\r\n'):
if line.lower().startswith('content-length:'):
content_length = int(line.split(':', 1)[1].strip())
break
body_start = headers_end + 4
if len(response_text[body_start:]) >= content_length:
break
except (ValueError, IndexError):
# Content-Length 파싱 실패 시 응답 종료로 간주
break
else:
break
except socket.timeout:
print("[CLIENT] 응답 수신 타임아웃")
break
response_text = response_data.decode('utf-8', errors='ignore')
print(f"\n[SERVER] TTS 응답 수신:")
print("-" * 50)
print(response_text if response_text else "응답 없음")
print("-" * 50)
return response_text
except Exception as e:
raise ConnectionError(f"요청 전송 실패: {e}")
def speak(self, text: str, voice_gender: str = "female",
language: str = "ko-KR", voice_age: str = "25") -> List[MRCPResponse]:
"""TTS SPEAK 요청 - 실제 서버와 통신"""
if not self.is_connected:
raise ConnectionError("서버에 연결되지 않았습니다.")
responses = []
try:
# MRCP SPEAK 요청 생성
url = f"rtsp://{self.server_host}:{self.server_port}/{self.session_id}"
cseq = self._get_next_cseq()
# MRCP 메시지 바디 생성
mrcp_body = f"""SPEAK {cseq} SPEAK 100\r
Voice-Gender: {voice_gender}\r
Voice-Age: {voice_age}\r
Speech-Language: {language}\r
Content-Type: text/plain\r
Content-Length: {len(text.encode('utf-8'))}\r
\r
{text}"""
# RTSP ANNOUNCE 요청 생성
announce_request = f"""ANNOUNCE {url} RTSP/1.0\r
CSeq: {cseq}\r
Content-Type: application/mrcp\r
Content-Length: {len(mrcp_body.encode('utf-8'))}\r
\r
{mrcp_body}"""
# 요청 전송 및 응답 수신
response_text = self._send_request(announce_request)
if response_text:
# 응답 파싱
try:
response = MRCPResponse.parse(response_text)
responses.append(response)
print(f"\n[SERVER] TTS 응답 분석:")
print(f"상태 코드: {response.status_code}")
print(f"상태 메시지: {response.status_message}")
print(f"요청 상태: {response.request_state.value}")
if response.completion_cause:
print(f"완료 원인: {response.completion_cause.value}")
# 성공/실패 판정
if response.status_code == 200:
if response.request_state == RequestState.COMPLETE:
if response.completion_cause == CompletionCause.NORMAL:
print("✅ TTS 성공! 음성이 생성되었습니다.")
else:
print(f"❌ TTS 완료되었지만 오류: {response.completion_cause}")
elif response.request_state == RequestState.IN_PROGRESS:
print("🔄 TTS 처리 중...")
# 실제 환경에서는 여기서 완료 응답을 기다려야 함
else:
print(f"⚠️ TTS 상태: {response.request_state.value}")
else:
print(f"❌ TTS 요청 실패: {response.status_code} {response.status_message}")
except Exception as parse_error:
print(f"❌ 응답 파싱 실패: {parse_error}")
# 파싱 실패 시 오류 응답 생성
error_response = MRCPResponse(
status_code=500,
status_message="Parse Error",
cseq=cseq,
content_length=0,
request_state=RequestState.FAILED,
completion_cause=CompletionCause.TTS_ERROR
)
responses.append(error_response)
else:
print("❌ 서버로부터 응답을 받지 못했습니다.")
# 응답 없음 시 오류 응답 생성
no_response = MRCPResponse(
status_code=408,
status_message="No Response",
cseq=cseq,
content_length=0,
request_state=RequestState.FAILED,
completion_cause=CompletionCause.CONNECTION_ERROR
)
responses.append(no_response)
except Exception as e:
print(f"❌ TTS 요청 중 오류 발생: {e}")
# 예외 발생 시 오류 응답 생성
error_response = MRCPResponse(
status_code=500,
status_message="Internal Error",
cseq=self.cseq,
content_length=0,
request_state=RequestState.FAILED,
completion_cause=CompletionCause.CONNECTION_ERROR
)
responses.append(error_response)
return responses
def demo_real_tts_test():
"""실제 TTS 서버 테스트 데모"""
print("\n" + "=" * 60)
print("MRCP 1.0 실제 TTS 서버 테스트")
print("=" * 60)
# 서버 설정 - 실제 MRCP 서버 주소로 변경하세요
SERVER_CONFIGS = [
("127.0.0.1", 554), # 로컬 서버
("localhost", 554), # 로컬호스트
("127.0.0.1", 8000), # 사용자 정의 포트 (예시에서 언급된 포트)
("192.168.1.100", 554), # 로컬 네트워크 서버 (실제 IP로 변경)
]
success = False
# 여러 서버 설정 시도
for server_host, server_port in SERVER_CONFIGS:
print(f"\n[테스트] {server_host}:{server_port} 서버 테스트")
client = MRCPTTSClient(server_host, server_port, timeout=10)
try:
# 서버 연결
if not client.connect():
print(f"❌ {server_host}:{server_port} 연결 실패")
continue
success = True
# TTS 요청
responses = client.speak(
text="안녕하세요, MRCP 음성합성 실제 테스트입니다.",
voice_gender="female",
language="ko-KR"
)
# 결과 분석
if responses:
final_response = responses[-1]
if (final_response.status_code == 200 and
final_response.request_state in [RequestState.COMPLETE, RequestState.IN_PROGRESS]):
print(f"\n🎉 TTS 테스트 성공! 서버: {server_host}:{server_port}")
break
else:
print(f"\n⚠️ TTS 응답 받았지만 처리 실패")
else:
print(f"\n❌ TTS 응답 없음")
except Exception as e:
print(f"❌ 테스트 중 오류: {e}")
finally:
client.disconnect()
if not success:
print(f"\n💡 모든 서버 연결 실패. 다음을 확인해주세요:")
print(" 1. MRCP 서버가 실행 중인지 확인")
print(" 2. 서버 주소와 포트 번호 확인")
print(" 3. 방화벽 설정 확인")
print(" 4. 네트워크 연결 상태 확인")
def demo_multiple_tts_requests():
"""여러 TTS 요청 테스트"""
print("\n" + "=" * 60)
print("MRCP 1.0 다중 TTS 요청 테스트")
print("=" * 60)
# 실제 서버 주소 설정 (필요시 수정)
client = MRCPTTSClient("127.0.0.1", 554, timeout=15)
try:
if not client.connect():
print("❌ 서버 연결 실패")
return
# 여러 텍스트로 TTS 테스트
test_texts = [
"첫 번째 음성합성 테스트입니다.",
"두 번째 텍스트를 음성으로 변환합니다.",
"세 번째 테스트: 숫자 1, 2, 3, 4, 5",
]
for i, text in enumerate(test_texts, 1):
print(f"\n--- TTS 테스트 {i}/{len(test_texts)} ---")
responses = client.speak(text, voice_gender="female", language="ko-KR")
if responses:
response = responses[-1]
if response.status_code == 200:
print(f"✅ 테스트 {i} 성공")
else:
print(f"❌ 테스트 {i} 실패: {response.status_code}")
time.sleep(1) # 요청 간 간격
except Exception as e:
print(f"❌ 다중 요청 테스트 오류: {e}")
finally:
client.disconnect()
if __name__ == "__main__":
print("MRCP 1.0 TTS 전용 실제 통신 클라이언트")
print("실제 MRCP 서버와 통신하여 TTS(Text-to-Speech) 테스트를 수행합니다.")
print()
# 실제 서버 테스트
demo_real_tts_test()
# 성공한 경우 다중 요청 테스트도 수행
time.sleep(2)
# 필요시 다중 요청 테스트 (주석 해제)
# demo_multiple_tts_requests()