Multimodal Fusion for Speaker Diarization

Lip Movement Detection

lip_movement_by_time = {}
prev_mouth_roi = None

while cap.isOpened():
    ret, frame = cap.read()
    if not ret:
        break

    if frame_count % frame_skip == 0:
        timestamp = frame_count / fps
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        faces = face_cascade.detectMultiScale(gray, 1.1, 4)

        lip_moving = False
        for (x, y, w, h) in faces:
            # Extract mouth region (lower 40% of face)
            mouth_roi_y = y + int(h * 0.6)
            mouth_roi_h = int(h * 0.4)
            mouth_region = gray[mouth_roi_y:mouth_roi_y + mouth_roi_h, x:x + w]

            if mouth_region.size > 0:
                if prev_mouth_roi is not None and prev_mouth_roi.shape == mouth_region.shape:
                    # Calculate movement score
                    diff = cv2.absdiff(mouth_region, prev_mouth_roi)
                    movement_score = np.mean(diff)
                    if movement_score > 10:  # Threshold for movement
                        lip_moving = True
                prev_mouth_roi = mouth_region.copy()
                break

        lip_movement_by_time[timestamp] = lip_moving

    frame_count += 1

Temporal Alignment

Visual features need to be aligned with audio timestamps:

def get_faces_at_time(timestamp, tolerance=0.5):
    """Get number of faces at a given timestamp"""
    if not faces_by_time:
        return 0
    closest = min(faces_by_time.keys(),
                  key=lambda t: abs(t - timestamp),
                  default=None)
    if closest and abs(closest - timestamp) < tolerance:
        return faces_by_time[closest]
    return 0

def get_lip_movement_at_time(timestamp, tolerance=0.5):
    """Check if lips are moving at a given timestamp"""
    if not lip_movement_by_time:
        return False
    closest = min(lip_movement_by_time.keys(),
                  key=lambda t: abs(t - timestamp),
                  default=None)
    if closest and abs(closest - timestamp) < tolerance:
        return lip_movement_by_time[closest]
    return False

Fusion Strategies

1. Visual-Aided Speaker Assignment

Use visual features to help assign speakers to audio segments:

# For each diarization turn
for turn in diarization_turns:
    turn_center = (turn['start'] + turn['end']) / 2
    faces_at_turn = get_faces_at_time(turn_center)
    lip_moving = get_lip_movement_at_time(turn_center)

    # Use visual cues to refine speaker assignment
    if lip_moving and faces_at_turn > 0:
        # High confidence: speaker is visible and speaking
        turn['confidence'] = 'high'
    elif faces_at_turn > 0:
        # Medium confidence: speaker visible but no clear lip movement
        turn['confidence'] = 'medium'
    else:
        # Low confidence: no visual confirmation
        turn['confidence'] = 'low'

2. Face Count Validation

Use face count to validate speaker count:

# Count unique faces over video duration
unique_faces = set()
for timestamp in faces_by_time.keys():
    if faces_by_time[timestamp] > 0:
        # In a real implementation, you'd track individual faces
        unique_faces.add(timestamp)

# Validate predicted speaker count
if len(unique_faces) > 0:
    visual_speaker_count = max(faces_by_time.values())
    if abs(visual_speaker_count - predicted_speaker_count) > 1:
        # Warning: mismatch between audio and visual speaker counts
        print(f"Warning: Audio predicts {predicted_speaker_count} speakers, "
              f"but video shows up to {visual_speaker_count} faces")

3. Lip Movement Filtering

Filter out segments where no one appears to be speaking:

# Filter diarization turns based on lip movement
filtered_turns = []
for turn in diarization_turns:
    turn_start = turn['start']
    turn_end = turn['end']

    # Check if lips are moving during this turn
    has_lip_movement = any(
        get_lip_movement_at_time(t)
        for t in np.arange(turn_start, turn_end, 0.1)
    )

    if has_lip_movement:
        filtered_turns.append(turn)
    else:
        # Low confidence: no visual confirmation of speech
        turn['confidence'] = 'low'
        filtered_turns.append(turn)

Best Practices

1. Process frames efficiently: Don't process every frame; use frame_skip
2. Handle missing visual data: Always have fallback to audio-only
3. Temporal alignment: Ensure visual and audio timestamps are synchronized
4. Confidence scoring: Use visual features to assign confidence scores
5. Error handling: Video processing can fail; handle exceptions gracefully

Integration Example

# Complete pipeline
def multimodal_diarization(video_path, audio_path):
    # 1. Extract visual features
    faces_by_time, lip_movement_by_time = extract_visual_features(video_path)

    # 2. Run audio-based diarization
    audio_turns = run_audio_diarization(audio_path)

    # 3. Fuse visual and audio features
    for turn in audio_turns:
        turn_center = (turn['start'] + turn['end']) / 2
        turn['faces_detected'] = get_faces_at_time(turn_center)
        turn['lip_movement'] = get_lip_movement_at_time(turn_center)
        turn['on_screen'] = turn['faces_detected'] > 0

    return audio_turns

Limitations

• Visual features require video files (not just audio)
• Face detection may fail in poor lighting or angles
• Lip movement detection is approximate
• Processing video is computationally expensive

When to Skip Visual Features

• Audio-only files
• Poor video quality
• No faces visible
• Processing time constraints