import spaces # Configure ZeroGPU @spaces.GPU def process_video_with_gpu(video, resize_option): """ZeroGPU-accelerated video processing""" # 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) 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 class CICE_Assessment: def __init__(self, client): self.client = client self.model_name = "gemini-2.0-flash-exp" def analyze_video(self, video_path): """Analyze video using the 18-point CICE 2.0 assessment with specific behavioral cues""" 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...") # CICE 2.0 Assessment Prompt prompt = """Analyze this healthcare team interaction video and provide a comprehensive assessment based on the CICE 2.0 instrument's 18 interprofessional competencies, looking for these SPECIFIC BEHAVIORAL CUES: For EACH competency, clearly state whether it was "OBSERVED" or "NOT OBSERVED" based on these specific behaviors: 1. IDENTIFIES FACTORS INFLUENCING HEALTH STATUS LOOK FOR: Team mentions allergy bracelet, fall-related trauma, multiple injuries, or states airway/breathing/circulation concerns out loud 2. IDENTIFIES TEAM GOALS FOR THE PATIENT LOOK FOR: Team verbalizes goals like: stabilize airway, CPR/AED, give epinephrine, control bleeding, preserve tooth, prepare EMS handoff 3. PRIORITIZES GOALS FOCUSED ON IMPROVING HEALTH OUTCOMES LOOK FOR: CPR/AED prioritized before bleeding/dental injury, EpiPen administered before addressing secondary injuries 4. VERBALIZES DISCIPLINE-SPECIFIC ROLE (PRE-BRIEF) LOOK FOR: Students acknowledge interprofessional communication expectations and scene safety review before scenario begins 5. OFFERS TO SEEK GUIDANCE FROM COLLEAGUES LOOK FOR: Peer-to-peer checks (e.g., dental to dental: confirm tooth storage; nursing to nursing: confirm CPR quality) 6. COMMUNICATES ABOUT COST-EFFECTIVE AND TIMELY CARE LOOK FOR: Team chooses readily available supplies (AED, saline, tourniquet) without delay, states need for rapid EMS transfer 7. DIRECTS QUESTIONS TO OTHER HEALTH PROFESSIONALS BASED ON EXPERTISE LOOK FOR: Asks discipline-specific expertise (e.g., "Dentalβwhat do we do with the tooth?"), invites pharmacy/medical input on epinephrine use 8. AVOIDS DISCIPLINE-SPECIFIC TERMINOLOGY LOOK FOR: Uses plain language like "no pulse" instead of "asystole" 9. EXPLAINS DISCIPLINE-SPECIFIC TERMINOLOGY WHEN NECESSARY LOOK FOR: Clarifies medical/dental terms for others when necessary 10. COMMUNICATES ROLES AND RESPONSIBILITIES CLEARLY LOOK FOR: Announces assignments out loud: "I'll do compressions," "I'll call 911," "I'll document" 11. ENGAGES IN ACTIVE LISTENING LOOK FOR: Repeats back instructions ("Everyone clear for shock"), pauses to hear teammates' updates 12. SOLICITS AND ACKNOWLEDGES PERSPECTIVES LOOK FOR: Leader asks "Anything else we need to address?", responds to peer input respectfully 13. RECOGNIZES APPROPRIATE CONTRIBUTIONS LOOK FOR: Affirms correct actions verbally ("Good catch on allergy bracelet"), non-verbal acknowledgment (nodding, thumbs up) 14. RESPECTFUL OF OTHER TEAM MEMBERS LOOK FOR: Listens without interrupting, values input across professions 15. COLLABORATIVELY WORKS THROUGH INTERPROFESSIONAL CONFLICTS LOOK FOR: Negotiates intervention priorities (airway vs. bleeding) respectfully 16. REFLECTS ON STRENGTHS OF TEAM INTERACTIONS (POST-BRIEF) LOOK FOR: Notes strong teamwork, communication, or role clarity after the scenario 17. REFLECTS ON CHALLENGES OF TEAM INTERACTIONS (POST-BRIEF) LOOK FOR: Identifies confusion, delays, or role overlap in debriefing 18. IDENTIFIES HOW TO IMPROVE TEAM EFFECTIVENESS (POST-BRIEF) LOOK FOR: Suggests faster role assignment, consistent closed-loop communication, earlier epi use STRUCTURE YOUR RESPONSE AS FOLLOWS: ## OVERALL ASSESSMENT Brief overview of the team interaction quality. ## DETAILED COMPETENCY EVALUATION For each of the 18 competencies, format as: Competency [number]: [name] Status: [OBSERVED/NOT OBSERVED] Evidence: [Specific behavioral cue observed or explanation of absence] ## STRENGTHS Top 3-5 key strengths with specific examples ## AREAS FOR IMPROVEMENT Top 3-5 areas needing work with specific suggestions ## AUDIO SUMMARY [Create a concise 60-second spoken summary focusing on: overall performance level, top 3 strengths, top 3 areas for improvement, and 2 key actionable recommendations. Write this in a natural, conversational tone suitable for text-to-speech narration.] ## FINAL SCORE Competencies Observed: X/18 Overall Performance Level: [Exemplary (85-100%)/Proficient (70-84%)/Developing (50-69%)/Needs Improvement (0-49%)]""" 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 except Exception as e: return f"Error during analysis: {str(e)}" 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"""CICE Healthcare Team Assessment Summary. {clean_text} Please refer to the detailed written report for complete competency 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""" # Parse scores observed_count = text.lower().count("observed") - text.lower().count("not observed") total = 18 percentage = (observed_count / total) * 100 # Determine performance level if percentage >= 85: level = "Exemplary" elif percentage >= 70: level = "Proficient" elif percentage >= 50: level = "Developing" else: level = "Needs Improvement" summary = f"""The team demonstrated {level} performance with {observed_count} out of {total} competencies observed, achieving {percentage:.0f} percent overall. Key strengths included strong team communication and role clarity. Areas for improvement include enhancing active listening and conflict resolution skills. The team should focus on pre-briefing protocols and post-scenario debriefing to enhance future performance. Emphasis should be placed on clear role assignment and closed-loop communication during critical interventions.""" return summary def parse_assessment_scores(self, assessment_text): """Parse assessment text to extract scores""" # Method 1: Look for "Status: OBSERVED" vs "Status: NOT OBSERVED" patterns import re # Find all status lines status_pattern = r'Status:\s*(OBSERVED|NOT OBSERVED)' matches = re.findall(status_pattern, assessment_text, re.IGNORECASE) # Count only "OBSERVED" (not "NOT OBSERVED") observed_count = sum(1 for match in matches if match.upper() == "OBSERVED") # If no matches found with Status: pattern, try alternative parsing if len(matches) == 0: # Alternative: Look for competency lines with OBSERVED/NOT OBSERVED lines = assessment_text.split('\n') observed_count = 0 for i, line in enumerate(lines): # Look for competency indicators followed by status if 'Competency' in line and i + 1 < len(lines): next_line = lines[i + 1] # Check if the status line indicates OBSERVED (not NOT OBSERVED) if 'OBSERVED' in next_line.upper() and 'NOT OBSERVED' not in next_line.upper(): observed_count += 1 # If still no matches, use a more robust pattern if observed_count == 0: # Count lines that say "OBSERVED" but not "NOT OBSERVED" for line in lines: # Clean line for better matching clean_line = line.strip().upper() if clean_line.startswith('STATUS:'): if 'NOT OBSERVED' in clean_line: continue elif 'OBSERVED' in clean_line: observed_count += 1 total_competencies = 18 percentage = (observed_count / total_competencies) * 100 if total_competencies > 0 else 0 # Professional color scheme with better contrast if percentage >= 85: level = "Exemplary" color = "#0F766E" # Deep teal elif percentage >= 70: level = "Proficient" color = "#1E40AF" # Professional blue elif percentage >= 50: level = "Developing" color = "#EA580C" # Professional orange else: level = "Needs Improvement" color = "#B91C1C" # Deep red return observed_count, total_competencies, percentage, level, color def generate_pdf_report(self, assessment_text): """Generate a PDF report from the assessment text""" 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 styles = getSampleStyleSheet() title_style = ParagraphStyle( 'CustomTitle', parent=styles['Heading1'], fontSize=24, textColor=HexColor('#111827'), # Darker gray for better readability spaceAfter=30, alignment=TA_CENTER ) heading_style = ParagraphStyle( 'CustomHeading', parent=styles['Heading2'], fontSize=14, textColor=HexColor('#1E40AF'), # Professional blue spaceAfter=12, spaceBefore=12, bold=True ) body_style = ParagraphStyle( 'CustomBody', parent=styles['BodyText'], fontSize=11, alignment=TA_JUSTIFY, spaceAfter=12 ) # Add title elements.append(Paragraph("CICE 2.0 Healthcare Team Assessment Report", title_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)) # 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('Competency'): # Competency item elements.append(Paragraph(f"{line}", body_style)) elif line.startswith('Status:') or line.startswith('Evidence:'): # Sub-items elements.append(Paragraph(line, 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("CICE 2.0 Healthcare Team Interaction Assessment\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(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 = """
π Listen to the 1-minute audio summary for key findings
π Download the PDF report for complete 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 CICE 2.0 Assessment (Text View)", lines=20, max_lines=30, interactive=False, placeholder="Detailed assessment will appear here..." ) # Footer gr.Markdown(""" --- ### About This Assessment This tool uses Google's Gemini AI to identify specific behavioral markers that indicate effective interprofessional collaboration in healthcare settings. The assessment focuses on observable actions such as: - Verbal role assignments ("I'll do compressions") - Recognition phrases ("Good catch on the allergy bracelet") - Plain language use instead of medical jargon - Pre-brief and post-brief team discussions **Output Files:** - π 1-minute audio summary (MP3 format) - π Complete PDF assessment report **Powered by Google Gemini 2.0 Flash | ZeroGPU on HuggingFace Spaces** """) # Auto-save video when recording stops with immediate status feedback 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 analyze_btn.click( fn=process_video, inputs=[video_input, resize_dropdown], outputs=[assessment_output, summary_output, audio_output, pdf_output], api_name="analyze" ) # Launch the app if __name__ == "__main__": demo.launch()