#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
MRCP 1.0 통신 시뮬레이터
실제 네트워크 통신 없이 MRCP 프로토콜 동작을 시뮬레이션합니다.
"""
import time
import random
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"
NO_INPUT_TIMEOUT = "001 no-input-timeout"
RECOGNITION_TIMEOUT = "002 recognition-timeout"
GRAMMAR_LOAD_FAILURE = "003 grammar-load-failure"
GRAMMAR_COMPILATION_FAILURE = "004 grammar-compilation-failure"
RECOGNIZER_ERROR = "005 recognizer-error"
TTS_ERROR = "006 tts-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)
class MRCPClient:
"""MRCP 1.0 클라이언트 시뮬레이터"""
def __init__(self, server_host: str = "mrcp-server.com", server_port: int = 554):
self.server_host = server_host
self.server_port = server_port
self.session_id = f"session_{int(time.time())}"
self.cseq = 0
self.is_connected = False
def connect(self):
"""서버 연결 시뮬레이션"""
print(f"[CLIENT] 서버 연결 시도: {self.server_host}:{self.server_port}")
time.sleep(0.1) # 연결 지연 시뮬레이션
self.is_connected = True
print(f"[CLIENT] 서버 연결 성공!")
def disconnect(self):
"""서버 연결 해제"""
print(f"[CLIENT] 서버 연결 해제")
self.is_connected = False
def _get_next_cseq(self) -> int:
"""다음 시퀀스 번호 생성"""
self.cseq += 1
return self.cseq
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 Exception("서버에 연결되지 않았습니다.")
# SPEAK 요청 생성
url = f"rtsp://{self.server_host}:{self.server_port}/{self.session_id}"
cseq = self._get_next_cseq()
request = MRCPRequest(
method="SPEAK",
url=url,
cseq=cseq,
content_type="text/plain",
content_length=len(text.encode('utf-8')),
headers={
"Voice-Gender": voice_gender,
"Voice-Age": voice_age,
"Speech-Language": language
},
body=text
)
print(f"\n[CLIENT] TTS 요청 전송:")
print("=" * 50)
print(request)
print("=" * 50)
# 서버 응답 시뮬레이션
responses = self._simulate_tts_response(cseq)
return responses
def recognize(self, grammar: str, timeout: int = 10000,
no_input_timeout: int = 5000) -> List[MRCPResponse]:
"""ASR RECOGNIZE 요청 시뮬레이션"""
if not self.is_connected:
raise Exception("서버에 연결되지 않았습니다.")
url = f"rtsp://{self.server_host}:{self.server_port}/{self.session_id}"
cseq = self._get_next_cseq()
request = MRCPRequest(
method="RECOGNIZE",
url=url,
cseq=cseq,
content_type="text/grammar",
content_length=len(grammar.encode('utf-8')),
headers={
"Recognition-Timeout": str(timeout),
"No-Input-Timeout": str(no_input_timeout),
"Speech-Language": "ko-KR"
},
body=grammar
)
print(f"\n[CLIENT] ASR 요청 전송:")
print("=" * 50)
print(request)
print("=" * 50)
# 서버 응답 시뮬레이션
responses = self._simulate_asr_response(cseq)
return responses
def _simulate_tts_response(self, cseq: int) -> List[MRCPResponse]:
"""TTS 응답 시뮬레이션"""
responses = []
# 1. 즉시 IN-PROGRESS 응답
progress_response = MRCPResponse(
status_code=200,
status_message="OK",
cseq=cseq,
content_length=0,
request_state=RequestState.IN_PROGRESS
)
responses.append(progress_response)
print(f"\n[SERVER] TTS 진행 중 응답:")
print("-" * 30)
print(progress_response)
print("-" * 30)
# 2. 처리 시간 시뮬레이션 (1-3초)
processing_time = random.uniform(1, 3)
print(f"[SERVER] TTS 처리 중... ({processing_time:.1f}초)")
time.sleep(processing_time)
# 3. 완료 응답 (90% 성공률)
if random.random() < 0.9:
complete_response = MRCPResponse(
status_code=200,
status_message="OK",
cseq=cseq,
content_length=0,
request_state=RequestState.COMPLETE,
completion_cause=CompletionCause.NORMAL,
headers={
"Speech-Marker": f"mark1={int(processing_time * 1000)}",
"Content-Duration": f"{processing_time:.1f}"
}
)
else:
# 오류 시뮬레이션
complete_response = MRCPResponse(
status_code=500,
status_message="Internal Server Error",
cseq=cseq,
content_length=0,
request_state=RequestState.FAILED,
completion_cause=CompletionCause.TTS_ERROR
)
responses.append(complete_response)
print(f"\n[SERVER] TTS 완료 응답:")
print("-" * 30)
print(complete_response)
print("-" * 30)
return responses
def _simulate_asr_response(self, cseq: int) -> List[MRCPResponse]:
"""ASR 응답 시뮬레이션"""
responses = []
# 1. 즉시 IN-PROGRESS 응답
progress_response = MRCPResponse(
status_code=200,
status_message="OK",
cseq=cseq,
content_length=0,
request_state=RequestState.IN_PROGRESS
)
responses.append(progress_response)
print(f"\n[SERVER] ASR 진행 중 응답:")
print("-" * 30)
print(progress_response)
print("-" * 30)
# 2. 인식 시간 시뮬레이션 (2-5초)
recognition_time = random.uniform(2, 5)
print(f"[SERVER] 음성 인식 중... ({recognition_time:.1f}초)")
time.sleep(recognition_time)
# 3. 완료 응답 (80% 성공률)
if random.random() < 0.8:
# 성공 시 인식 결과 포함
recognition_results = ["안녕하세요", "감사합니다", "도움말"]
recognized_text = random.choice(recognition_results)
result_body = f"""<?xml version="1.0"?>
<result>
<interpretation grammar="session" confidence="0.85">
<instance>{recognized_text}</instance>
<input mode="speech">{recognized_text}</input>
</interpretation>
</result>"""
complete_response = MRCPResponse(
status_code=200,
status_message="OK",
cseq=cseq,
content_length=len(result_body.encode('utf-8')),
request_state=RequestState.COMPLETE,
completion_cause=CompletionCause.NORMAL,
headers={
"Content-Type": "application/xml"
},
body=result_body
)
else:
# 인식 실패 시뮬레이션
failure_causes = [CompletionCause.NO_INPUT_TIMEOUT,
CompletionCause.RECOGNITION_TIMEOUT,
CompletionCause.RECOGNIZER_ERROR]
cause = random.choice(failure_causes)
complete_response = MRCPResponse(
status_code=200,
status_message="OK",
cseq=cseq,
content_length=0,
request_state=RequestState.COMPLETE,
completion_cause=cause
)
responses.append(complete_response)
print(f"\n[SERVER] ASR 완료 응답:")
print("-" * 30)
print(complete_response)
print("-" * 30)
return responses
def demo_tts_test():
"""TTS 테스트 데모"""
print("\n" + "=" * 60)
print("MRCP 1.0 TTS 테스트 시뮬레이션")
print("=" * 60)
client = MRCPClient()
try:
# 서버 연결
client.connect()
# TTS 요청
responses = client.speak(
text="안녕하세요, MRCP 음성합성 테스트입니다.",
voice_gender="female",
language="ko-KR"
)
# 응답 확인
final_response = responses[-1]
if final_response.request_state == RequestState.COMPLETE:
if final_response.completion_cause == CompletionCause.NORMAL:
print(f"\n✅ TTS 성공! 음성이 생성되었습니다.")
else:
print(f"\n❌ TTS 실패: {final_response.completion_cause.value}")
finally:
client.disconnect()
def demo_asr_test():
"""ASR 테스트 데모"""
print("\n" + "=" * 60)
print("MRCP 1.0 ASR 테스트 시뮬레이션")
print("=" * 60)
client = MRCPClient()
try:
# 서버 연결
client.connect()
# 문법 정의
grammar = """#ABNF 1.0 UTF-8;
language ko-KR;
mode voice;
root $main;
$main = 안녕하세요 | 감사합니다 | 도움말;"""
# ASR 요청
responses = client.recognize(
grammar=grammar,
timeout=10000,
no_input_timeout=5000
)
# 응답 확인
final_response = responses[-1]
if final_response.request_state == RequestState.COMPLETE:
if final_response.completion_cause == CompletionCause.NORMAL:
print(f"\n✅ ASR 성공! 인식 결과:")
print(final_response.body)
else:
print(f"\n❌ ASR 실패: {final_response.completion_cause.value}")
finally:
client.disconnect()
def demo_concurrent_requests():
"""동시 요청 테스트 데모"""
print("\n" + "=" * 60)
print("MRCP 1.0 동시 요청 테스트 시뮬레이션")
print("=" * 60)
def run_tts_client(client_id: int):
client = MRCPClient()
client.session_id = f"session_tts_{client_id}"
try:
client.connect()
responses = client.speak(f"클라이언트 {client_id} 음성합성 테스트")
print(f"[클라이언트 {client_id}] TTS 완료")
finally:
client.disconnect()
def run_asr_client(client_id: int):
client = MRCPClient()
client.session_id = f"session_asr_{client_id}"
try:
client.connect()
grammar = f"#ABNF 1.0 UTF-8;\nlanguage ko-KR;\nmode voice;\nroot $main;\n$main = 클라이언트{client_id};"
responses = client.recognize(grammar)
print(f"[클라이언트 {client_id}] ASR 완료")
finally:
client.disconnect()
# 동시 실행
threads = []
for i in range(3):
tts_thread = threading.Thread(target=run_tts_client, args=(i,))
asr_thread = threading.Thread(target=run_asr_client, args=(i,))
threads.extend([tts_thread, asr_thread])
tts_thread.start()
asr_thread.start()
# 모든 스레드 완료 대기
for thread in threads:
thread.join()
print("\n모든 동시 요청 완료!")
if __name__ == "__main__":
print("MRCP 1.0 통신 시뮬레이터")
print("실제 네트워크 통신 없이 MRCP 프로토콜 동작을 시뮬레이션합니다.")
# 데모 실행
demo_tts_test()
time.sleep(1)
demo_asr_test()
time.sleep(1)
demo_concurrent_requests()