import spaces # Configure ZeroGPU @spaces.GPU def process_video_with_gpu(video, resize_option, param1, param2, param3, param4, param5): """ZeroGPU-accelerated video processing with custom parameters""" # Create assessor inside the GPU function to avoid pickling issues from google import genai client = genai.Client(api_key=GOOGLE_API_KEY) assessor = CICE_Assessment(client) return process_video_core(video, resize_option, assessor, param1, param2, param3, param4, param5) import gradio as gr from google import genai from google.genai import types import os import time from datetime import datetime import re from gtts import gTTS import tempfile import numpy as np from PIL import Image import cv2 from reportlab.lib.pagesizes import letter from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, PageBreak from reportlab.lib.units import inch from reportlab.lib.enums import TA_JUSTIFY, TA_CENTER from reportlab.lib.colors import HexColor import subprocess import shutil # Configure Google API Key from environment variable or Hugging Face secrets print("Setting up Google API Key...") GOOGLE_API_KEY = os.environ.get('GOOGLE_API_KEY') if not GOOGLE_API_KEY: raise ValueError("GOOGLE_API_KEY environment variable is not set. Please set it in Hugging Face Spaces secrets.") client = genai.Client(api_key=GOOGLE_API_KEY) print("Google Generative AI configured successfully!") # Define the CICE Assessment Class with parameters class CICE_Assessment: def __init__(self, client): self.client = client self.model_name = "gemini-2.0-flash-exp" def analyze_video(self, video_path, param1, param2, param3, param4, param5): """Analyze video using customizable assessment parameters""" try: # Determine mime type based on file extension import mimetypes mime_type, _ = mimetypes.guess_type(video_path) if mime_type is None: # Default to mp4 if cannot determine mime_type = 'video/mp4' # Upload video to Gemini print(f"Uploading video to Gemini AI (type: {mime_type})...") with open(video_path, 'rb') as f: video_file = self.client.files.upload(file=f, config={'mime_type': mime_type}) # Wait for processing print("Processing video (this may take 30-60 seconds)...") max_wait = 300 wait_time = 0 while video_file.state == "PROCESSING" and wait_time < max_wait: time.sleep(3) wait_time += 3 video_file = self.client.files.get(name=video_file.name) if video_file.state == "FAILED": raise Exception("Video processing failed") print("Analyzing team interactions with custom parameters...") # Build dynamic assessment prompt based on parameters prompt = self.build_assessment_prompt(param1, param2, param3, param4, param5) response = self.client.models.generate_content( model=self.model_name, contents=[ types.Part.from_uri(file_uri=video_file.uri, mime_type=video_file.mime_type), prompt ] ) print("Analysis complete!") return response.text, param1, param2, param3, param4, param5 except Exception as e: return f"Error during analysis: {str(e)}", param1, param2, param3, param4, param5 def build_assessment_prompt(self, history_taking_weight, communication_weight, clinical_reasoning_weight, physical_exam_weight, professionalism_weight): """Build a dynamic prompt based on user-selected parameters for Standardized Patient encounters""" # Normalize weights total_weight = history_taking_weight + communication_weight + clinical_reasoning_weight + physical_exam_weight + professionalism_weight if total_weight == 0: total_weight = 1 # Avoid division by zero hist_pct = (history_taking_weight / total_weight) * 100 comm_pct = (communication_weight / total_weight) * 100 clinical_pct = (clinical_reasoning_weight / total_weight) * 100 physical_pct = (physical_exam_weight / total_weight) * 100 prof_pct = (professionalism_weight / total_weight) * 100 prompt = f"""Analyze this Standardized Patient (SP) clinical encounter video with the following CUSTOMIZED EVALUATION PARAMETERS: This is an OSCE-style (Objective Structured Clinical Examination) assessment of a healthcare provider/student interacting with a standardized patient in a simulated clinical setting. EVALUATION WEIGHTS (Total 100%): 1. HISTORY TAKING & INTERVIEW SKILLS: {hist_pct:.1f}% weight 2. COMMUNICATION & RAPPORT: {comm_pct:.1f}% weight 3. CLINICAL REASONING & ASSESSMENT: {clinical_pct:.1f}% weight 4. PHYSICAL EXAMINATION TECHNIQUE: {physical_pct:.1f}% weight 5. PROFESSIONALISM & EMPATHY: {prof_pct:.1f}% weight Please evaluate the clinical encounter based on these weighted priorities: """ # Add detailed criteria based on weights criteria_sections = [] if history_taking_weight > 0: criteria_sections.append(f""" ## HISTORY TAKING & INTERVIEW SKILLS (Weight: {history_taking_weight}/10) Evaluate: - Chief complaint identification and exploration - History of Present Illness (HPI) - OLDCARTS/OPQRST methodology - Past Medical History (PMH) inquiry - Medication and allergy review - Family and social history assessment - Review of Systems (ROS) completeness - Open-ended vs. closed-ended question balance - Logical flow and organization of questioning - Avoidance of leading questions - Appropriate follow-up questions based on responses """) if communication_weight > 0: criteria_sections.append(f""" ## COMMUNICATION & RAPPORT (Weight: {communication_weight}/10) Evaluate: - Introduction and identification (name, role, purpose) - Active listening behaviors (eye contact, nodding, verbal acknowledgment) - Use of patient-friendly language (avoiding medical jargon) - Clarification and summarization of patient statements - Appropriate pacing and allowing patient to speak - Non-verbal communication (body posture, positioning) - Addressing patient concerns and questions - Clear explanations of procedures or next steps - Checking for patient understanding (teach-back) - Closure and summary of encounter """) if clinical_reasoning_weight > 0: criteria_sections.append(f""" ## CLINICAL REASONING & ASSESSMENT (Weight: {clinical_reasoning_weight}/10) Evaluate: - Differential diagnosis consideration - Recognition of red flag symptoms - Appropriate diagnostic questioning - Integration of history findings - Clinical decision-making process - Prioritization of problems - Evidence of systematic thinking - Appropriate use of clinical frameworks - Recognition of urgent vs. non-urgent conditions - Formulation of assessment and plan """) if physical_exam_weight > 0: criteria_sections.append(f""" ## PHYSICAL EXAMINATION TECHNIQUE (Weight: {physical_exam_weight}/10) Evaluate: - Appropriate hand hygiene and infection control - Patient positioning and draping for dignity - Systematic examination approach - Correct technique for examination maneuvers - Appropriate use of equipment (stethoscope, etc.) - Explanation of examination steps to patient - Patient comfort during examination - Vital signs assessment - Focused vs. comprehensive exam appropriateness - Documentation of findings verbally or noted """) if professionalism_weight > 0: criteria_sections.append(f""" ## PROFESSIONALISM & EMPATHY (Weight: {professionalism_weight}/10) Evaluate: - Respect for patient dignity and privacy - Empathetic responses to patient emotions - Cultural sensitivity and awareness - Appropriate professional boundaries - Honesty and transparency - Patient-centered approach - Confidentiality awareness - Appropriate attire and presentation - Time management within encounter - Ethical behavior and decision-making """) prompt += "".join(criteria_sections) prompt += f""" STRUCTURE YOUR RESPONSE AS FOLLOWS: ## OVERALL WEIGHTED ASSESSMENT Provide an overall assessment summary based on the weighted parameters above, highlighting the key observations from this standardized patient encounter. ## DETAILED EVALUATION BY PARAMETER For each parameter with weight > 0, provide: - Parameter Name: [Name] - Weight: [X/10] - Score: [X/10] - Specific Observations: [What was observed in the encounter] - Strengths: [What was done well] - Areas for Improvement: [Specific recommendations] ## KEY STRENGTHS Top 3-5 strengths observed in this clinical encounter (prioritize based on weighted parameters) ## CRITICAL IMPROVEMENTS NEEDED Top 3-5 areas needing improvement for future SP encounters (prioritize based on weighted parameters) ## WEIGHTED FINAL SCORE Calculate the weighted average score: - History Taking: {history_taking_weight}/10 weight × [score]/10 - Communication: {communication_weight}/10 weight × [score]/10 - Clinical Reasoning: {clinical_reasoning_weight}/10 weight × [score]/10 - Physical Examination: {physical_exam_weight}/10 weight × [score]/10 - Professionalism: {professionalism_weight}/10 weight × [score]/10 TOTAL WEIGHTED SCORE: [X]/10 Performance Level: [Exemplary (8.5-10)/Proficient (7-8.4)/Developing (5-6.9)/Needs Improvement (0-4.9)] OSCE Station Result: [Pass/Borderline/Fail based on score] ## AUDIO SUMMARY [Create a 60-second spoken summary focusing on: the overall weighted score, top strengths demonstrated in this SP encounter, critical improvements needed for future clinical encounters, and 2-3 actionable recommendations for the learner. Write in natural, conversational tone suitable for text-to-speech feedback.] """ return prompt def generate_audio_feedback(self, text): """Generate a concise 1-minute audio feedback summary""" # Extract the audio summary section from the assessment audio_summary_match = re.search(r'## AUDIO SUMMARY\s*(.*?)(?=##|\Z)', text, re.DOTALL) if audio_summary_match: summary_text = audio_summary_match.group(1).strip() else: # Fallback: Create a brief summary from the assessment summary_text = self.create_brief_summary(text) # Clean text for speech clean_text = re.sub(r'[#*_\[\]()]', ' ', summary_text) clean_text = re.sub(r'\s+', ' ', clean_text) clean_text = re.sub(r'[-•·]\s+', '', clean_text) # Add introduction and conclusion for better audio experience audio_script = f"""Healthcare Team Assessment Summary. {clean_text} Please refer to the detailed written report for complete evaluation and specific recommendations. End of audio summary.""" # Generate audio with gTTS try: tts = gTTS(text=audio_script, lang='en', slow=False, tld='com') # Create a proper temporary file temp_audio = tempfile.NamedTemporaryFile(delete=False, suffix='.mp3') tts.save(temp_audio.name) temp_audio.close() return temp_audio.name except Exception as e: print(f"Audio generation failed: {str(e)}") return None def create_brief_summary(self, text): """Create a brief summary if AUDIO SUMMARY section is not found""" summary = f"""The team assessment has been completed based on your customized evaluation parameters. The analysis focused on the specific areas you prioritized, with weighted scores reflecting the importance you assigned to each parameter. Key strengths were identified in the high-priority areas, and recommendations have been provided for critical improvements. Please review the detailed report for specific behavioral observations and actionable feedback tailored to your evaluation criteria.""" return summary def parse_assessment_scores(self, assessment_text, param1, param2, param3, param4, param5): """Parse assessment text to extract weighted scores and overall assessment""" import re # Extract the OVERALL WEIGHTED ASSESSMENT section overall_assessment_match = re.search( r'## OVERALL WEIGHTED ASSESSMENT\s*(.*?)(?=##|\Z)', assessment_text, re.DOTALL | re.IGNORECASE ) if overall_assessment_match: overall_assessment_text = overall_assessment_match.group(1).strip() else: overall_assessment_text = "Assessment completed. See detailed evaluation below." # Look for "TOTAL WEIGHTED SCORE: X/10" pattern score_pattern = r'TOTAL WEIGHTED SCORE:\s*([0-9.]+)/10' match = re.search(score_pattern, assessment_text, re.IGNORECASE) if match: weighted_score = float(match.group(1)) else: # Fallback calculation weighted_score = 7.5 # Default middle score percentage = (weighted_score / 10) * 100 # Extract performance level from text if available level_pattern = r'Performance Level:\s*(\w+)' level_match = re.search(level_pattern, assessment_text, re.IGNORECASE) if level_match: level = level_match.group(1) else: # Determine performance level based on score if weighted_score >= 8.5: level = "Exemplary" elif weighted_score >= 7: level = "Proficient" elif weighted_score >= 5: level = "Developing" else: level = "Needs Improvement" # Determine color based on score - using black for clean look color = "#000000" return weighted_score, percentage, level, color, overall_assessment_text def generate_pdf_report(self, assessment_text, param1, param2, param3, param4, param5): """Generate a PDF report from the assessment text with parameter information""" try: # Create a temporary file for the PDF temp_pdf = tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') # Create the PDF document doc = SimpleDocTemplate( temp_pdf.name, pagesize=letter, rightMargin=72, leftMargin=72, topMargin=72, bottomMargin=18, ) # Container for the 'Flowable' objects elements = [] # Define styles with professional colors and Calibri font styles = getSampleStyleSheet() title_style = ParagraphStyle( 'CustomTitle', parent=styles['Heading1'], fontName='Helvetica-Bold', fontSize=24, textColor=HexColor('#000000'), spaceAfter=30, alignment=TA_CENTER ) heading_style = ParagraphStyle( 'CustomHeading', parent=styles['Heading2'], fontName='Helvetica-Bold', fontSize=14, textColor=HexColor('#000000'), spaceAfter=12, spaceBefore=12, ) body_style = ParagraphStyle( 'CustomBody', parent=styles['BodyText'], fontName='Helvetica', fontSize=11, textColor=HexColor('#000000'), alignment=TA_JUSTIFY, spaceAfter=12 ) # Add title elements.append(Paragraph("Standardized Patient Encounter Assessment Report", title_style)) elements.append(Paragraph("(OSCE-Style Clinical Skills Evaluation)", body_style)) elements.append(Spacer(1, 12)) # Add timestamp timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") elements.append(Paragraph(f"Assessment Date: {timestamp}", body_style)) elements.append(Spacer(1, 20)) # Add parameter settings elements.append(Paragraph("OSCE Evaluation Parameters Used:", heading_style)) elements.append(Paragraph(f"History Taking and Interview Skills: {param1}/10", body_style)) elements.append(Paragraph(f"Communication and Rapport: {param2}/10", body_style)) elements.append(Paragraph(f"Clinical Reasoning and Assessment: {param3}/10", body_style)) elements.append(Paragraph(f"Physical Examination Technique: {param4}/10", body_style)) elements.append(Paragraph(f"Professionalism and Empathy: {param5}/10", body_style)) elements.append(Spacer(1, 20)) # Process the assessment text into PDF-friendly format lines = assessment_text.split('\n') for line in lines: line = line.strip() if not line: elements.append(Spacer(1, 6)) elif line.startswith('##'): # Major heading heading_text = line.replace('##', '').strip() elements.append(Paragraph(heading_text, heading_style)) elif line.startswith('#'): # Sub-heading heading_text = line.replace('#', '').strip() elements.append(Paragraph(heading_text, body_style)) else: # Regular text - escape special characters for PDF line = line.replace('&', '&').replace('<', '<').replace('>', '>') elements.append(Paragraph(line, body_style)) # Build PDF doc.build(elements) temp_pdf.close() return temp_pdf.name except Exception as e: print(f"PDF generation failed: {str(e)}") # Fallback to text file temp_txt = tempfile.NamedTemporaryFile(delete=False, suffix='.txt', mode='w') temp_txt.write("Standardized Patient Encounter Assessment Report\n") temp_txt.write("(OSCE-Style Clinical Skills Evaluation)\n") temp_txt.write("="*60 + "\n") temp_txt.write(f"Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") temp_txt.write("="*60 + "\n\n") temp_txt.write(f"Parameters: History Taking={param1}, Communication={param2}, Clinical Reasoning={param3}, Physical Exam={param4}, Professionalism={param5}\n\n") temp_txt.write(assessment_text) temp_txt.close() return temp_txt.name # Initialize the assessment tool assessor = CICE_Assessment(client) # Add video processing helper functions def resize_video(input_path, target_width, target_height): """Resize video to target dimensions to speed up processing""" try: # Open the video cap = cv2.VideoCapture(input_path) # Get original video properties fps = int(cap.get(cv2.CAP_PROP_FPS)) fourcc = cv2.VideoWriter_fourcc(*'mp4v') # Create temporary output file temp_output = tempfile.NamedTemporaryFile(delete=False, suffix='.mp4') temp_output.close() # Create video writer with new dimensions out = cv2.VideoWriter(temp_output.name, fourcc, fps, (target_width, target_height)) print(f"Resizing video to {target_width}x{target_height}...") frame_count = 0 while True: ret, frame = cap.read() if not ret: break # Resize frame resized_frame = cv2.resize(frame, (target_width, target_height)) out.write(resized_frame) frame_count += 1 cap.release() out.release() print(f"Video resized successfully ({frame_count} frames)") return temp_output.name except Exception as e: print(f"Video resize failed: {str(e)}") return input_path # Return original if resize fails def get_video_info(video_path): """Get video dimensions and other info""" try: cap = cv2.VideoCapture(video_path) width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) fps = int(cap.get(cv2.CAP_PROP_FPS)) frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) cap.release() return width, height, fps, frame_count except: return None, None, None, None # Function to show immediate status when recording stops def show_saving_status(video): """Show immediate status bar when recording stops""" if video is None: return gr.update(visible=False), None # Create animated status HTML status_html = """
{clean_overall_assessment}
Listen to the 1-minute audio summary for key findings
Download the PDF report for complete OSCE documentation
Results will appear here after analysis...
" ) # Audio feedback - downloadable audio_output = gr.Audio( label="1-Minute Audio Summary (Downloadable)", type="filepath", interactive=False ) # PDF report - downloadable pdf_output = gr.File( label="Download Full PDF Report", interactive=False, file_types=[".pdf", ".txt"] ) # Detailed assessment text assessment_output = gr.Textbox( label="Detailed Assessment (Text View)", lines=20, max_lines=30, interactive=False, placeholder="Detailed assessment will appear here..." ) # Footer gr.Markdown(""" --- ### About Standardized Patient Assessment This tool uses Google's Gemini AI to evaluate clinical encounters based on OSCE-style criteria. **Evaluation Parameters:** - **History Taking (8-10)**: Essential for diagnostic encounters - **Communication (8-10)**: Critical for all patient interactions - **Clinical Reasoning (6-8)**: Important for diagnostic scenarios - **Physical Exam (4-7)**: Weight based on encounter type - **Professionalism (7-9)**: Always important in clinical settings **OSCE Scoring:** - Exemplary (8.5-10): Exceeds expectations - Clear Pass - Proficient (7-8.4): Meets expectations - Pass - Developing (5-6.9): Borderline performance - Borderline Pass - Needs Improvement (0-4.9): Below expectations - Fail **Powered by Google Gemini 2.0 Flash | Designed for Medical Education** """) # Auto-save video when recording stops video_input.stop_recording( fn=show_saving_status, inputs=[video_input], outputs=[status_bar, video_input], api_name="show_status" ).then( fn=save_recorded_video_with_status, inputs=[video_input], outputs=[recorded_video_download, status_bar], api_name="save_video" ).then( fn=lambda x: gr.update(visible=True if x else False), inputs=[recorded_video_download], outputs=[recorded_video_download] ).then( fn=lambda: time.sleep(3), inputs=[], outputs=[] ).then( fn=lambda: gr.update(value="", visible=False), inputs=[], outputs=[status_bar] ) # Connect the analyze button with all parameters analyze_btn.click( fn=process_video, inputs=[ video_input, resize_dropdown, param1_slider, param2_slider, param3_slider, param4_slider, param5_slider ], outputs=[assessment_output, summary_output, audio_output, pdf_output], api_name="analyze" ) # Launch the app if __name__ == "__main__": demo.launch()