import traceback from datetime import datetime from pathlib import Path import os import random import string import tempfile import re import io import PyPDF2 import docx from reportlab.pdfgen import canvas from reportlab.lib.pagesizes import letter from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle from reportlab.lib.enums import TA_JUSTIFY from ai_config import n_of_questions, load_model, openai_api_key, convert_text_to_speech from knowledge_retrieval import setup_knowledge_retrieval, generate_report # Initialize settings n_of_questions = n_of_questions() current_datetime = datetime.now() human_readable_datetime = current_datetime.strftime("%B %d, %Y at %H:%M") current_date = current_datetime.strftime("%Y-%m-%d") # Initialize the model and retrieval chain try: llm = load_model(openai_api_key) interview_retrieval_chain, report_retrieval_chain, combined_retriever = setup_knowledge_retrieval(llm) knowledge_base_connected = True print("Successfully connected to the knowledge base.") except Exception as e: print(f"Error initializing the model or retrieval chain: {str(e)}") knowledge_base_connected = False print("Falling back to basic mode without knowledge base.") question_count = 0 interview_history = [] last_audio_path = None # Variable to store the path of the last audio file initial_audio_path = None # Variable to store the path of the initial audio file language = None def generate_random_string(length=5): return ''.join(random.choices(string.ascii_letters + string.digits, k=length)) def respond(message, history, voice, selected_interviewer): global question_count, interview_history, combined_retriever, last_audio_path, initial_audio_path, language, interview_retrieval_chain, report_retrieval_chain if not isinstance(history, list): history = [] if not history or not history[-1]: history.append(["", ""]) # Extract the actual message text if isinstance(message, list): message = message[-1][0] if message and isinstance(message[-1], list) else message[-1] question_count += 1 interview_history.append(f"Q{question_count}: {message}") history_str = "\n".join(interview_history) print("Starting interview", question_count) try: if knowledge_base_connected: if question_count == 1: # Capture the language from the first response language = message.strip().lower() # Reinitialize the interview chain with the new language interview_retrieval_chain, report_retrieval_chain, combined_retriever = setup_knowledge_retrieval( llm, language, selected_interviewer) if question_count < n_of_questions: result = interview_retrieval_chain.invoke({ "input": f"Based on the patient's statement: '{message}', what should be the next question?", "history": history_str, "question_number": question_count + 1, "language": language }) question = result.get("answer", f"Can you tell me more about that? (in {language})") else: result = generate_report(report_retrieval_chain, interview_history, language) question = result speech_file_path = None # Skip audio generation for the report if question: random_suffix = generate_random_string() speech_file_path = Path(__file__).parent / f"question_{question_count}_{random_suffix}.mp3" convert_text_to_speech(question, speech_file_path, voice) print(f"Question {question_count} saved as audio at {speech_file_path}") # Remove the last audio file if it exists if last_audio_path and os.path.exists(last_audio_path): os.remove(last_audio_path) last_audio_path = speech_file_path else: speech_file_path = None # Skip audio generation for the report else: # Fallback mode without knowledge base question = f"Can you elaborate on that? (in {language})" if question_count < n_of_questions: speech_file_path = Path(__file__).parent / f"question_{question_count}.mp3" convert_text_to_speech(question, speech_file_path, voice) print(f"Question {question_count} saved as audio at {speech_file_path}") if last_audio_path and os.path.exists(last_audio_path): os.remove(last_audio_path) last_audio_path = speech_file_path else: speech_file_path = None history[-1][1] = f"{question}" # Remove the initial question audio file after the first user response if initial_audio_path and os.path.exists(initial_audio_path): os.remove(initial_audio_path) initial_audio_path = None # Clean up older files based on question_count if question_count > 1: previous_audio_path = Path(__file__).parent / f"question_{question_count-1}_{random_suffix}.mp3" if os.path.exists(previous_audio_path): os.remove(previous_audio_path) return history, str(speech_file_path) if speech_file_path else None except Exception as e: print(f"Error in retrieval chain: {str(e)}") print(traceback.format_exc()) return history, None def reset_interview(): """Reset the interview state.""" global question_count, interview_history, last_audio_path, initial_audio_path question_count = 0 interview_history = [] if last_audio_path and os.path.exists(last_audio_path): os.remove(last_audio_path) last_audio_path = None initial_audio_path = None def read_file(file): if file is None: return "No file uploaded" if isinstance(file, str): with open(file, 'r', encoding='utf-8') as f: return f.read() if hasattr(file, 'name'): # Check if it's a file-like object if file.name.endswith('.txt'): return file.content elif file.name.endswith('.pdf'): pdf_reader = PyPDF2.PdfReader(io.BytesIO(file.content)) return "\n".join(page.extract_text() for page in pdf_reader.pages) elif file.name.endswith('.docx'): doc = docx.Document(io.BytesIO(file.content)) return "\n".join(paragraph.text for paragraph in doc.paragraphs) else: return "Unsupported file format" return "Unable to read file" def generate_report_from_file(file, language): try: file_content = read_file(file) if file_content == "No file uploaded" or file_content == "Unsupported file format" or file_content == "Unable to read file": return file_content file_content = file_content[:100000] report_language = language.strip().lower() if language else "english" print('preferred language:', report_language) print(f"Generating report in language: {report_language}") # For debugging # Reinitialize the report chain with the new language _, report_retrieval_chain, _ = setup_knowledge_retrieval(llm, report_language) result = report_retrieval_chain.invoke({ "input": "Please provide a clinical report based on the following content:", "history": file_content, "language": report_language }) report_content = result.get("answer", "Unable to generate report due to insufficient information.") pdf_path = create_pdf(report_content) return report_content, pdf_path except Exception as e: return f"An error occurred while processing the file: {str(e)}", None def generate_interview_report(interview_history, language): try: report_language = language.strip().lower() if language else "english" print('preferred report_language language:', report_language) _, report_retrieval_chain, _ = setup_knowledge_retrieval(llm, report_language) result = report_retrieval_chain.invoke({ "input": "Please provide a clinical report based on the following interview:", "history": "\n".join(interview_history), "language": report_language }) report_content = result.get("answer", "Unable to generate report due to insufficient information.") pdf_path = create_pdf(report_content) return report_content, pdf_path except Exception as e: return f"An error occurred while generating the report: {str(e)}", None def create_pdf(content): random_string = generate_random_string() temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=f'_report.pdf') doc = SimpleDocTemplate(temp_file.name, pagesize=letter) styles = getSampleStyleSheet() # Create a custom style for bold text bold_style = ParagraphStyle('Bold', parent=styles['Normal'], fontName='Helvetica-Bold', fontSize=10) # Create a custom style for normal text with justification normal_style = ParagraphStyle('Normal', parent=styles['Normal'], alignment=TA_JUSTIFY) flowables = [] for line in content.split('\n'): # Use regex to find words surrounded by ** parts = re.split(r'(\*\*.*?\*\*)', line) paragraph_parts = [] for part in parts: if part.startswith('**') and part.endswith('**'): # Bold text bold_text = part.strip('**') paragraph_parts.append(Paragraph(bold_text, bold_style)) else: # Normal text paragraph_parts.append(Paragraph(part, normal_style)) flowables.extend(paragraph_parts) flowables.append(Spacer(1, 12)) # Add space between paragraphs doc.build(flowables) return temp_file.name