Update services/gemini_realtime_service.py

services/gemini_realtime_service.py

@@ -2,7 +2,7 @@ import asyncio
 import base64
 import json
 import os
-from typing import Callable, Optional
+from typing import Callable, Optional, AsyncGenerator
 import numpy as np
 from google import genai
 from google.genai.types import (
@@ -13,7 +13,7 @@ from google.genai.types import (
 )
 
 class GeminiRealtimeService:
-    """Dịch vụ Gemini Realtime API cho streaming chất lượng cao"""
+    """Dịch vụ Gemini Realtime API với audio streaming thực"""
     
     def __init__(self, api_key: str = None):
         self.api_key = api_key or os.getenv("GEMINI_API_KEY")
@@ -22,6 +22,8 @@ class GeminiRealtimeService:
         self.is_active = False
         self.callback = None
         self.voice_name = "Puck"
+        self.input_queue = asyncio.Queue()
+        self.output_queue = asyncio.Queue()
         
     async def initialize(self):
         """Khởi tạo client Gemini"""
@@ -34,8 +36,12 @@ class GeminiRealtimeService:
         )
         return True
     
+    def encode_audio(self, data: np.ndarray) -> str:
+        """Encode audio data to base64"""
+        return base64.b64encode(data.tobytes()).decode("UTF-8")
+    
     async def start_session(self, voice_name: str = "Puck", callback: Callable = None):
-        """Bắt đầu session Gemini Realtime"""
+        """Bắt đầu session Gemini Realtime với audio streaming"""
         try:
             if not self.client:
                 await self.initialize()
@@ -55,32 +61,31 @@ class GeminiRealtimeService:
                 ),
             )
             
-            # Kết nối session - SỬA LỖI Ở ĐÂY
-            self.session = self.client.aio.live.connect(
+            # Kết nối session
+            self.session = await self.client.aio.live.connect(
                 model="gemini-2.0-flash-exp", 
                 config=config
             )
             
-            # Sử dụng async with để quản lý session
-            async with self.session as session:
-                self.is_active = True
-                
-                if self.callback:
-                    await self.callback({
-                        'type': 'status',
-                        'message': f'✅ Đã kết nối Gemini - Giọng: {voice_name}',
-                        'status': 'connected'
-                    })
-                
-                print("✅ Gemini Realtime session started")
-                
-                # Xử lý realtime communication
-                await self._handle_realtime_session(session)
-                
+            self.is_active = True
+            
+            if self.callback:
+                await self.callback({
+                    'type': 'status',
+                    'message': f'✅ Đã kết nối Gemini Audio Streaming - Giọng: {voice_name}',
+                    'status': 'connected'
+                })
+            
+            print("✅ Gemini Realtime Audio session started")
+            
+            # Khởi động background tasks
+            asyncio.create_task(self._audio_sender())
+            asyncio.create_task(self._audio_receiver())
+            
             return True
             
         except Exception as e:
-            error_msg = f"❌ Lỗi khởi động Gemini Realtime: {e}"
+            error_msg = f"❌ Lỗi khởi động Gemini Audio: {e}"
             if self.callback:
                 await self.callback({
                     'type': 'error',
@@ -90,34 +95,65 @@ class GeminiRealtimeService:
             print(error_msg)
             return False
     
-    async def _handle_realtime_session(self, session):
-        """Xử lý session realtime"""
+    async def _audio_sender(self):
+        """Gửi audio data đến Gemini"""
+        try:
+            async with self.session as session:
+                async for audio_chunk in self._audio_stream_generator():
+                    await session.send(audio_chunk)
+        except Exception as e:
+            error_msg = f"❌ Lỗi gửi audio: {e}"
+            if self.callback:
+                await self.callback({
+                    'type': 'error',
+                    'message': error_msg,
+                    'status': 'error'
+                })
+    
+    async def _audio_stream_generator(self) -> AsyncGenerator[bytes, None]:
+        """Generator cho audio streaming"""
+        while self.is_active:
+            try:
+                audio_data = await asyncio.wait_for(self.input_queue.get(), timeout=1.0)
+                yield audio_data
+            except asyncio.TimeoutError:
+                continue
+            except Exception as e:
+                print(f"❌ Lỗi audio generator: {e}")
+                break
+    
+    async def _audio_receiver(self):
+        """Nhận audio response từ Gemini"""
         try:
-            async for response in session:
-                if hasattr(response, 'data') and response.data:
-                    # Audio response from Gemini
-                    audio_data = np.frombuffer(response.data, dtype=np.int16)
-                    
-                    if self.callback:
-                        await self.callback({
-                            'type': 'audio',
-                            'audio_data': audio_data,
-                            'sample_rate': 24000,
-                            'status': 'audio_streaming'
-                        })
+            async with self.session as session:
+                async for response in session:
+                    if hasattr(response, 'data') and response.data:
+                        # Audio response
+                        audio_data = np.frombuffer(response.data, dtype=np.int16)
                         
-                elif hasattr(response, 'text') and response.text:
-                    # Text response from Gemini
-                    if self.callback:
-                        await self.callback({
-                            'type': 'text',
-                            'content': response.text,
-                            'role': 'assistant',
-                            'status': 'text_response'
-                        })
+                        if self.callback:
+                            await self.callback({
+                                'type': 'audio',
+                                'audio_data': audio_data,
+                                'sample_rate': 24000,
+                                'status': 'audio_streaming'
+                            })
                         
+                        # Đưa vào output queue để phát ra loa
+                        self.output_queue.put_nowait((24000, audio_data))
+                        
+                    elif hasattr(response, 'text') and response.text:
+                        # Text response
+                        if self.callback:
+                            await self.callback({
+                                'type': 'text',
+                                'content': response.text,
+                                'role': 'assistant',
+                                'status': 'text_response'
+                            })
+                            
         except Exception as e:
-            error_msg = f"❌ Lỗi trong session: {e}"
+            error_msg = f"❌ Lỗi nhận audio: {e}"
             if self.callback:
                 await self.callback({
                     'type': 'error',
@@ -125,28 +161,51 @@ class GeminiRealtimeService:
                     'status': 'error'
                 })
     
-    async def send_audio(self, audio_data: bytes):
-        """Gửi audio data đến Gemini session"""
-        if not self.session or not self.is_active:
+    async def send_audio_chunk(self, audio_chunk: np.ndarray, sample_rate: int = 16000):
+        """Gửi audio chunk đến Gemini"""
+        if not self.is_active:
             return False
             
         try:
-            # Trong implementation thực tế, sẽ gửi qua session
-            # await self.session.send(audio_data)
-            print(f"📤 Sent audio data: {len(audio_data)} bytes")
+            # Resample nếu cần (Gemini cần 16kHz)
+            if sample_rate != 16000:
+                audio_chunk = self._resample_audio(audio_chunk, sample_rate, 16000)
+            
+            # Encode và gửi
+            audio_bytes = audio_chunk.tobytes()
+            await self.input_queue.put(audio_bytes)
             return True
             
         except Exception as e:
-            print(f"❌ Lỗi gửi audio: {e}")
+            print(f"❌ Lỗi gửi audio chunk: {e}")
             return False
     
+    async def receive_audio(self) -> tuple[int, np.ndarray] | None:
+        """Nhận audio từ Gemini"""
+        try:
+            return await asyncio.wait_for(self.output_queue.get(), timeout=1.0)
+        except asyncio.TimeoutError:
+            return None
+    
+    def _resample_audio(self, audio_chunk: np.ndarray, original_rate: int, target_rate: int):
+        """Resample audio chunk (đơn giản)"""
+        if original_rate == target_rate:
+            return audio_chunk
+            
+        ratio = target_rate / original_rate
+        new_length = int(len(audio_chunk) * ratio)
+        return np.interp(
+            np.linspace(0, len(audio_chunk) - 1, new_length),
+            np.arange(len(audio_chunk)),
+            audio_chunk
+        ).astype(np.int16)
+    
     async def send_text(self, text: str):
-        """Gửi text message đến Gemini (fallback)"""
-        if not self.client or not self.is_active:
+        """Gửi text message (fallback)"""
+        if not self.client:
             return None
             
         try:
-            # Fallback: sử dụng chat completion thông thường
             response = await self.client.aio.models.generate_content(
                 model="gemini-2.0-flash-exp",
                 contents=text
@@ -173,14 +232,14 @@ class GeminiRealtimeService:
             return None
     
     async def close(self):
-        """Đóng kết nối Gemini"""
+        """Đóng kết nối"""
         self.is_active = False
         
         if self.callback:
             await self.callback({
                 'type': 'status',
-                'message': '🛑 Đã đóng kết nối Gemini',
+                'message': '🛑 Đã đóng kết nối Gemini Audio',
                 'status': 'disconnected'
             })
             
-        print("🛑 Gemini Realtime session closed")
+        print("🛑 Gemini Audio session closed")