import asyncio import base64 import json import time from uuid import uuid4 from typing import Optional from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile from fastapi.responses import FileResponse from sqlalchemy.orm import Session from app.database.database import get_db from app.database.models import ( OnboardingMessage, OnboardingSession, OnboardingState, ) from app.schemas.onboarding import ( CategoryContext, CompleteOnboardingRequest, CompleteOnboardingResponse, ) from app.schemas.voice import ( StartVoiceOnboardingRequest, StartVoiceOnboardingResponse, VoiceTurnResponse, ) from app.services.ai_onboarding_service import AIOnboardingService from app.services.completion_service import CompletionService from app.services.onboarding_service import OnboardingService from app.services.patient_submission_service import PatientSubmissionService from app.services.result_extraction_service import ResultExtractionService from app.services.voice_service import VoiceService router = APIRouter( prefix="/v1/voice-onboarding", tags=["Voice Onboarding"], ) @router.post( "/start", response_model=StartVoiceOnboardingResponse, ) async def start_voice_onboarding( request: StartVoiceOnboardingRequest, db: Session = Depends(get_db), ): start_time = time.time() try: onboarding_service = OnboardingService() voice_service = VoiceService() session_id = f"VOICE-{uuid4().hex[:8].upper()}" # Default category to General Medicine / Auto-Triage if omitted by client category = request.category or CategoryContext( type="general", code="general_medicine", name="General Medicine / Primary Care (Auto-Triage)", ) session = OnboardingSession( session_id=session_id, patient_id=request.patient.id, patient_name=request.patient.name, age=request.patient.age, gender=request.patient.gender, category_type=category.type, category_code=category.code, category_name=category.name, language=request.language, status="in_progress", stage="greeting", ) db.add(session) db.commit() db.refresh(session) # Initialize empty state onboarding_service.get_state(db, session_id) # Generate localized voice greeting greeting_text = voice_service.get_initial_greeting( request.patient.name, request.language, ) # Synthesize audio greeting in background for sub-50ms response audio_id = f"aud_{uuid4().hex[:12]}" asyncio.create_task( voice_service.synthesize_to_file(audio_id, greeting_text, request.language) ) # Save initial assistant message onboarding_service.save_message( db=db, session_id=session_id, role="assistant", message=greeting_text, stage="greeting", ) elapsed = time.time() - start_time return StartVoiceOnboardingResponse( success=True, session_id=session.session_id, patient_id=session.patient_id, category=category, language=session.language, status=session.status, stage=session.stage, greeting_text=greeting_text, audio_url=f"/v1/voice-onboarding/audio/{audio_id}", response_time_ms=round(elapsed * 1000, 2), response_time_sec=round(elapsed, 3), ) except Exception as err: import traceback print(f"[StartVoiceOnboarding Error]: {err}") traceback.print_exc() raise HTTPException(status_code=500, detail=str(err)) @router.post( "/{session_id}/turn", response_model=VoiceTurnResponse, ) async def voice_turn( session_id: str, audio: Optional[UploadFile] = File(None), text_fallback: Optional[str] = Form(None), db: Session = Depends(get_db), ): start_time = time.time() onboarding_service = OnboardingService() voice_service = VoiceService() ai_service = AIOnboardingService() # 1. Fetch active session session = onboarding_service.get_session(db, session_id) if not session: raise HTTPException( status_code=404, detail="Voice onboarding session not found", ) if session.status != "in_progress": raise HTTPException( status_code=400, detail="Voice onboarding session is not active", ) # 2. Fast-path transcript: use real-time Web Speech text_fallback first for zero-latency processing patient_transcript = "" if text_fallback and text_fallback.strip(): from app.services.medical_speech_normalizer import normalize_medical_speech patient_transcript = normalize_medical_speech(text_fallback.strip(), language_code=session.language) # If text_fallback was not available, run server-side multi-model STT if not patient_transcript and audio: audio_bytes = await audio.read() if len(audio_bytes) > 0: patient_transcript = await voice_service.speech_to_text( audio_bytes=audio_bytes, filename=audio.filename or "recording.webm", language_code=session.language, specialty_code=session.category_code or "general", patient_context={ "name": session.patient_name, "age": session.age, "gender": session.gender, }, ) # If STILL empty, user spoke nothing or silence was recorded. if not patient_transcript or not patient_transcript.strip(): lang = session.language.lower() if lang.startswith("ta"): no_speech_msg = "உங்க குரல் கேட்கலைங்க, தயவுசெய்து மைக்கில பேசுங்க. உங்க உடம்பு பிரச்சனையை சொல்லுங்க." elif lang.startswith("hi"): no_speech_msg = "आपकी आवाज़ सुनाई नहीं दी। कृपया अपने माइक में बोलें और अपनी समस्या बताएं।" elif lang.startswith("te"): no_speech_msg = "మీ స్వరం వినబడలేదు. దయచేసి మైక్‌లో మాట్లాడి మీ ఆరోగ్య సమస్యను చెప్పండి." elif lang.startswith("kn"): no_speech_msg = "ನಿಮ್ಮ ಧ್ವನಿ ಕೇಳಿಸಲಿಲ್ಲ. ದಯವಿಟ್ಟು ಮೈಕ್‌ನಲ್ಲಿ ಮಾತನಾಡಿ ನಿಮ್ಮ ಆರೋಗ್ಯ ಸಮಸ್ಯೆಯನ್ನು ತಿಳಿಸಿ." else: no_speech_msg = "I didn't hear anything. Please speak into your microphone and describe your health symptoms." audio_id, _ = await voice_service.text_to_speech( text=no_speech_msg, language_code=session.language, ) elapsed = time.time() - start_time return VoiceTurnResponse( success=True, session_id=session_id, patient_transcript="", ai_text=no_speech_msg, stage=session.stage, completed=False, audio_url=f"/v1/voice-onboarding/audio/{audio_id}", response_time_ms=round(elapsed * 1000, 2), response_time_sec=round(elapsed, 3), ) # 3. Save patient message onboarding_service.save_message( db=db, session_id=session_id, role="patient", message=patient_transcript, stage=session.stage, ) # 4. Fetch state & history state = onboarding_service.get_state(db, session_id) messages = ( db.query(OnboardingMessage) .filter(OnboardingMessage.session_id == session_id) .order_by(OnboardingMessage.created_at.asc()) .all() ) conversation_history = [ { "role": m.role, "message": m.message, "stage": m.stage, } for m in messages ] try: # 5. AI Reasoning & Extraction ai_response = await ai_service.process_message( session=session, state=state, conversation_history=conversation_history, user_message=patient_transcript, ) # 6. Update Structured Clinical State current_state = json.loads(state.state_json) updates = ai_response.get("state_updates", {}) for k, v in updates.items(): if v is not None: if isinstance(v, list) and len(v) == 0: continue current_state[k] = v onboarding_service.update_state( db=db, state=state, data=current_state, ) # 7. Update session stage, status & auto-assigned specialty ai_reply_text = ai_response["reply"] session.stage = ai_response.get("next_stage", session.stage) is_completed = bool(ai_response.get("completed", False)) sug_spec = ai_response.get("suggested_specialty") if sug_spec and sug_spec.get("code") and sug_spec["code"] != "general_medicine": session.category_code = sug_spec["code"] session.category_name = sug_spec["name"] if is_completed: session.status = "completed" db.commit() # 8. Save Assistant Message onboarding_service.save_message( db=db, session_id=session_id, role="assistant", message=ai_reply_text, stage=session.stage, ) # 9. Synthesize AI Speech Audio (TTS) asynchronously in background for sub-50ms response audio_id = f"aud_{uuid4().hex[:12]}" asyncio.create_task( voice_service.synthesize_to_file(audio_id, ai_reply_text, session.language) ) elapsed = time.time() - start_time return VoiceTurnResponse( success=True, session_id=session_id, patient_transcript=patient_transcript, ai_text=ai_reply_text, stage=session.stage, completed=is_completed, audio_url=f"/v1/voice-onboarding/audio/{audio_id}", audio_base64=None, suggested_specialty=sug_spec, response_time_ms=round(elapsed * 1000, 2), response_time_sec=round(elapsed, 3), ) except Exception as err: import traceback print(f"[VoiceTurn Error]: {err}") traceback.print_exc() raise HTTPException(status_code=500, detail=str(err)) @router.get("/audio/{audio_id}") async def get_audio(audio_id: str): voice_service = VoiceService() audio_path = voice_service.get_audio_path(audio_id) # Wait up to 6.0s for background TTS synthesis to complete for _ in range(60): if audio_path and audio_path.exists(): if audio_path.stat().st_size > 1000: break await asyncio.sleep(0.1) audio_path = voice_service.get_audio_path(audio_id) if not audio_path or not audio_path.exists(): from app.services.voice_service import AUDIO_DIR audio_data = voice_service._generate_fallback_wav() file_path = AUDIO_DIR / f"{audio_id}.wav" file_path.write_bytes(audio_data) audio_path = file_path media_type = "audio/mpeg" if audio_path.suffix == ".mp3" else "audio/wav" return FileResponse( path=str(audio_path), media_type=media_type, filename=audio_path.name, headers={ "Content-Disposition": "inline", "Accept-Ranges": "bytes", "Cache-Control": "no-cache, no-store, must-revalidate", "Access-Control-Allow-Origin": "*", }, ) @router.post( "/{session_id}/complete", response_model=CompleteOnboardingResponse, ) async def complete_voice_onboarding( session_id: str, request: CompleteOnboardingRequest, db: Session = Depends(get_db), ): start_time = time.time() onboarding_service = OnboardingService() completion_service = CompletionService() result_service = ResultExtractionService() submission_service = PatientSubmissionService() session = onboarding_service.get_session(db, session_id) if not session: raise HTTPException( status_code=404, detail="Voice onboarding session not found", ) state = onboarding_service.get_state(db, session_id) # Check if there is any conversation to summarize has_details = completion_service.is_complete(session, state) messages = ( db.query(OnboardingMessage) .filter(OnboardingMessage.session_id == session_id) .order_by(OnboardingMessage.created_at.asc()) .all() ) conversation_history = [ {"role": m.role, "message": m.message, "stage": m.stage} for m in messages ] result = await result_service.extract_result( session=session, state=state, conversation_history=conversation_history, ) elapsed = time.time() - start_time resp_ms = round(elapsed * 1000, 2) resp_sec = round(elapsed, 3) if not request.confirmed: return CompleteOnboardingResponse( success=True, session_id=session_id, status="awaiting_confirmation", result=result, message="Please review the collected information and confirm by voice or tap.", response_time_ms=resp_ms, response_time_sec=resp_sec, ) submission = await submission_service.submit(result) if submission.get("success"): db.query(OnboardingMessage).filter(OnboardingMessage.session_id == session_id).delete() db.query(OnboardingState).filter(OnboardingState.session_id == session_id).delete() db.delete(session) db.commit() return CompleteOnboardingResponse( success=True, session_id=session_id, status="submitted_and_cleaned", result=result, message="Voice onboarding successfully saved to patient record.", response_time_ms=resp_ms, response_time_sec=resp_sec, ) session.status = "submission_pending_retry" db.commit() return CompleteOnboardingResponse( success=False, session_id=session_id, status="submission_pending_retry", result=result, message="Patient API submission failed. Session preserved for retry.", response_time_ms=resp_ms, response_time_sec=resp_sec, )