Active speaker detection in Python on a CPU, tested against a podcast transcript

curl -O https://api.clipsdk.com/f/lib/demo/rocket-ranch-podcast.mp4
ffmpeg -ss 367 -t 24 -i rocket-ranch-podcast.mp4 -an -c:v libx264 -crf 18 two-shot.mp4
curl -LO https://github.com/opencv/opencv_zoo/raw/main/models/face_detection_yunet/face_detection_yunet_2023mar.onnx
pip install "opencv-python-headless==5.0.0.93"

import json
import cv2
import numpy as np

detector = cv2.FaceDetectorYN.create("face_detection_yunet_2023mar.onnx", "", (640, 360), 0.6, 0.3, 16)
video = cv2.VideoCapture("two-shot.mp4")
fps = video.get(cv2.CAP_PROP_FPS)
step = round(fps / 5)


def patch(gray, cx, cy, half_w, half_h):
    x0, y0 = int(max(cx - half_w, 0)), int(max(cy - half_h, 0))
    region = gray[y0 : int(cy + half_h) + 1, x0 : int(cx + half_w) + 1]
    return cv2.resize(region, (16, 16), interpolation=cv2.INTER_AREA).astype(np.float32)


# The mouth, and an eyes-to-nose patch whose motion is head motion and noise
def mouth_and_ref(gray, face):
    eyes = (face[4:6] + face[6:8]) / 2
    nose = face[8:10]
    mouth = (face[10:12] + face[12:14]) / 2
    span = max(2.0, float(np.linalg.norm(mouth - eyes)))
    return (
        patch(gray, mouth[0], mouth[1] + 0.15 * span, 0.5 * span, 0.5 * span),
        patch(gray, (eyes[0] + nose[0]) / 2, (eyes[1] + nose[1]) / 2, 0.7 * span, 0.35 * span),
    )


faces, previous, motion, samples = {}, {}, {}, []
frame_no = 0
while True:
    ok, frame = video.read()
    if not ok:
        break
    small = cv2.resize(frame, (640, 360))
    gray = cv2.cvtColor(small, cv2.COLOR_BGR2GRAY)

    # Every 0.2 s: close the window, then find the faces again
    if frame_no % step == 0:
        if frame_no > 0:
            samples.append({side: m / (r + 1) for side, (m, r) in motion.items()})
        _, found = detector.detect(small)
        found = sorted(found if found is not None else [], key=lambda f: -f[2] * f[3])[:2]
        faces = {("left" if f[0] + f[2] / 2 < 320 else "right"): f for f in found}
        motion, previous = {}, {}

    current = {side: mouth_and_ref(gray, f) for side, f in faces.items()}
    for side, (mouth, ref) in current.items():
        if side in previous:
            m, r = motion.get(side, (0.0, 0.0))
            motion[side] = (m + float(np.abs(mouth - previous[side][0]).mean()), r + float(np.abs(ref - previous[side][1]).mean()))
    previous = current
    frame_no += 1

json.dump(samples, open("activity.json", "w"))

# Mean activity per face over each 2-second window
for i in range(0, len(samples), 10):
    window = samples[i : i + 10]
    left = np.mean([s.get("left", 0.0) for s in window])
    right = np.mean([s.get("right", 0.0) for s in window])
    print(f"{i / 5:4.0f}-{(i + len(window)) / 5:<3.0f} left {left:.2f}  right {right:.2f}")

# $ python activity.py
#    0-2   left 0.82  right 0.39
#    2-4   left 0.99  right 0.37
#    4-6   left 0.94  right 0.37
#    6-8   left 0.96  right 0.25
#    8-10  left 1.64  right 0.16
#   10-12  left 1.09  right 0.29
#   12-14  left 1.13  right 0.38
#   14-16  left 1.05  right 0.44
#   16-18  left 0.70  right 0.54
#   18-20  left 0.81  right 0.92
#   20-22  left 0.95  right 1.78
#   22-24  left 0.91  right 0.65

import json

import numpy as np

samples = json.load(open("activity.json"))

# Smooth over 7 samples, then switch only on a 1.5x lead held for 1.2 s, at least 2 s after the last switch
kernel = np.ones(7) / 7
activity = {side: np.convolve([s.get(side, 0.0) for s in samples], kernel, mode="same") for side in ("left", "right")}
times = [(i + 1) / 5 for i in range(len(samples))]

speaker = max(activity, key=lambda side: activity[side][:10].mean())
switches, since, lead_from = [(0.0, speaker)], 0.0, None
for i, t in enumerate(times):
    other = "left" if speaker == "right" else "right"
    if activity[other][i] > 1.5 * activity[speaker][i]:
        lead_from = t if lead_from is None else lead_from
        if t - lead_from >= 1.2 and lead_from - since >= 2.0:
            speaker, since, lead_from = other, lead_from, None
            switches.append((since, speaker))
    else:
        lead_from = None

ends = [t for t, _ in switches[1:]] + [times[-1]]
for (start, side), end in zip(switches, ends):
    print(f"{start:5.1f} {end:5.1f}  {side}")

# $ python switch.py
#   0.0  20.8  left
#  20.8  23.8  right