Detecting Microexpressions with CNN for Clinical Diagnosis

# data_preprocessing.py
import cv2
import numpy as np
import dlib
from typing import Tuple, List
import os
from tqdm import tqdm
 
class FacialExpressionPreprocessor:
    """
    Preprocess facial images for microexpression detection
    
    Steps:
    1. Detect faces using Haar Cascade or dlib
    2. Extract facial landmarks (68 points)
    3. Align face to canonical pose
    4. Crop to face region only
    5. Normalize lighting
    6. Augment for training
    """
    
    def __init__(self):
        # Load face detector
        self.face_cascade = cv2.CascadeClassifier(
            cv2.data.haarcascades + 'haarcascade_frontalface_default.xml'
        )
        
        # Load facial landmark predictor
        self.landmark_predictor = dlib.shape_predictor(
            'models/shape_predictor_68_face_landmarks.dat'
        )
        
        # Emotion labels
        self.emotions = [
            'neutral',
            'happiness',
            'sadness',
            'surprise',
            'fear',
            'disgust',
            'anger'
        ]
        
    def detect_face(self, image: np.ndarray) -> Tuple[int, int, int, int]:
        """
        Detect face bounding box
        
        Returns:
            (x, y, w, h) or None if no face detected
        """
        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
        faces = self.face_cascade.detectMultiScale(
            gray,
            scaleFactor=1.1,
            minNeighbors=5,
            minSize=(48, 48)
        )
        
        if len(faces) == 0:
            return None
        
        # Return largest face
        return max(faces, key=lambda f: f[2] * f[3])
    
    def get_facial_landmarks(
        self,
        image: np.ndarray,
        bbox: Tuple[int, int, int, int]
    ) -> np.ndarray:
        """
        Extract 68 facial landmarks
        
        Returns:
            (68, 2) array of (x, y) coordinates
        """
        x, y, w, h = bbox
        
        # Convert to dlib rectangle
        rect = dlib.rectangle(x, y, x + w, y + h)
        
        # Detect landmarks
        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
        landmarks = self.landmark_predictor(gray, rect)
        
        # Convert to numpy array
        points = np.array([
            [landmarks.part(i).x, landmarks.part(i).y]
            for i in range(68)
        ])
        
        return points
    
    def align_face(
        self,
        image: np.ndarray,
        landmarks: np.ndarray
    ) -> np.ndarray:
        """
        Align face to canonical pose using eye positions
        """
        # Get eye centers
        left_eye = landmarks[36:42].mean(axis=0)
        right_eye = landmarks[42:48].mean(axis=0)
        
        # Calculate angle between eyes
        dy = right_eye[1] - left_eye[1]
        dx = right_eye[0] - left_eye[0]
        angle = np.degrees(np.arctan2(dy, dx))
        
        # Calculate center point between eyes
        eyes_center = ((left_eye + right_eye) / 2).astype(int)
        
        # Get rotation matrix
        M = cv2.getRotationMatrix2D(
            tuple(eyes_center),
            angle,
            scale=1.0
        )
        
        # Apply rotation
        aligned = cv2.warpAffine(
            image,
            M,
            (image.shape[1], image.shape[0])
        )
        
        return aligned
    
    def crop_face(
        self,
        image: np.ndarray,
        bbox: Tuple[int, int, int, int],
        padding: float = 0.2
    ) -> np.ndarray:
        """
        Crop image to face region with padding
        """
        x, y, w, h = bbox
        
        # Add padding
        pad_w = int(w * padding)
        pad_h = int(h * padding)
        
        x1 = max(0, x - pad_w)
        y1 = max(0, y - pad_h)
        x2 = min(image.shape[1], x + w + pad_w)
        y2 = min(image.shape[0], y + h + pad_h)
        
        cropped = image[y1:y2, x1:x2]
        
        # Resize to standard size
        cropped = cv2.resize(cropped, (224, 224))
        
        return cropped
    
    def normalize_lighting(self, image: np.ndarray) -> np.ndarray:
        """
        Normalize lighting using histogram equalization
        """
        # Convert to LAB color space
        lab = cv2.cvtColor(image, cv2.COLOR_BGR2LAB)
        
        # Split channels
        l, a, b = cv2.split(lab)
        
        # Apply CLAHE to L channel
        clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
        l = clahe.apply(l)
        
        # Merge channels
        lab = cv2.merge([l, a, b])
        
        # Convert back to BGR
        normalized = cv2.cvtColor(lab, cv2.COLOR_LAB2BGR)
        
        return normalized
    
    def augment_image(self, image: np.ndarray) -> List[np.ndarray]:
        """
        Data augmentation for training
        
        Returns:
            List of augmented images
        """
        augmented = [image]
        
        # Horizontal flip
        augmented.append(cv2.flip(image, 1))
        
        # Slight rotations
        for angle in [-10, 10]:
            M = cv2.getRotationMatrix2D(
                (image.shape[1] // 2, image.shape[0] // 2),
                angle,
                1.0
            )
            rotated = cv2.warpAffine(image, M, (image.shape[1], image.shape[0]))
            augmented.append(rotated)
        
        # Brightness variations
        for beta in [-20, 20]:
            adjusted = cv2.convertScaleAbs(image, alpha=1.0, beta=beta)
            augmented.append(adjusted)
        
        return augmented
    
    def preprocess_dataset(
        self,
        data_dir: str,
        output_dir: str,
        augment: bool = True
    ):
        """
        Preprocess entire dataset
        
        Directory structure:
        data_dir/
            emotion_0/
                img1.jpg
                img2.jpg
            emotion_1/
                ...
        """
        os.makedirs(output_dir, exist_ok=True)
        
        for emotion_idx, emotion in enumerate(self.emotions):
            emotion_dir = os.path.join(data_dir, emotion)
            output_emotion_dir = os.path.join(output_dir, emotion)
            os.makedirs(output_emotion_dir, exist_ok=True)
            
            if not os.path.exists(emotion_dir):
                continue
            
            image_files = [f for f in os.listdir(emotion_dir) 
                          if f.endswith(('.jpg', '.png', '.jpeg'))]
            
            print(f"\nProcessing {emotion} ({len(image_files)} images)...")
            
            for img_file in tqdm(image_files):
                try:
                    # Load image
                    img_path = os.path.join(emotion_dir, img_file)
                    image = cv2.imread(img_path)
                    
                    if image is None:
                        continue
                    
                    # Detect face
                    bbox = self.detect_face(image)
                    if bbox is None:
                        continue
                    
                    # Get landmarks
                    landmarks = self.get_facial_landmarks(image, bbox)
                    
                    # Align face
                    aligned = self.align_face(image, landmarks)
                    
                    # Crop face
                    cropped = self.crop_face(aligned, bbox)
                    
                    # Normalize lighting
                    normalized = self.normalize_lighting(cropped)
                    
                    # Save processed image
                    base_name = os.path.splitext(img_file)[0]
                    output_path = os.path.join(
                        output_emotion_dir,
                        f"{base_name}_processed.jpg"
                    )
                    cv2.imwrite(output_path, normalized)
                    
                    # Augment if training
                    if augment:
                        augmented_images = self.augment_image(normalized)
                        for aug_idx, aug_img in enumerate(augmented_images[1:]):
                            aug_path = os.path.join(
                                output_emotion_dir,
                                f"{base_name}_aug{aug_idx}.jpg"
                            )
                            cv2.imwrite(aug_path, aug_img)
                    
                except Exception as e:
                    print(f"Error processing {img_file}: {e}")
                    continue
        
        print(f"\nPreprocessing complete! Output: {output_dir}")
 
# Usage
preprocessor = FacialExpressionPreprocessor()
preprocessor.preprocess_dataset(
    data_dir='data/raw_faces',
    output_dir='data/processed_faces',
    augment=True
)

# emotion_cnn.py
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import numpy as np
 
class EmotionCNN(keras.Model):
    """
    CNN for microexpression detection
    
    Architecture:
    - 4 convolutional blocks with increasing filters
    - Batch normalization and dropout for regularization
    - Global average pooling to reduce parameters
    - Dense layers with softmax for 7-emotion classification
    
    Optimized for:
    - Real-time inference on mobile devices
    - Generalization across diverse faces
    - Robustness to lighting and pose variations
    """
    
    def __init__(
        self,
        num_emotions: int = 7,
        input_shape: Tuple[int, int, int] = (224, 224, 3),
        dropout_rate: float = 0.5
    ):
        super(EmotionCNN, self).__init__()
        
        self.num_emotions = num_emotions
        
        # Block 1: Initial feature extraction
        self.conv1_1 = layers.Conv2D(64, (3, 3), activation='relu', padding='same')
        self.conv1_2 = layers.Conv2D(64, (3, 3), activation='relu', padding='same')
        self.bn1 = layers.BatchNormalization()
        self.pool1 = layers.MaxPooling2D((2, 2))
        self.dropout1 = layers.Dropout(0.25)
        
        # Block 2: Mid-level features
        self.conv2_1 = layers.Conv2D(128, (3, 3), activation='relu', padding='same')
        self.conv2_2 = layers.Conv2D(128, (3, 3), activation='relu', padding='same')
        self.bn2 = layers.BatchNormalization()
        self.pool2 = layers.MaxPooling2D((2, 2))
        self.dropout2 = layers.Dropout(0.25)
        
        # Block 3: High-level features
        self.conv3_1 = layers.Conv2D(256, (3, 3), activation='relu', padding='same')
        self.conv3_2 = layers.Conv2D(256, (3, 3), activation='relu', padding='same')
        self.conv3_3 = layers.Conv2D(256, (3, 3), activation='relu', padding='same')
        self.bn3 = layers.BatchNormalization()
        self.pool3 = layers.MaxPooling2D((2, 2))
        self.dropout3 = layers.Dropout(0.25)
        
        # Block 4: Deep features
        self.conv4_1 = layers.Conv2D(512, (3, 3), activation='relu', padding='same')
        self.conv4_2 = layers.Conv2D(512, (3, 3), activation='relu', padding='same')
        self.conv4_3 = layers.Conv2D(512, (3, 3), activation='relu', padding='same')
        self.bn4 = layers.BatchNormalization()
        self.pool4 = layers.MaxPooling2D((2, 2))
        self.dropout4 = layers.Dropout(0.25)
        
        # Global pooling (reduces parameters vs flatten)
        self.global_pool = layers.GlobalAveragePooling2D()
        
        # Dense layers
        self.dense1 = layers.Dense(512, activation='relu')
        self.bn5 = layers.BatchNormalization()
        self.dropout5 = layers.Dropout(dropout_rate)
        
        self.dense2 = layers.Dense(256, activation='relu')
        self.dropout6 = layers.Dropout(dropout_rate)
        
        # Output layer
        self.output_layer = layers.Dense(num_emotions, activation='softmax')
    
    def call(self, inputs, training=False):
        """Forward pass"""
        # Block 1
        x = self.conv1_1(inputs)
        x = self.conv1_2(x)
        x = self.bn1(x, training=training)
        x = self.pool1(x)
        x = self.dropout1(x, training=training)
        
        # Block 2
        x = self.conv2_1(x)
        x = self.conv2_2(x)
        x = self.bn2(x, training=training)
        x = self.pool2(x)
        x = self.dropout2(x, training=training)
        
        # Block 3
        x = self.conv3_1(x)
        x = self.conv3_2(x)
        x = self.conv3_3(x)
        x = self.bn3(x, training=training)
        x = self.pool3(x)
        x = self.dropout3(x, training=training)
        
        # Block 4
        x = self.conv4_1(x)
        x = self.conv4_2(x)
        x = self.conv4_3(x)
        x = self.bn4(x, training=training)
        x = self.pool4(x)
        x = self.dropout4(x, training=training)
        
        # Global pooling
        x = self.global_pool(x)
        
        # Dense layers
        x = self.dense1(x)
        x = self.bn5(x, training=training)
        x = self.dropout5(x, training=training)
        
        x = self.dense2(x)
        x = self.dropout6(x, training=training)
        
        # Output
        output = self.output_layer(x)
        
        return output
    
    def predict_emotion(self, image: np.ndarray) -> Tuple[str, float, np.ndarray]:
        """
        Predict emotion from single image
        
        Args:
            image: (224, 224, 3) RGB image
        
        Returns:
            emotion_label: Predicted emotion string
            confidence: Confidence score [0, 1]
            probabilities: (7,) array of probabilities for each emotion
        """
        # Preprocess
        if image.shape != (224, 224, 3):
            image = cv2.resize(image, (224, 224))
        
        # Normalize
        image = image.astype(np.float32) / 255.0
        
        # Add batch dimension
        image_batch = np.expand_dims(image, axis=0)
        
        # Predict
        probabilities = self(image_batch, training=False)[0].numpy()
        
        # Get prediction
        emotion_idx = np.argmax(probabilities)
        confidence = probabilities[emotion_idx]
        
        emotions = ['neutral', 'happiness', 'sadness', 'surprise', 
                   'fear', 'disgust', 'anger']
        emotion_label = emotions[emotion_idx]
        
        return emotion_label, confidence, probabilities
 
# Build model
model = EmotionCNN(num_emotions=7)
 
# Compile
model.compile(
    optimizer=keras.optimizers.Adam(learning_rate=1e-4),
    loss='categorical_crossentropy',
    metrics=['accuracy', keras.metrics.TopKCategoricalAccuracy(k=2)]
)
 
model.build((None, 224, 224, 3))
model.summary()

# train.py
import tensorflow as tf
from tensorflow import keras
import numpy as np
from sklearn.utils.class_weight import compute_class_weight
import wandb
 
def create_dataset(data_dir: str, batch_size: int = 32, augment: bool = True):
    """Create TensorFlow dataset with augmentation"""
    
    # Load images and labels
    datagen = keras.preprocessing.image.ImageDataGenerator(
        rescale=1./255,
        rotation_range=20 if augment else 0,
        width_shift_range=0.2 if augment else 0,
        height_shift_range=0.2 if augment else 0,
        horizontal_flip=True if augment else False,
        zoom_range=0.2 if augment else 0,
        fill_mode='nearest'
    )
    
    dataset = datagen.flow_from_directory(
        data_dir,
        target_size=(224, 224),
        batch_size=batch_size,
        class_mode='categorical',
        shuffle=True
    )
    
    return dataset
 
def compute_class_weights(train_dataset):
    """Compute class weights to handle imbalance"""
    labels = train_dataset.classes
    class_weights = compute_class_weight(
        'balanced',
        classes=np.unique(labels),
        y=labels
    )
    
    class_weight_dict = dict(enumerate(class_weights))
    print("Class weights:", class_weight_dict)
    
    return class_weight_dict
 
def train_model(
    model: EmotionCNN,
    train_dataset,
    val_dataset,
    epochs: int = 100
):
    """Train emotion detection model"""
    
    # Initialize W&B
    wandb.init(
        project="emotion-detection-clinical",
        config={
            "epochs": epochs,
            "batch_size": 32,
            "learning_rate": 1e-4,
            "architecture": "EmotionCNN"
        }
    )
    
    # Compute class weights
    class_weights = compute_class_weights(train_dataset)
    
    # Callbacks
    callbacks = [
        keras.callbacks.ModelCheckpoint(
            'checkpoints/emotion_cnn_best.h5',
            save_best_only=True,
            monitor='val_accuracy',
            mode='max'
        ),
        keras.callbacks.EarlyStopping(
            monitor='val_loss',
            patience=10,
            restore_best_weights=True
        ),
        keras.callbacks.ReduceLROnPlateau(
            monitor='val_loss',
            factor=0.5,
            patience=5,
            min_lr=1e-7
        ),
        wandb.keras.WandbCallback(save_model=False)
    ]
    
    # Train
    history = model.fit(
        train_dataset,
        validation_data=val_dataset,
        epochs=epochs,
        class_weight=class_weights,
        callbacks=callbacks
    )
    
    wandb.finish()
    
    return history
 
if __name__ == "__main__":
    # Load datasets
    train_dataset = create_dataset('data/processed_faces/train', batch_size=32, augment=True)
    val_dataset = create_dataset('data/processed_faces/val', batch_size=32, augment=False)
    
    # Build model
    model = EmotionCNN(num_emotions=7)
    model.compile(
        optimizer=keras.optimizers.Adam(1e-4),
        loss='categorical_crossentropy',
        metrics=['accuracy']
    )
    
    # Train
    history = train_model(model, train_dataset, val_dataset, epochs=100)
    
    # Evaluate on test set
    test_dataset = create_dataset('data/processed_faces/test', batch_size=32, augment=False)
    test_loss, test_acc = model.evaluate(test_dataset)
    
    print(f"\nTest Accuracy: {test_acc*100:.2f}%")
    
    # Save final model
    model.save('models/emotion_cnn_final.h5')
    print("Model saved!")

# voice_stress_analyzer.py
import librosa
import numpy as np
from tensorflow import keras
from tensorflow.keras import layers
 
class VoiceStressAnalyzer:
    """
    Analyze voice for stress markers
    
    Features extracted:
    - MFCC (Mel-frequency cepstral coefficients)
    - Pitch variation
    - Speech rate
    - Energy/amplitude
    - Spectral features
    """
    
    def __init__(self, model_path: str = None):
        if model_path:
            self.model = keras.models.load_model(model_path)
        else:
            self.model = self._build_model()
    
    def _build_model(self):
        """LSTM model for voice stress detection"""
        model = keras.Sequential([
            layers.Input(shape=(None, 40)),  # Variable length, 40 MFCC features
            
            layers.LSTM(128, return_sequences=True),
            layers.Dropout(0.3),
            
            layers.LSTM(64),
            layers.Dropout(0.3),
            
            layers.Dense(32, activation='relu'),
            layers.Dropout(0.3),
            
            layers.Dense(1, activation='sigmoid')  # Stress probability
        ])
        
        model.compile(
            optimizer='adam',
            loss='binary_crossentropy',
            metrics=['accuracy']
        )
        
        return model
    
    def extract_features(self, audio_path: str) -> np.ndarray:
        """
        Extract audio features for stress detection
        
        Returns:
            (time_steps, 40) MFCC features
        """
        # Load audio
        y, sr = librosa.load(audio_path, sr=22050)
        
        # Extract MFCCs
        mfccs = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=40)
        
        # Transpose to (time, features)
        mfccs = mfccs.T
        
        return mfccs
    
    def predict_stress(self, audio_path: str) -> Tuple[float, Dict]:
        """
        Predict stress level from audio
        
        Returns:
            stress_score: 0-1 (0=calm, 1=stressed)
            features: Dict of extracted features
        """
        # Extract features
        mfccs = self.extract_features(audio_path)
        
        # Predict
        stress_score = self.model.predict(np.expand_dims(mfccs, 0))[0][0]
        
        # Additional features for interpretation
        y, sr = librosa.load(audio_path, sr=22050)
        
        features = {
            'stress_score': float(stress_score),
            'pitch_mean': float(librosa.feature.zero_crossing_rate(y).mean()),
            'energy_mean': float(librosa.feature.rms(y=y).mean()),
            'speech_rate': self._estimate_speech_rate(y, sr)
        }
        
        return stress_score, features
    
    def _estimate_speech_rate(self, y: np.ndarray, sr: int) -> float:
        """Estimate syllables per second"""
        # Simple onset detection
        onset_env = librosa.onset.onset_strength(y=y, sr=sr)
        onsets = librosa.onset.onset_detect(onset_envelope=onset_env, sr=sr)
        
        duration = len(y) / sr
        speech_rate = len(onsets) / duration
        
        return speech_rate

# congruence_analyzer.py
import numpy as np
from typing import Dict, List, Tuple
 
class CongruenceAnalyzer:
    """
    Analyze emotional congruence across modalities
    
    Compares:
    1. Facial expression (CNN)
    2. Voice stress (LSTM)
    3. Verbal sentiment (NLP)
    
    Flags incongruence when modalities disagree
    """
    
    def __init__(
        self,
        facial_model,
        voice_model,
        sentiment_model
    ):
        self.facial_model = facial_model
        self.voice_model = voice_model
        self.sentiment_model = sentiment_model
        
        # Emotion mappings to valence/arousal
        self.emotion_valence = {
            'happiness': 0.8,
            'surprise': 0.3,
            'neutral': 0.0,
            'fear': -0.7,
            'sadness': -0.8,
            'anger': -0.6,
            'disgust': -0.7
        }
        
        self.emotion_arousal = {
            'happiness': 0.6,
            'surprise': 0.9,
            'neutral': 0.0,
            'fear': 0.8,
            'sadness': -0.5,
            'anger': 0.8,
            'disgust': 0.5
        }
    
    def analyze_congruence(
        self,
        video_frame: np.ndarray,
        audio_segment: str,
        transcript: str
    ) -> Dict:
        """
        Analyze emotional congruence
        
        Returns:
            {
                'facial_emotion': str,
                'voice_stress': float,
                'text_sentiment': float,
                'congruence_score': float (0-1, 1=congruent),
                'incongruence_type': str or None,
                'alert_therapist': bool
            }
        """
        # Analyze each modality
        facial_emotion, facial_conf, _ = self.facial_model.predict_emotion(video_frame)
        voice_stress, _ = self.voice_model.predict_stress(audio_segment)
        text_sentiment = self._analyze_sentiment(transcript)
        
        # Map to valence/arousal space
        facial_valence = self.emotion_valence[facial_emotion]
        facial_arousal = self.emotion_arousal[facial_emotion]
        
        # Voice stress maps to high arousal
        voice_arousal = voice_stress
        
        # Text sentiment maps to valence
        text_valence = text_sentiment
        
        # Calculate congruence
        # Congruence = agreement across modalities
        valence_agreement = 1 - abs(facial_valence - text_valence) / 2
        arousal_agreement = 1 - abs(facial_arousal - voice_arousal) / 2
        
        congruence_score = (valence_agreement + arousal_agreement) / 2
        
        # Detect incongruence patterns
        incongruence_type = None
        alert_therapist = False
        
        # Pattern 1: Says positive, looks negative
        if text_valence > 0.3 and facial_valence < -0.3:
            incongruence_type = "verbal_facial_mismatch_positive_mask"
            alert_therapist = True
        
        # Pattern 2: Says calm, voice shows stress
        if text_valence > 0 and voice_stress > 0.7:
            incongruence_type = "verbal_voice_mismatch_suppressed_stress"
            alert_therapist = True
        
        # Pattern 3: Neutral face but high stress voice
        if facial_emotion == 'neutral' and voice_stress > 0.7:
            incongruence_type = "emotional_suppression"
            alert_therapist = True
        
        # Pattern 4: High confidence in negative emotion but positive words
        if facial_conf > 0.8 and facial_valence < -0.5 and text_valence > 0.3:
            incongruence_type = "strong_negative_emotion_denied"
            alert_therapist = True
        
        return {
            'facial_emotion': facial_emotion,
            'facial_confidence': facial_conf,
            'voice_stress': voice_stress,
            'text_sentiment': text_sentiment,
            'congruence_score': congruence_score,
            'incongruence_type': incongruence_type,
            'alert_therapist': alert_therapist,
            'timestamp': None  # To be filled by caller
        }
    
    def _analyze_sentiment(self, text: str) -> float:
        """
        Analyze text sentiment
        
        Returns:
            sentiment: -1 (negative) to 1 (positive)
        """
        # Use pretrained sentiment model (BERT/RoBERTa)
        # Placeholder implementation
        from transformers import pipeline
        
        sentiment_analyzer = pipeline(
            "sentiment-analysis",
            model="distilbert-base-uncased-finetuned-sst-2-english"
        )
        
        result = sentiment_analyzer(text)[0]
        score = result['score']
        
        # Convert to -1 to 1 scale
        if result['label'] == 'NEGATIVE':
            sentiment = -score
        else:
            sentiment = score
        
        return sentiment
    
    def analyze_session_timeline(
        self,
        congruence_results: List[Dict]
    ) -> Dict:
        """
        Analyze entire session for patterns
        
        Returns:
            {
                'avg_congruence': float,
                'num_alerts': int,
                'emotional_trajectory': List[float],
                'key_moments': List[Dict]
            }
        """
        avg_congruence = np.mean([r['congruence_score'] for r in congruence_results])
        num_alerts = sum([r['alert_therapist'] for r in congruence_results])
        
        # Track emotional valence over time
        emotional_trajectory = [
            self.emotion_valence[r['facial_emotion']]
            for r in congruence_results
        ]
        
        # Find key moments (low congruence spikes)
        key_moments = []
        for i, result in enumerate(congruence_results):
            if result['congruence_score'] < 0.5 and result['alert_therapist']:
                key_moments.append({
                    'timestamp': result.get('timestamp', i),
                    'type': result['incongruence_type'],
                    'score': result['congruence_score']
                })
        
        return {
            'avg_congruence': avg_congruence,
            'num_alerts': num_alerts,
            'emotional_trajectory': emotional_trajectory,
            'key_moments': key_moments
        }

// ClinicalDashboard.tsx
import React, { useState, useEffect } from 'react';
import { Line } from 'react-chartjs-2';
import axios from 'axios';
 
interface CongruenceData {
  timestamp: number;
  facialEmotion: string;
  voiceStress: number;
  textSentiment: number;
  congruenceScore: number;
  alertTherapist: boolean;
  incongruenceType?: string;
}
 
export const ClinicalDashboard: React.FC<{ sessionId: string }> = ({ sessionId }) => {
  const [congruenceData, setCongruenceData] = useState<CongruenceData[]>([]);
  const [sessionStats, setSessionStats] = useState<any>(null);
  const [loading, setLoading] = useState(true);
 
  useEffect(() => {
    loadSessionData();
    
    // Real-time updates every 5 seconds
    const interval = setInterval(loadSessionData, 5000);
    return () => clearInterval(interval);
  }, [sessionId]);
 
  const loadSessionData = async () => {
    try {
      const response = await axios.get(
        `https://api.congruence.health/sessions/${sessionId}/analysis`,
        {
          headers: {
            'Authorization': `Bearer ${localStorage.getItem('token')}`,
            'X-HIPAA-Consent': 'true'
          }
        }
      );
 
      setCongruenceData(response.data.timeline);
      setSessionStats(response.data.stats);
      setLoading(false);
    } catch (error) {
      console.error('Failed to load session data:', error);
    }
  };
 
  // Prepare chart data
  const chartData = {
    labels: congruenceData.map(d => new Date(d.timestamp).toLocaleTimeString()),
    datasets: [
      {
        label: 'Emotional Congruence',
        data: congruenceData.map(d => d.congruenceScore),
        borderColor: 'rgb(75, 192, 192)',
        backgroundColor: 'rgba(75, 192, 192, 0.2)',
        tension: 0.4
      },
      {
        label: 'Voice Stress',
        data: congruenceData.map(d => d.voiceStress),
        borderColor: 'rgb(255, 99, 132)',
        backgroundColor: 'rgba(255, 99, 132, 0.2)',
        tension: 0.4
      }
    ]
  };
 
  if (loading) {
    return <div>Loading session analysis...</div>;
  }
 
  return (
    <div className="clinical-dashboard">
      <div className="header">
        <h2>Session Analysis - Real-time</h2>
        <div className="status">
          {sessionStats && (
            <>
              <span className="stat">
                Avg Congruence: {(sessionStats.avgCongruence * 100).toFixed(1)}%
              </span>
              <span className="stat alerts">
                {sessionStats.numAlerts} Incongruence Alerts
              </span>
            </>
          )}
        </div>
      </div>
 
      {/* Emotional Timeline */}
      <div className="chart-container">
        <h3>Emotional Timeline</h3>
        <Line data={chartData} options={{
          responsive: true,
          scales: {
            y: {
              min: 0,
              max: 1,
              title: { display: true, text: 'Score (0-1)' }
            }
          }
        }} />
      </div>
 
      {/* Incongruence Alerts */}
      <div className="alerts-panel">
        <h3>Incongruence Alerts</h3>
        {congruenceData.filter(d => d.alertTherapist).map((alert, idx) => (
          <div key={idx} className="alert-card">
            <div className="alert-time">
              {new Date(alert.timestamp).toLocaleTimeString()}
            </div>
            <div className="alert-type">
              {formatIncongruenceType(alert.incongruenceType)}
            </div>
            <div className="alert-details">
              <span>Facial: {alert.facialEmotion}</span>
              <span>Voice Stress: {(alert.voiceStress * 100).toFixed(0)}%</span>
              <span>Sentiment: {(alert.textSentiment * 100).toFixed(0)}%</span>
            </div>
            <div className="alert-confidence">
              Congruence: {(alert.congruenceScore * 100).toFixed(1)}%
            </div>
          </div>
        ))}
      </div>
 
      {/* Key Moments */}
      {sessionStats && sessionStats.keyMoments.length > 0 && (
        <div className="key-moments">
          <h3>Key Moments to Review</h3>
          {sessionStats.keyMoments.map((moment: any, idx: number) => (
            <div key={idx} className="moment-card">
              <button onClick={() => seekToTimestamp(moment.timestamp)}>
                {formatTime(moment.timestamp)}
              </button>
              <span>{moment.type}</span>
              <span className="score">{(moment.score * 100).toFixed(0)}%</span>
            </div>
          ))}
        </div>
      )}
    </div>
  );
};
 
function formatIncongruenceType(type?: string): string {
  if (!type) return 'Unknown';
  
  const map: Record<string, string> = {
    'verbal_facial_mismatch_positive_mask': 'Patient masking negative emotions',
    'verbal_voice_mismatch_suppressed_stress': 'Suppressed stress detected',
    'emotional_suppression': 'Emotional suppression',
    'strong_negative_emotion_denied': 'Strong negative emotion denied verbally'
  };
  
  return map[type] || type;
}
 
function formatTime(ms: number): string {
  const seconds = Math.floor(ms / 1000);
  const minutes = Math.floor(seconds / 60);
  const remainingSeconds = seconds % 60;
  return `${minutes}:${remainingSeconds.toString().padStart(2, '0')}`;
}
 
function seekToTimestamp(timestamp: number) {
  // Seek video player to timestamp
  const videoPlayer = document.getElementById('session-video') as HTMLVideoElement;
  if (videoPlayer) {
    videoPlayer.currentTime = timestamp / 1000;
  }
}

# Handle low-quality frames
def preprocess_clinical_frame(frame):
    # Denoise
    denoised = cv2.fastNlMeansDenoisingColored(frame)
    
    # Enhance contrast
    lab = cv2.cvtColor(denoised, cv2.COLOR_BGR2LAB)
    l, a, b = cv2.split(lab)
    clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8, 8))
    l = clahe.apply(l)
    enhanced = cv2.merge([l, a, b])
    enhanced = cv2.cvtColor(enhanced, cv2.COLOR_LAB2BGR)
    
    return enhanced

# HIPAA-compliant processing
def process_session_hipaa_compliant(video_path, session_id):
    # Process frame-by-frame without storing
    cap = cv2.VideoCapture(video_path)
    
    results = []
    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break
        
        # Analyze frame
        analysis = congruence_analyzer.analyze_congruence(
            frame, audio_segment, transcript
        )
        
        # Store only numerical data (no video)
        results.append({
            'timestamp': cap.get(cv2.CAP_PROP_POS_MSEC),
            'emotion': analysis['facial_emotion'],
            'stress': analysis['voice_stress'],
            'congruence': analysis['congruence_score']
        })
    
    cap.release()
    
    # Delete original video
    os.remove(video_path)
    
    # Save encrypted results only
    save_encrypted(results, session_id)

# Convert to TensorFlow Lite
converter = tf.lite.TFLiteConverter.from_keras_model(model)
converter.optimizations = [tf.lite.Optimize.DEFAULT]
converter.target_spec.supported_types = [tf.int8]
 
tflite_model = converter.convert()
 
# Save
with open('emotion_model_quantized.tflite', 'wb') as f:
    f.write(tflite_model)

def detect_multiple_faces(frame):
    faces = face_detector.detect_multi_scale(frame)
    
    emotions = []
    for (x, y, w, h) in faces:
        face_crop = frame[y:y+h, x:x+w]
        emotion, conf, _ = model.predict_emotion(face_crop)
        emotions.append({
            'bbox': (x, y, w, h),
            'emotion': emotion,
            'confidence': conf
        })
    
    return emotions

class EmotionTrajectoryPredictor:
    def predict_future_state(self, emotion_history: List[str]) -> Dict:
        # Use LSTM to predict next 30 seconds of emotional state
        future_emotions = lstm_predictor.predict(emotion_history)
        
        # Alert if trending toward disengagement
        if 'neutral' in future_emotions[-5:]:
            return {'alert': 'patient_disengaging', 'confidence': 0.82}

models = {
    'western': load_model('emotion_cnn_western.h5'),
    'east_asian': load_model('emotion_cnn_east_asian.h5'),
    'middle_eastern': load_model('emotion_cnn_middle_eastern.h5')
}
 
# Select based on patient demographics
emotion = models[patient_culture].predict_emotion(frame)

def generate_clinical_note(session_analysis):
    note = f"""
    Session Date: {session_analysis['date']}
    Duration: {session_analysis['duration']} minutes
    
    Emotional State Summary:
    - Average Congruence: {session_analysis['avg_congruence']:.1%}
    - Predominant Emotion: {session_analysis['primary_emotion']}
    - Stress Level: {session_analysis['avg_stress']:.1%}
    
    Key Observations:
    {format_key_moments(session_analysis['key_moments'])}
    
    Incongruence Alerts: {session_analysis['num_alerts']}
    {format_alerts(session_analysis['alerts'])}
    
    Clinical Recommendations:
    {generate_recommendations(session_analysis)}
    """
    
    return note