Patient-RAG / app.py
AI-RESEARCHER-2024's picture
Update app.py
9930667 verified
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"<b>Assessment Date:</b> {timestamp}", body_style))
elements.append(Spacer(1, 20))
# Add parameter settings
elements.append(Paragraph("<b>OSCE Evaluation Parameters Used:</b>", 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('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;')
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 = """
<div style="background: white; padding: 20px; border-radius: 8px; margin: 20px 0; border: 1px solid #000000; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">
<style>
@keyframes pulse {
0%, 100% { opacity: 1; }
50% { opacity: 0.6; }
}
</style>
<div style="text-align: center; color: #000000; animation: pulse 1.5s ease-in-out infinite;">
<div style="font-size: 24px; font-weight: bold; margin-bottom: 10px; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">
Processing Your Recording...
</div>
<div style="font-size: 16px; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">
Saving video file - Preparing for download
</div>
</div>
</div>
"""
return gr.update(value=status_html, visible=True), video
# Enhanced save function with status updates
def save_recorded_video_with_status(video):
"""Save the recorded video with status updates"""
if video is None:
return None, gr.update(value="", visible=False)
try:
# Create a copy of the video file with a timestamp
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
output_filename = f"recorded_video_{timestamp}.mp4"
temp_output = tempfile.NamedTemporaryFile(delete=False, suffix='.mp4', prefix=f"recorded_{timestamp}_")
# Copy the video file
shutil.copy2(video, temp_output.name)
temp_output.close()
# Success status
success_html = """
<div style="background: white; padding: 15px; border-radius: 8px; margin: 20px 0; border: 1px solid #000000; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">
<div style="text-align: center; color: #000000;">
<div style="font-size: 20px; font-weight: bold; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">
Video Saved Successfully!
</div>
<div style="font-size: 14px; margin-top: 5px; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">
Ready for download - Click Analyze Video to assess
</div>
</div>
</div>
"""
print(f"Video saved: {output_filename}")
return temp_output.name, gr.update(value=success_html, visible=True)
except Exception as e:
print(f"Failed to save video: {str(e)}")
error_html = """
<div style="background: white; padding: 15px; border-radius: 8px; margin: 20px 0; border: 1px solid #000000; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">
<div style="text-align: center; color: #000000;">
<div style="font-size: 20px; font-weight: bold; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">
Error Saving Video
</div>
<div style="font-size: 14px; margin-top: 5px; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">
Please try recording again
</div>
</div>
</div>
"""
return None, gr.update(value=error_html, visible=True)
# Define the core processing function (separate from GPU wrapper)
def process_video_core(video, resize_option, assessor, param1, param2, param3, param4, param5):
"""Process uploaded or recorded video with custom parameters"""
if video is None:
return "Please upload or record a video first.", None, None, None
try:
# Get original video info
orig_width, orig_height, fps, frame_count = get_video_info(video)
if orig_width and orig_height:
print(f"Original video: {orig_width}x{orig_height} @ {fps}fps ({frame_count} frames)")
# Get file size
file_size_mb = os.path.getsize(video) / (1024 * 1024)
print(f"Processing video ({file_size_mb:.1f}MB)...")
# Apply resizing based on user selection
video_to_process = video
temp_resized_file = None
if resize_option != "Original (No Resize)":
# Parse the resolution from the option string
if "640x480" in resize_option:
target_width, target_height = 640, 480
elif "800x600" in resize_option:
target_width, target_height = 800, 600
elif "1280x720" in resize_option:
target_width, target_height = 1280, 720
else:
target_width, target_height = orig_width, orig_height
# Only resize if different from original
if orig_width and orig_height and (orig_width != target_width or orig_height != target_height):
temp_resized_file = resize_video(video, target_width, target_height)
video_to_process = temp_resized_file
# Check new file size
new_file_size_mb = os.path.getsize(video_to_process) / (1024 * 1024)
print(f"Resized video: {new_file_size_mb:.1f}MB (saved {file_size_mb - new_file_size_mb:.1f}MB)")
# Start assessment with parameters
print(f"Starting Standardized Patient Encounter Assessment...")
print(f"Parameters: History Taking={param1}, Communication={param2}, Clinical Reasoning={param3}, Physical Exam={param4}, Professionalism={param5}")
assessment_result, p1, p2, p3, p4, p5 = assessor.analyze_video(video_to_process, param1, param2, param3, param4, param5)
# Clean up temporary resized file if created
if temp_resized_file and temp_resized_file != video:
try:
os.unlink(temp_resized_file)
except:
pass
if "Error" in assessment_result:
return assessment_result, None, None, None
# Generate 1-minute audio feedback
print("Generating 1-minute audio summary...")
audio_path = assessor.generate_audio_feedback(assessment_result)
# Generate PDF report with parameters
print("Generating PDF report...")
pdf_path = assessor.generate_pdf_report(assessment_result, param1, param2, param3, param4, param5)
# Parse scores for visual summary
weighted_score, percentage, level, color, overall_assessment_text = assessor.parse_assessment_scores(assessment_result, param1, param2, param3, param4, param5)
# Clean the overall assessment text for HTML display
clean_overall_assessment = overall_assessment_text.replace('\n', '<br>').replace('*', '').replace('#', '')
# Create enhanced visual summary HTML with parameter information
summary_html = f"""
<div style="max-width:800px; margin:20px auto; padding:30px; border-radius:10px; background:white; border:1px solid #e0e0e0; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">
<h2 style="text-align:center; color:#000000; margin-bottom:30px; font-weight:600; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">Standardized Patient Assessment Summary</h2>
<div style="background:white; padding:20px; border-radius:8px; margin-bottom:30px; border:1px solid #e0e0e0;">
<h3 style="color:#000000; margin-top:0; margin-bottom:15px; font-weight:600; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">Overall Weighted Assessment</h3>
<p style="color:#000000; line-height:1.8; margin:0; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">{clean_overall_assessment}</p>
</div>
<div style="display:flex; justify-content:space-around; margin:30px 0;">
<div style="text-align:center;">
<div style="font-size:48px; font-weight:bold; color:#000000; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">{weighted_score:.1f}/10</div>
<div style="color:#000000; margin-top:10px; font-weight:500; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">OSCE Score</div>
</div>
<div style="text-align:center;">
<div style="font-size:48px; font-weight:bold; color:#000000; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">{percentage:.0f}%</div>
<div style="color:#000000; margin-top:10px; font-weight:500; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">Overall Performance</div>
</div>
</div>
<div style="text-align:center; padding:20px; background:white; border-radius:8px; margin:20px 0; border:1px solid #e0e0e0;">
<div style="font-size:24px; font-weight:bold; color:#000000; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">Performance Level: {level}</div>
</div>
<div style="margin-top:30px;">
<h3 style="color:#000000; margin-bottom:20px; font-weight:600; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">Your OSCE Evaluation Parameters:</h3>
<div style="background:white; padding:20px; border-radius:8px; border:1px solid #e0e0e0;">
<div style="display:flex; justify-content:space-between; margin:10px 0;">
<span style="color:#000000; font-weight:500; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">History Taking & Interview:</span>
<span style="color:#000000; font-weight:bold; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">{param1}/10</span>
</div>
<div style="height:8px; background:#e0e0e0; border-radius:4px; margin:5px 0;">
<div style="height:100%; background:#000000; border-radius:4px; width:{param1*10}%;"></div>
</div>
<div style="display:flex; justify-content:space-between; margin:10px 0; margin-top:20px;">
<span style="color:#000000; font-weight:500; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">Communication & Rapport:</span>
<span style="color:#000000; font-weight:bold; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">{param2}/10</span>
</div>
<div style="height:8px; background:#e0e0e0; border-radius:4px; margin:5px 0;">
<div style="height:100%; background:#000000; border-radius:4px; width:{param2*10}%;"></div>
</div>
<div style="display:flex; justify-content:space-between; margin:10px 0; margin-top:20px;">
<span style="color:#000000; font-weight:500; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">Clinical Reasoning:</span>
<span style="color:#000000; font-weight:bold; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">{param3}/10</span>
</div>
<div style="height:8px; background:#e0e0e0; border-radius:4px; margin:5px 0;">
<div style="height:100%; background:#000000; border-radius:4px; width:{param3*10}%;"></div>
</div>
<div style="display:flex; justify-content:space-between; margin:10px 0; margin-top:20px;">
<span style="color:#000000; font-weight:500; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">Physical Examination:</span>
<span style="color:#000000; font-weight:bold; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">{param4}/10</span>
</div>
<div style="height:8px; background:#e0e0e0; border-radius:4px; margin:5px 0;">
<div style="height:100%; background:#000000; border-radius:4px; width:{param4*10}%;"></div>
</div>
<div style="display:flex; justify-content:space-between; margin:10px 0; margin-top:20px;">
<span style="color:#000000; font-weight:500; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">Professionalism & Empathy:</span>
<span style="color:#000000; font-weight:bold; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">{param5}/10</span>
</div>
<div style="height:8px; background:#e0e0e0; border-radius:4px; margin:5px 0;">
<div style="height:100%; background:#000000; border-radius:4px; width:{param5*10}%;"></div>
</div>
</div>
</div>
<div style="margin-top:30px; padding:20px; background:white; border-radius:8px; border:1px solid #000000;">
<p style="text-align:center; color:#000000; margin:0; font-weight:600; font-family: Calibri, 'Segoe UI', Arial, sans-serif;">
Listen to the 1-minute audio summary for key findings<br>
Download the PDF report for complete OSCE documentation
</p>
</div>
</div>
"""
return assessment_result, summary_html, audio_path, pdf_path
except Exception as e:
error_msg = f"Error during processing: {str(e)}"
print(error_msg)
return error_msg, None, None, None
# Wrapper function that calls the GPU-accelerated version
def process_video(video, resize_option, param1, param2, param3, param4, param5):
"""Wrapper function to call GPU-accelerated processing with parameters"""
return process_video_with_gpu(video, resize_option, param1, param2, param3, param4, param5)
# Create and launch the Gradio interface with parameter controls
print("Launching Standardized Patient Assessment Tool...")
with gr.Blocks(title="Standardized Patient Assessment Tool") as demo:
gr.Markdown("""
# Standardized Patient Encounter Assessment Tool
**OSCE-Style Clinical Skills Evaluation with Customizable Parameters**
This tool analyzes Standardized Patient (SP) encounter videos and evaluates clinical competencies
based on your prioritized assessment criteria. Perfect for medical education, nursing programs,
and healthcare professional training.
Set higher values for areas you want to prioritize in the assessment.
---
""")
with gr.Row():
with gr.Column(scale=1):
gr.Markdown("### Video Input")
# Video resolution dropdown
resize_dropdown = gr.Dropdown(
choices=[
"Original (No Resize)",
"640x480 (Fastest - Recommended for quick tests)",
"800x600 (Fast - Good balance)",
"1280x720 (HD - Best quality, slower)"
],
value="800x600 (Fast - Good balance)",
label="Video Resolution",
info="Lower resolutions process faster and use less API quota"
)
video_input = gr.Video(
label="Upload or Record Video",
sources=["upload", "webcam"],
format="mp4",
include_audio=True,
interactive=True,
autoplay=False,
show_download_button=True
)
# Status bar for immediate feedback
status_bar = gr.HTML(
value="",
visible=False,
elem_id="status-bar"
)
# Add download component for recorded videos
recorded_video_download = gr.File(
label="Download Recorded Video",
interactive=False,
visible=False
)
gr.Markdown("### Evaluation Parameters")
gr.Markdown("**Set the importance (0-10) for each OSCE assessment area:**")
# Add the 5 parameter sliders for SP encounters
param1_slider = gr.Slider(
minimum=0,
maximum=10,
value=8,
step=1,
label="History Taking & Interview Skills",
info="HPI, PMH, medications, allergies, social history, ROS, questioning technique"
)
param2_slider = gr.Slider(
minimum=0,
maximum=10,
value=9,
step=1,
label="Communication & Rapport",
info="Introduction, active listening, patient-friendly language, non-verbal cues"
)
param3_slider = gr.Slider(
minimum=0,
maximum=10,
value=7,
step=1,
label="Clinical Reasoning & Assessment",
info="Differential diagnosis, red flags, diagnostic thinking, clinical frameworks"
)
param4_slider = gr.Slider(
minimum=0,
maximum=10,
value=6,
step=1,
label="Physical Examination Technique",
info="Hand hygiene, systematic approach, correct technique, patient comfort"
)
param5_slider = gr.Slider(
minimum=0,
maximum=10,
value=8,
step=1,
label="Professionalism & Empathy",
info="Patient dignity, empathetic responses, cultural sensitivity, ethics"
)
gr.Markdown("""
### Instructions:
1. **Set your OSCE evaluation parameters** (higher = more important)
2. **Select video resolution** (lower = faster processing)
3. **Upload** a recorded SP encounter or **Record** live
4. Click **Analyze Video** to start assessment
5. Review OSCE-style results weighted by your priorities
""")
with gr.Column(scale=2):
gr.Markdown("### Assessment Results")
# Move analyze button here
analyze_btn = gr.Button("Analyze Video", variant="primary", size="lg")
# Visual summary
summary_output = gr.HTML(
label="Visual Summary",
value="<p style='text-align:center; color:#000000; padding:40px; font-family: Calibri, Arial, sans-serif;'>Results will appear here after analysis...</p>"
)
# 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()