Update flare-ui/src/app/services/conversation-manager.service.ts

flare-ui/src/app/services/conversation-manager.service.ts

@@ -1,210 +1,290 @@
-import { Injectable } from '@angular/core';
-import { Subject, Observable, timer } from 'rxjs';
-import { retry, tap } from 'rxjs/operators';
+import { Injectable, OnDestroy } from '@angular/core';
+import { Subject, Subscription, BehaviorSubject } from 'rxjs';
+import { WebSocketService } from './websocket.service';
+import { AudioStreamService } from './audio-stream.service';
 
-export interface WebSocketMessage {
-  type: string;
-  [key: string]: any;
-}
+export type ConversationState = 
+  | 'idle' 
+  | 'listening' 
+  | 'processing_stt' 
+  | 'processing_llm' 
+  | 'processing_tts' 
+  | 'playing_audio';
 
-export interface TranscriptionResult {
+export interface ConversationMessage {
+  role: 'user' | 'assistant';
   text: string;
-  is_final: boolean;
-  confidence: number;
-}
-
-export interface StateChangeMessage {
-  from: string;
-  to: string;
+  timestamp: Date;
+  audioUrl?: string;
 }
 
 @Injectable({
   providedIn: 'root'
 })
-export class WebSocketService {
-  private socket: WebSocket | null = null;
-  private url: string = '';
-  private reconnectAttempts = 0;
-  private maxReconnectAttempts = 5;
-  private reconnectDelay = 1000;
+export class ConversationManagerService implements OnDestroy {
+  private subscriptions = new Subscription();
+  private audioQueue: string[] = [];
+  private isInterrupting = false;
   
-  // Subjects for different message types
-  private messageSubject = new Subject<WebSocketMessage>();
-  private transcriptionSubject = new Subject<TranscriptionResult>();
-  private stateChangeSubject = new Subject<StateChangeMessage>();
-  private errorSubject = new Subject<string>();
-  private connectionSubject = new Subject<boolean>();
+  // State management
+  private currentStateSubject = new BehaviorSubject<ConversationState>('idle');
+  public currentState$ = this.currentStateSubject.asObservable();
   
-  // Public observables
-  public message$ = this.messageSubject.asObservable();
+  // Message history
+  private messagesSubject = new BehaviorSubject<ConversationMessage[]>([]);
+  public messages$ = this.messagesSubject.asObservable();
+  
+  // Current transcription
+  private transcriptionSubject = new BehaviorSubject<string>('');
   public transcription$ = this.transcriptionSubject.asObservable();
-  public stateChange$ = this.stateChangeSubject.asObservable();
-  public error$ = this.errorSubject.asObservable();
-  public connection$ = this.connectionSubject.asObservable();
-
-  constructor() {}
-
-  connect(sessionId: string): Promise<void> {
-    return new Promise((resolve, reject) => {
-      try {
-        // Construct WebSocket URL
-        const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
-        const host = window.location.host;
-        this.url = `${protocol}//${host}/ws/conversation/${sessionId}`;
-        
-        console.log(`🔌 Connecting to WebSocket: ${this.url}`);
-        
-        this.socket = new WebSocket(this.url);
-        
-        this.socket.onopen = () => {
-          console.log('✅ WebSocket connected');
-          this.reconnectAttempts = 0;
-          this.connectionSubject.next(true);
-          
-          // Start keep-alive ping
-          this.startKeepAlive();
-          
-          resolve();
-        };
-        
-        this.socket.onmessage = (event) => {
-          try {
-            const message: WebSocketMessage = JSON.parse(event.data);
-            this.handleMessage(message);
-          } catch (error) {
-            console.error('Failed to parse WebSocket message:', error);
-          }
-        };
-        
-        this.socket.onerror = (error) => {
-          console.error('❌ WebSocket error:', error);
-          this.errorSubject.next('WebSocket bağlantı hatası');
-          reject(error);
-        };
-        
-        this.socket.onclose = () => {
-          console.log('🔌 WebSocket disconnected');
-          this.connectionSubject.next(false);
-          this.stopKeepAlive();
-          
-          // Attempt reconnection
-          if (this.reconnectAttempts < this.maxReconnectAttempts) {
-            this.attemptReconnect(sessionId);
-          }
-        };
-        
-      } catch (error) {
-        console.error('Failed to create WebSocket:', error);
-        reject(error);
-      }
-    });
-  }
+  
+  // Audio player reference
+  private audioPlayer: HTMLAudioElement | null = null;
 
-  disconnect(): void {
-    this.stopKeepAlive();
-    
-    if (this.socket) {
-      this.socket.close();
-      this.socket = null;
-    }
-    
-    this.connectionSubject.next(false);
+  constructor(
+    private wsService: WebSocketService,
+    private audioService: AudioStreamService
+  ) {}
+
+  ngOnDestroy(): void {
+    this.cleanup();
   }
 
-  send(message: WebSocketMessage): void {
-    if (this.socket && this.socket.readyState === WebSocket.OPEN) {
-      this.socket.send(JSON.stringify(message));
-    } else {
-      console.warn('WebSocket is not connected');
-      this.errorSubject.next('WebSocket bağlantısı yok');
+  async startConversation(sessionId: string): Promise<void> {
+    try {
+      // Connect WebSocket
+      await this.wsService.connect(sessionId);
+      
+      // Start audio recording
+      await this.audioService.startRecording();
+      
+      // Set up subscriptions
+      this.setupSubscriptions();
+      
+      // Send start signal
+      this.wsService.sendControl('start_session', {
+        language: 'tr-TR',
+        stt_engine: 'google',
+        enable_barge_in: true
+      });
+      
+      console.log('✅ Conversation started');
+      
+    } catch (error) {
+      console.error('Failed to start conversation:', error);
+      throw error;
     }
   }
 
-  sendAudioChunk(audioData: string): void {
-    this.send({
-      type: 'audio_chunk',
-      data: audioData,
-      timestamp: Date.now()
-    });
+  stopConversation(): void {
+    this.cleanup();
   }
 
-  sendControl(action: string, config?: any): void {
-    this.send({
-      type: 'control',
-      action: action,
-      config: config
-    });
+  private setupSubscriptions(): void {
+    // Audio chunks from microphone
+    this.subscriptions.add(
+      this.audioService.audioChunk$.subscribe(chunk => {
+        if (!this.isInterrupting) {
+          this.wsService.sendAudioChunk(chunk.data);
+        }
+      })
+    );
+    
+    // WebSocket messages
+    this.subscriptions.add(
+      this.wsService.message$.subscribe(message => {
+        this.handleMessage(message);
+      })
+    );
+    
+    // Transcription updates
+    this.subscriptions.add(
+      this.wsService.transcription$.subscribe(result => {
+        if (!result.is_final) {
+          this.transcriptionSubject.next(result.text);
+        }
+      })
+    );
+    
+    // State changes
+    this.subscriptions.add(
+      this.wsService.stateChange$.subscribe(change => {
+        this.currentStateSubject.next(change.to as ConversationState);
+        this.handleStateChange(change.from, change.to);
+      })
+    );
+    
+    // Errors
+    this.subscriptions.add(
+      this.wsService.error$.subscribe(error => {
+        console.error('WebSocket error:', error);
+        this.addSystemMessage(`Hata: ${error}`);
+      })
+    );
   }
 
-  private handleMessage(message: WebSocketMessage): void {
-    // Emit to general message stream
-    this.messageSubject.next(message);
-    
-    // Handle specific message types
+  private handleMessage(message: any): void {
     switch (message.type) {
       case 'transcription':
-        this.transcriptionSubject.next({
-          text: message.text,
-          is_final: message.is_final,
-          confidence: message.confidence
-        });
-        break;
-        
-      case 'state_change':
-        this.stateChangeSubject.next({
-          from: message['from'],
-          to: message['to'] 
-        });
+        if (message['is_final']) {
+          this.addMessage('user', message['text']);
+          this.transcriptionSubject.next('');
+        }
         break;
         
-      case 'error':
-        this.errorSubject.next(message['message']);
+      case 'assistant_response':
+        this.addMessage('assistant', message['text']);
         break;
         
       case 'tts_audio':
-        // Handle TTS audio chunks
-        this.messageSubject.next(message);
+        this.handleTTSAudio(message);
         break;
         
-      case 'assistant_response':
-        // Handle assistant text response
-        this.messageSubject.next(message);
+      case 'control':
+        if (message['action'] === 'stop_playback') {
+          this.stopAudioPlayback();
+        }
         break;
     }
   }
 
-  private attemptReconnect(sessionId: string): void {
-    this.reconnectAttempts++;
-    const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts - 1);
+  private handleStateChange(from: string, to: string): void {
+    console.log(`📊 State: ${from} → ${to}`);
+    
+    // Handle state-specific logic
+    if (to === 'listening') {
+      // Clear transcription when starting to listen
+      this.transcriptionSubject.next('');
+    } else if (to === 'playing_audio') {
+      // Start playing accumulated audio
+      this.playQueuedAudio();
+    }
+  }
+
+  private handleTTSAudio(message: any): void {
+    // Accumulate audio chunks
+    this.audioQueue.push(message['data']);
     
-    console.log(`🔄 Attempting reconnection ${this.reconnectAttempts}/${this.maxReconnectAttempts} in ${delay}ms`);
+    if (message['is_last']) {
+      // All chunks received, create audio blob
+      const audioData = this.audioQueue.join('');
+      const audioBlob = this.base64ToBlob(audioData, 'audio/mpeg');
+      const audioUrl = URL.createObjectURL(audioBlob);
+      
+      // Update last message with audio URL
+      const messages = this.messagesSubject.value;
+      if (messages.length > 0 && messages[messages.length - 1].role === 'assistant') {
+        messages[messages.length - 1].audioUrl = audioUrl;
+        this.messagesSubject.next([...messages]);
+      }
+      
+      // Clear queue
+      this.audioQueue = [];
+    }
+  }
+
+  private playQueuedAudio(): void {
+    const messages = this.messagesSubject.value;
+    const lastMessage = messages[messages.length - 1];
+    
+    if (lastMessage?.audioUrl && lastMessage.role === 'assistant') {
+      this.playAudio(lastMessage.audioUrl);
+    }
+  }
+
+  private playAudio(audioUrl: string): void {
+    if (!this.audioPlayer) {
+      this.audioPlayer = new Audio();
+    }
+    
+    this.audioPlayer.src = audioUrl;
+    this.audioPlayer.play().catch(error => {
+      console.error('Audio playback error:', error);
+    });
+    
+    this.audioPlayer.onended = () => {
+      // Notify that audio playback ended
+      this.wsService.sendControl('audio_ended');
+      this.currentStateSubject.next('idle');
+    };
+  }
+
+  private stopAudioPlayback(): void {
+    if (this.audioPlayer) {
+      this.audioPlayer.pause();
+      this.audioPlayer.currentTime = 0;
+    }
+  }
+
+  // Barge-in handling
+  performBargeIn(): void {
+    const currentState = this.currentStateSubject.value;
     
-    setTimeout(() => {
-      this.connect(sessionId).catch(error => {
-        console.error('Reconnection failed:', error);
+    if (currentState !== 'idle' && currentState !== 'listening') {
+      console.log('🛑 Performing barge-in');
+      this.isInterrupting = true;
+      
+      // Stop audio playback
+      this.stopAudioPlayback();
+      
+      // Notify server
+      this.wsService.sendControl('interrupt', {
+        at_state: currentState
       });
-    }, delay);
+      
+      // Reset interruption flag after a delay
+      setTimeout(() => {
+        this.isInterrupting = false;
+      }, 500);
+    }
   }
 
-  // Keep-alive mechanism
-  private keepAliveInterval: any;
-  
-  private startKeepAlive(): void {
-    this.keepAliveInterval = setInterval(() => {
-      if (this.socket && this.socket.readyState === WebSocket.OPEN) {
-        this.send({ type: 'ping' });
-      }
-    }, 30000); // Ping every 30 seconds
+  private addMessage(role: 'user' | 'assistant', text: string): void {
+    const messages = this.messagesSubject.value;
+    messages.push({
+      role,
+      text,
+      timestamp: new Date()
+    });
+    this.messagesSubject.next([...messages]);
   }
-  
-  private stopKeepAlive(): void {
-    if (this.keepAliveInterval) {
-      clearInterval(this.keepAliveInterval);
-      this.keepAliveInterval = null;
+
+  private addSystemMessage(text: string): void {
+    console.log(`📢 System: ${text}`);
+  }
+
+  private base64ToBlob(base64: string, mimeType: string): Blob {
+    const byteCharacters = atob(base64);
+    const byteNumbers = new Array(byteCharacters.length);
+    
+    for (let i = 0; i < byteCharacters.length; i++) {
+      byteNumbers[i] = byteCharacters.charCodeAt(i);
     }
+    
+    const byteArray = new Uint8Array(byteNumbers);
+    return new Blob([byteArray], { type: mimeType });
+  }
+
+  private cleanup(): void {
+    this.subscriptions.unsubscribe();
+    this.audioService.stopRecording();
+    this.wsService.disconnect();
+    this.stopAudioPlayback();
+    this.audioQueue = [];
+    this.currentStateSubject.next('idle');
+    console.log('🧹 Conversation cleaned up');
+  }
+
+  // Public methods for UI
+  getCurrentState(): ConversationState {
+    return this.currentStateSubject.value;
+  }
+
+  getMessages(): ConversationMessage[] {
+    return this.messagesSubject.value;
   }
 
-  isConnected(): boolean {
-    return this.socket !== null && this.socket.readyState === WebSocket.OPEN;
+  clearMessages(): void {
+    this.messagesSubject.next([]);
   }
 }