""" Gradio UI for Chloe's Voice Komentle Game Connects to FastAPI backend for voice analysis """ import os # Set Gradio temp directory BEFORE importing gradio _upload_dir = os.path.join(os.path.dirname(__file__), "gradio_uploads") os.makedirs(_upload_dir, exist_ok=True) os.environ["GRADIO_TEMP_DIR"] = _upload_dir import gradio as gr from datetime import datetime import uuid import asyncio from sqlalchemy import create_engine, text from dotenv import load_dotenv # Load environment variables load_dotenv() # Import backend functions from backend import ( analyze_voice_logic, get_puzzle_by_date, lifespan, app as backend_app, ) # Database connection DATABASE_URL = os.getenv("DATABASE_URL") engine = create_engine( DATABASE_URL, pool_size=10, # 기본 연결 풀 크기 max_overflow=20, # 최대 추가 연결 수 pool_pre_ping=True, # 연결 사용 전 유효성 검사 pool_recycle=3600, # 1시간마다 연결 재생성 connect_args={ "connect_timeout": 10, # 연결 타임아웃 10초 "options": "-c statement_timeout=30000" # 쿼리 타임아웃 30초 } ) # Session ID (persistent across attempts) session_id = str(uuid.uuid4()) # Backend initialization flag backend_initialized = False async def analyze_voice_async(audio_file, date_str): """ Analyze voice using backend logic directly Args: audio_file: Path to recorded audio file date_str: Date string for puzzle lookup Returns: tuple: (result_text, scores_text, hint_text, image_path) """ if audio_file is None: return "❌ 오디오를 먼저 녹음해주세요!", "", "", None try: # Read audio file with open(audio_file, "rb") as f: audio_bytes = f.read() # Call backend logic directly result = await analyze_voice_logic(audio_bytes, date_str, session_id) # Handle errors if result.get("status") == "error": return f"❌ {result.get('message', 'Unknown error')}", "", "", None # Parse response (already in 0-100 range from backend) category = result.get("category", "unknown") pitch = result.get("pitch", 0.0) rhythm = result.get("rhythm", 0.0) energy = result.get("energy", 0.0) pronunciation = result.get("pronunciation", 0.0) transcript = result.get("transcript", 0.0) overall = result.get("overall", 0.0) advice = result.get("advice", "") is_correct = result.get("is_correct", False) hints = {} # hints are embedded in advice now # Format result message if is_correct: result_msg = f"🎉 정답입니다! 전체 점수: {overall:.1f}/100" else: result_msg = f"📊 전체 점수: {overall:.1f}/100 - 다시 시도해보세요!" # Format scores scores_text = f""" ### 📊 점수 상세 **카테고리:** {category.upper()} - **발음 (Pronunciation):** {pronunciation:.1f}/100 - **음높이 (Pitch):** {pitch:.1f}/100 - **리듬 (Rhythm):** {rhythm:.1f}/100 - **에너지 (Energy):** {energy:.1f}/100 - **전사 (Transcript):** {transcript:.1f}/100 - **전체 (Overall):** {overall:.1f}/100 """ # Format hints hint_text = "" hint_image = None if hints and "answer" in hints: hint_type = hints.get("type", "hint") hint_items = hints.get("answer", []) if hint_type == "hint": hint_text = "💡 **힌트:**\n\n" else: hint_text = "🎯 **발음 조언:**\n\n" for item in hint_items: hint_text += f"{item.get('text', '')}\n\n" # Get image path if exists img_path = item.get("path", "") if img_path and os.path.exists(img_path): hint_image = img_path # Add advice if no hints if not hint_text and advice: hint_text = f"💬 **조언:**\n\n{advice}" return result_msg, scores_text, hint_text, hint_image except Exception as e: return f"❌ 오류 발생: {str(e)}", "", "", None def analyze_voice(audio_file, date_str): """Synchronous wrapper for async analyze_voice_async""" return asyncio.run(analyze_voice_async(audio_file, date_str)) def get_today_puzzle(): """Get today's puzzle information from database""" try: today = datetime.now().strftime("%Y-%m-%d") # Use backend function to get puzzle puzzle = get_puzzle_by_date(today) print(puzzle) if puzzle: return f""" ### 📅 오늘의 퍼즐 **날짜:** {puzzle.get('puzzle_date', 'N/A')} **퍼즐 번호:** #{puzzle.get('puzzle_number', 'N/A')} **카테고리:** {puzzle.get('category', 'N/A').upper()} **난이도:** {puzzle.get('difficulty', 'N/A')} 정답 단어를 발음해보세요! (최대 6회 시도) """ else: return "❌ 오늘의 퍼즐을 찾을 수 없습니다." except Exception as e: return f"❌ 퍼즐 정보를 가져올 수 없습니다: {str(e)}" def reset_session(): """Reset session for new game""" global session_id session_id = str(uuid.uuid4()) return "✅ 새 게임 시작! 오디오를 녹음해주세요.", "", "", None # Create Gradio Interface with gr.Blocks(title="Chloe's Voice Komentle") as demo: gr.Markdown("# 🎤 Chloe's Voice Komentle") # Puzzle info section with gr.Row(): puzzle_info = gr.Markdown(value=get_today_puzzle()) refresh_btn = gr.Button("🔄 퍼즐 정보 새로고침", size="sm") with gr.Row(): with gr.Column(scale=1): # Audio recording gr.Markdown("### 🎙️ 음성 녹음") audio_input = gr.Audio( sources=["microphone"], type="filepath", label="마이크로 녹음", format="wav", ) # Date input (auto-filled with today) date_input = gr.Textbox( label="날짜 (YYYY-MM-DD)", value=datetime.now().strftime("%Y-%m-%d"), interactive=True, ) # Submit button submit_btn = gr.Button("🎯 분석하기", variant="primary", size="lg") reset_btn = gr.Button("🔄 새 게임 시작", variant="secondary") with gr.Column(scale=1): # Results gr.Markdown("### 📊 결과") result_output = gr.Markdown(label="결과") scores_output = gr.Markdown(label="점수 상세") # Hints section with gr.Row(): with gr.Column(): hint_output = gr.Markdown(label="힌트 및 조언") with gr.Column(): hint_image = gr.Image(label="힌트 이미지", show_label=True) # Event handlers submit_btn.click( fn=analyze_voice, inputs=[audio_input, date_input], outputs=[result_output, scores_output, hint_output, hint_image], ) reset_btn.click( fn=reset_session, inputs=[], outputs=[result_output, scores_output, hint_output, hint_image], ) refresh_btn.click(fn=get_today_puzzle, inputs=[], outputs=[puzzle_info]) # Footer gr.Markdown("---\n**Powered by:** VoiceKit MCP + Gemini AI") # Launch configuration if __name__ == "__main__": print("🚀 Starting Chloe's Voice Komentle...") # Initialize backend (VoiceKit MCP session) print("⏳ Initializing VoiceKit MCP...") async def init_backend(): """Initialize backend resources""" async with lifespan(backend_app): print("✓ VoiceKit MCP initialized") # Keep the lifespan context active await asyncio.Event().wait() # Wait forever # Run backend initialization in background import threading def run_backend_init(): asyncio.run(init_backend()) backend_thread = threading.Thread(target=run_backend_init, daemon=True) backend_thread.start() # Wait a bit for initialization import time time.sleep(5) print("✓ Backend initialized") # Launch Gradio server_host = os.getenv("SERVER_HOST") frontend_port = int(os.getenv("FRONTEND_PORT")) demo.launch( server_name=server_host, # Listen on all interfaces server_port=frontend_port, # Default Gradio port share=False, # Set to True for public link show_error=True, allowed_paths=[os.path.join(os.path.dirname(__file__), "hints", "audio")], # Allow serving TTS audio hints )