# utils/oneclick.py from typing import Tuple, Optional, Dict from .meldrx import MeldRxAPI from .responseparser import PatientDataExtractor from .pdfutils import PDFGenerator from .verifier import DischargeVerifier import logging import json from huggingface_hub import InferenceClient import os logger = logging.getLogger(__name__) HF_TOKEN = os.getenv("HF_TOKEN") if not HF_TOKEN: raise ValueError("HF_TOKEN environment variable not set.") client = InferenceClient(api_key=HF_TOKEN) MODEL_NAME = "meta-llama/Llama-3.3-70B-Instruct" verifier = DischargeVerifier() def generate_ai_discharge_summary(patient_dict: Dict[str, str], client) -> Tuple[Optional[str], Optional[str]]: """Generate a discharge summary using AI and verify it for hallucinations.""" try: formatted_summary = format_discharge_summary(patient_dict) logger.info("Generating AI discharge summary with patient info: %s", formatted_summary) messages = [ { "role": "assistant", "content": ( "You are a senior medical practitioner tasked with creating discharge summaries. " "Generate a complete discharge summary based on the provided patient information." ) }, {"role": "user", "content": formatted_summary} ] stream = client.chat.completions.create( model=MODEL_NAME, messages=messages, temperature=0.4, max_tokens=3584, top_p=0.7, stream=True ) discharge_summary = "" for chunk in stream: content = chunk.choices[0].delta.content if content: discharge_summary += content discharge_summary = discharge_summary.strip() logger.info("AI discharge summary generated successfully") question = "Provide a complete discharge summary based on the patient information." verified_summary = verifier.verify_discharge_summary( context=formatted_summary, question=question, answer=discharge_summary ) return discharge_summary, verified_summary except Exception as e: logger.error(f"Error generating AI discharge summary: {str(e)}", exc_info=True) return None, None def generate_discharge_paper_one_click( api: MeldRxAPI, client, patient_id: str = "", first_name: str = "", last_name: str = "" ) -> Tuple[Optional[str], str, Optional[str], Optional[str], Optional[str]]: try: patients_data = api.get_patients() if not patients_data or "entry" not in patients_data: logger.error("No patient data received from MeldRx API") return None, "Failed to fetch patient data from MeldRx API", None, None, None logger.debug(f"Raw patient data from API: {patients_data}") extractor = PatientDataExtractor(patients_data, "json") if not extractor.patients: logger.error("No patients found in the parsed data") return None, "No patients found in the data", None, None, None logger.info(f"Found {len(extractor.patients)} patients in the data") matching_patient = None patient_id_input = str(patient_id).strip().lower() first_name_input = str(first_name).strip().lower() last_name_input = str(last_name).strip().lower() for i in range(len(extractor.patients)): extractor.set_patient_by_index(i) patient_data = extractor.get_patient_dict() logger.debug(f"Processing patient {i}: {patient_data}") # Add debug log patient_id_from_data = str(patient_data.get('id', '')).strip().lower() if patient_id_input and patient_id_from_data == patient_id_input: matching_patient = patient_data logger.info(f"Exact match found for patient ID: {patient_id_from_data}") break if not matching_patient and (first_name_input or last_name_input): for i in range(len(extractor.patients)): extractor.set_patient_by_index(i) patient_data = extractor.get_patient_dict() first_name_from_data = str(patient_data.get('first_name', '')).strip().lower() last_name_from_data = str(patient_data.get('last_name', '')).strip().lower() if (first_name_input == first_name_from_data and last_name_input == last_name_from_data): matching_patient = patient_data logger.info(f"Match found by name: {first_name_from_data} {last_name_from_data}") break if not matching_patient: search_criteria = f"ID: {patient_id or 'N/A'}, First: {first_name or 'N/A'}, Last: {last_name or 'N/A'}" all_patient_ids = [str(p.get('id', '')) for p in extractor.get_all_patients()] all_patient_names = [f"{p.get('first_name', '')} {p.get('last_name', '')}".strip() for p in extractor.get_all_patients()] logger.warning(f"No patients matched criteria: {search_criteria}") return None, (f"No patients found matching criteria: {search_criteria}\n" f"Available IDs: {', '.join(all_patient_ids)}\n" f"Available Names: {', '.join(all_patient_names)}"), None, None, None logger.info(f"Selected patient data: {matching_patient}") basic_summary = format_discharge_summary(matching_patient) ai_summary, verified_summary = generate_ai_discharge_summary(matching_patient, client) if not ai_summary or not verified_summary: return None, "Failed to generate or verify AI summary", basic_summary, None, None pdf_gen = PDFGenerator() filename = f"discharge_{matching_patient.get('id', 'unknown')}_{matching_patient.get('last_name', 'patient')}.pdf" pdf_path = pdf_gen.generate_pdf_from_text(ai_summary, filename) if pdf_path: return pdf_path, "Discharge summary generated and verified successfully", basic_summary, ai_summary, verified_summary return None, "Failed to generate PDF file", basic_summary, ai_summary, verified_summary except Exception as e: logger.error(f"Error in one-click discharge generation: {str(e)}", exc_info=True) return None, f"Error generating discharge summary: {str(e)}", None, None, None def format_discharge_summary(patient_data: dict) -> str: """Format patient data into a discharge summary text.""" patient_data.setdefault('name_prefix', '') patient_data.setdefault('first_name', '') patient_data.setdefault('last_name', '') patient_data.setdefault('dob', 'Unknown') patient_data.setdefault('age', 'Unknown') patient_data.setdefault('sex', 'Unknown') patient_data.setdefault('id', 'Unknown') patient_data.setdefault('address', 'Unknown') patient_data.setdefault('city', 'Unknown') patient_data.setdefault('state', 'Unknown') patient_data.setdefault('zip_code', 'Unknown') patient_data.setdefault('phone', 'Unknown') patient_data.setdefault('admission_date', 'Unknown') patient_data.setdefault('discharge_date', 'Unknown') patient_data.setdefault('diagnosis', 'Unknown') patient_data.setdefault('medications', 'None specified') patient_data.setdefault('doctor_first_name', 'Unknown') patient_data.setdefault('doctor_last_name', 'Unknown') patient_data.setdefault('hospital_name', 'Unknown') patient_data.setdefault('doctor_address', 'Unknown') patient_data.setdefault('doctor_city', 'Unknown') patient_data.setdefault('doctor_state', 'Unknown') patient_data.setdefault('doctor_zip', 'Unknown') summary = [ "DISCHARGE SUMMARY", "", "PATIENT INFORMATION", f"Name: {patient_data['name_prefix']} {patient_data['first_name']} {patient_data['last_name']}".strip(), f"Date of Birth: {patient_data['dob']}", f"Age: {patient_data['age']}", f"Gender: {patient_data['sex']}", f"Patient ID: {patient_data['id']}", "", "CONTACT INFORMATION", f"Address: {patient_data['address']}", f"City: {patient_data['city']}, {patient_data['state']} {patient_data['zip_code']}", f"Phone: {patient_data['phone']}", "", "ADMISSION INFORMATION", f"Admission Date: {patient_data['admission_date']}", f"Discharge Date: {patient_data['discharge_date']}", f"Diagnosis: {patient_data['diagnosis']}", "", "MEDICATIONS", f"{patient_data['medications']}", "", "PHYSICIAN INFORMATION", f"Physician: Dr. {patient_data['doctor_first_name']} {patient_data['doctor_last_name']}".strip(), f"Hospital: {patient_data['hospital_name']}", f"Address: {patient_data['doctor_address']}", f"City: {patient_data['doctor_city']}, {patient_data['doctor_state']} {patient_data['doctor_zip']}", ] return "\n".join(line for line in summary if line.strip() or line == "")