From 1a836086ca8da6a3fb787cb59d67edff5417589a Mon Sep 17 00:00:00 2001 From: TomOdellSheetMusic Date: Wed, 12 Aug 2026 21:11:24 +0200 Subject: [PATCH] adds per user voice activity support related to https://github.com/SableClient/SableCall/issues/9 --- sdk/main.ts | 142 ++++++++++----- src/room/InCallView.tsx | 17 +- src/state/CallViewModel/CallViewModel.ts | 23 +++ src/state/media/RemoteUserMediaViewModel.ts | 14 ++ src/state/media/UserMediaViewModel.ts | 23 +++ src/state/media/observeAudioLevel.test.ts | 185 ++++++++++++++++++++ src/state/media/observeAudioLevel.ts | 111 ++++++++++++ src/widget.ts | 1 + 8 files changed, 469 insertions(+), 47 deletions(-) create mode 100644 src/state/media/observeAudioLevel.test.ts create mode 100644 src/state/media/observeAudioLevel.ts diff --git a/sdk/main.ts b/sdk/main.ts index a001af65c0..4e2d4a0453 100644 --- a/sdk/main.ts +++ b/sdk/main.ts @@ -84,6 +84,8 @@ interface MatrixRTCSdk { connection: Connection | null; membership: CallMembership; participant: LocalParticipant | RemoteParticipant | null; + speaking: boolean; + audioLevel: number; }[] >; /** @@ -93,7 +95,17 @@ interface MatrixRTCSdk { connection: Connection | null; membership: CallMembership; participant: LocalParticipant | null; + speaking: boolean; + audioLevel: number; } | null>; + activeSpeakers$: Behavior< + { + connection: Connection | null; + membership: CallMembership; + participant: LocalParticipant | RemoteParticipant | null; + audioLevel: number; + }[] + >; /** Use the LocalMemberConnectionState returned from `join` for a more detailed connection state */ connected$: Behavior; sendData?: (data: unknown) => Promise; @@ -302,6 +314,87 @@ export async function createMatrixRTCSdk( logger.info("createMatrixRTCSdk done"); + const voiceActivityForMember$ = (member: { + userId: string; + membership$: Behavior; + }): Observable<{ speaking: boolean; audioLevel: number }> => + combineLatest([member.membership$, callViewModel.userMedia$]).pipe( + switchMap(([membership, mediaItems]) => { + const media = mediaItems.find( + (m) => + m.userId === member.userId && + m.id.startsWith(`${member.userId}:${membership.deviceId}:`), + ); + return media + ? combineLatest([media.voiceActivity$, media.audioLevel$]).pipe( + map(([speaking, audioLevel]) => ({ speaking, audioLevel })), + ) + : of({ speaking: false, audioLevel: 0 }); + }), + ); + + const localMember$ = scope.behavior( + callViewModel.localMatrixLivekitMember$.pipe( + tap((member) => logger.info("localMatrixLivekitMember$ next: ", member)), + switchMap((member) => { + if (member === null) return of(null); + return combineLatest([ + member.connection$, + member.membership$, + member.participant.value$, + voiceActivityForMember$(member), + ]).pipe( + map(([connection, membership, participant, voice]) => ({ + connection, + membership, + participant, + speaking: voice.speaking, + audioLevel: voice.audioLevel, + })), + ); + }), + tap((member) => logger.info("localMember$ next: ", member)), + ), + ); + + const remoteMembers$ = scope.behavior( + callViewModel.remoteMatrixLivekitMembers$.pipe( + switchMap((members) => { + const listOfMemberObservables = members.map((member) => + combineLatest([ + member.connection$, + member.membership$, + member.participant.value$, + voiceActivityForMember$(member), + ]).pipe( + map(([connection, membership, participant, voice]) => ({ + connection, + membership, + participant, + speaking: voice.speaking, + audioLevel: voice.audioLevel, + })), + // using shareReplay instead of a Behavior here because the behavior would need + // a tricky scope.end() setup. + shareReplay({ bufferSize: 1, refCount: true }), + ), + ); + return combineLatest(listOfMemberObservables); + }), + ), + [], + ); + const activeSpeakers$ = scope.behavior( + combineLatest([localMember$, remoteMembers$]).pipe( + map(([local, remote]) => + [...(local && local.speaking ? [local] : []), ...remote].filter( + (m) => m.speaking, + ), + ), + ), + [], + ); + return { join: (): void => { // first lets try making the widget sticky @@ -317,53 +410,10 @@ export async function createMatrixRTCSdk( scope.end(); }, data$, - localMember$: scope.behavior( - callViewModel.localMatrixLivekitMember$.pipe( - tap((member) => - logger.info("localMatrixLivekitMember$ next: ", member), - ), - switchMap((member) => { - if (member === null) return of(null); - return combineLatest([ - member.connection$, - member.membership$, - member.participant.value$, - ]).pipe( - map(([connection, membership, participant]) => ({ - connection, - membership, - participant, - })), - ); - }), - tap((member) => logger.info("localMember$ next: ", member)), - ), - ), + localMember$, connected$: callViewModel.connected$, - remoteMembers$: scope.behavior( - callViewModel.remoteMatrixLivekitMembers$.pipe( - switchMap((members) => { - const listOfMemberObservables = members.map((member) => - combineLatest([ - member.connection$, - member.membership$, - member.participant.value$, - ]).pipe( - map(([connection, membership, participant]) => ({ - connection, - membership, - participant, - })), - // using shareReplay instead of a Behavior here because the behavior would need - // a tricky scope.end() setup. - shareReplay({ bufferSize: 1, refCount: true }), - ), - ); - return combineLatest(listOfMemberObservables); - }), - ), - [], - ), + remoteMembers$, + activeSpeakers$, sendData, sendRoomMessage, }; diff --git a/src/room/InCallView.tsx b/src/room/InCallView.tsx index 58b378aec2..2a5e7c0cf1 100644 --- a/src/room/InCallView.tsx +++ b/src/room/InCallView.tsx @@ -28,7 +28,7 @@ import { useTranslation } from "react-i18next"; import { Header, LeftNav, RightNav, RoomHeaderInfo } from "../Header"; import { HeaderStyle, useUrlParams } from "../UrlParams"; import { useCallViewKeyboardShortcuts } from "../useCallViewKeyboardShortcuts"; -import { widget } from "../widget"; +import { widget, ElementWidgetActions } from "../widget"; import styles from "./InCallView.module.css"; import { GridTile } from "../tile/GridTile"; import { SettingsModal, defaultSettingsTab } from "../settings/SettingsModal"; @@ -142,6 +142,21 @@ export const ActiveCall: FC = (props) => { vm.leave$.pipe(scope.bind()).subscribe(props.onLeft); + // Forward currently speaking user IDs to the host client + if (widget) { + const widgetApi = widget.api; + vm.activeSpeakers$.pipe(scope.bind()).subscribe((speakers) => { + const userIds = speakers + .map((m) => m.userId) + .filter((id): id is string => typeof id === "string" && id !== ""); + widgetApi.transport + .send(ElementWidgetActions.ActiveSpeakers, { userIds }) + .catch((e) => + rootLogger.error("Failed to send active speakers action", e), + ); + }); + } + return (): void => { scope.end(); }; diff --git a/src/state/CallViewModel/CallViewModel.ts b/src/state/CallViewModel/CallViewModel.ts index d34e9160f8..f299dd7d8b 100644 --- a/src/state/CallViewModel/CallViewModel.ts +++ b/src/state/CallViewModel/CallViewModel.ts @@ -312,6 +312,8 @@ export interface CallViewModel { /** use the layout instead, this is just for the sdk export. */ remoteMatrixLivekitMembers$: Behavior; localMatrixLivekitMember$: Behavior; + /** All user media (local + remote) with their live speaking status */ + userMedia$: Behavior; /** List of participants raising their hand */ handsRaised$: Behavior>; /** List of reactions. Keys are: membership.membershipId (currently predefined as: `${membershipEvent.userId}:${membershipEvent.deviceId}`)*/ @@ -353,6 +355,7 @@ export interface CallViewModel { showSpotlightIndicators$: Behavior; showSpeakingIndicators$: Behavior; showNameTags$: Behavior; + activeSpeakers$: Behavior; spotlightExpanded$: Behavior; toggleSpotlightExpanded$: Behavior<(() => void) | null>; gridMode$: Behavior; @@ -939,6 +942,24 @@ export function createCallViewModel$( }, undefined), ), ); + // All active speakers in a call + const activeSpeakers$ = scope.behavior( + userMedia$.pipe( + switchMap((mediaItems) => + mediaItems.length === 0 + ? of([]) + : combineLatest( + mediaItems.map((m) => + m.voiceActivity$.pipe(map((v) => [m, v] as const)), + ), + ), + ), + map((mediaItems) => + mediaItems.filter(([, v]) => v).map(([m]) => m), + ), + distinctUntilChanged(shallowEquals), + ), + ); const grid$ = scope.behavior( userMedia$.pipe( @@ -1783,6 +1804,7 @@ export function createCallViewModel$( setGridMode: setGridMode, layout$: layout$, localMatrixLivekitMember$, + userMedia$, remoteMatrixLivekitMembers$: scope.behavior( remoteMatrixLivekitMembers$.pipe( map((members) => members.value), @@ -1803,6 +1825,7 @@ export function createCallViewModel$( showSpotlightIndicators$: showSpotlightIndicators$, showSpeakingIndicators$: showSpeakingIndicators$, showNameTags$, + activeSpeakers$, showHeader$: showHeader$, showFooter$: showFooter$, settingsOpen$: settingsOpen$, diff --git a/src/state/media/RemoteUserMediaViewModel.ts b/src/state/media/RemoteUserMediaViewModel.ts index 7d0ed9111d..602399e3fd 100644 --- a/src/state/media/RemoteUserMediaViewModel.ts +++ b/src/state/media/RemoteUserMediaViewModel.ts @@ -66,6 +66,20 @@ export function createRemoteUserMedia( ), ), ), + audioLevel$: scope.behavior( + pretendToBeDisconnected$.pipe( + switchMap((disconnected) => + disconnected ? of(0) : baseUserMedia.audioLevel$, + ), + ), + ), + voiceActivity$: scope.behavior( + pretendToBeDisconnected$.pipe( + switchMap((disconnected) => + disconnected ? of(false) : baseUserMedia.voiceActivity$, + ), + ), + ), videoEnabled$: scope.behavior( pretendToBeDisconnected$.pipe( switchMap((disconnected) => diff --git a/src/state/media/UserMediaViewModel.ts b/src/state/media/UserMediaViewModel.ts index ea03310302..fc512ccd6d 100644 --- a/src/state/media/UserMediaViewModel.ts +++ b/src/state/media/UserMediaViewModel.ts @@ -29,6 +29,10 @@ import { type MemberMediaInputs, type BaseMemberMediaViewModel, } from "./MemberMediaViewModel"; +import { + observeSpeakingFromLevel$, + observeTrackAudioLevel$, +} from "./observeAudioLevel"; import { type RemoteUserMediaViewModel } from "./RemoteUserMediaViewModel"; import { type ObservableScope } from "../ObservableScope"; import { showConnectionStats } from "../../settings/settings"; @@ -45,6 +49,8 @@ export type UserMediaViewModel = export interface BaseUserMediaViewModel extends BaseMemberMediaViewModel { type: "user"; speaking$: Behavior; + audioLevel$: Behavior; + voiceActivity$: Behavior; audioEnabled$: Behavior; videoEnabled$: Behavior; videoFit$: Behavior<"cover" | "contain">; @@ -106,6 +112,21 @@ export function createBaseUserMedia( >(undefined); const videoSize$ = videoSizeFromParticipant$(participant$); + + // Client-side voice activity detection using the audio track itself + const audioLevel$ = scope.behavior( + participant$.pipe( + switchMap((p) => { + if (!p) return of(0); + return observeTrackAudioLevel$( + observeParticipantMedia(p).pipe( + map((m) => m.microphoneTrack?.track), + ), + ); + }), + ), + ); + return { ...createMemberMedia(scope, { ...inputs, @@ -125,6 +146,8 @@ export function createBaseUserMedia( ), ), ), + audioLevel$, + voiceActivity$: scope.behavior(observeSpeakingFromLevel$(audioLevel$)), audioEnabled$: scope.behavior( media$.pipe(map((m) => m?.microphoneTrack?.isMuted === false)), ), diff --git a/src/state/media/observeAudioLevel.test.ts b/src/state/media/observeAudioLevel.test.ts new file mode 100644 index 0000000000..f24f54df58 --- /dev/null +++ b/src/state/media/observeAudioLevel.test.ts @@ -0,0 +1,185 @@ +/* +SableCall +Copyright (C) 2026 TomOdellSheetMusic + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU Affero General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU Affero General Public License for more details. + +You should have received a copy of the GNU Affero General Public License +along with this program. If not, see . +*/ + +import { describe, expect, test, vi, beforeEach, afterEach } from "vitest"; +import { BehaviorSubject, of } from "rxjs"; +import type { LocalAudioTrack } from "livekit-client"; + +import { + observeSpeakingFromLevel$, + observeTrackAudioLevel$, + type AudioAnalyserFactory, +} from "./observeAudioLevel"; + +function mockAudioTrack(): LocalAudioTrack { + return { + kind: "audio", + mediaStreamTrack: {} as MediaStreamTrack, + isMuted: false, + } as unknown as LocalAudioTrack; +} + +describe("observeTrackAudioLevel$", () => { + let analyserFactory: ReturnType>; + let cleanup: ReturnType Promise>>; + + beforeEach(() => { + vi.useFakeTimers(); + cleanup = vi.fn<() => Promise>().mockResolvedValue(undefined); + analyserFactory = vi.fn(() => ({ + calculateVolume: () => 0, + cleanup, + })); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + test("emits 0 when there is no track", () => { + const levels: number[] = []; + observeTrackAudioLevel$(of(undefined), analyserFactory).subscribe( + (level) => levels.push(level), + ); + expect(levels).toEqual([0]); + expect(analyserFactory).not.toHaveBeenCalled(); + }); + + test("emits the calculated volume for an audio track", async () => { + analyserFactory = vi.fn(() => ({ + calculateVolume: () => 0.7, + cleanup, + })); + + const levels: number[] = []; + observeTrackAudioLevel$(of(mockAudioTrack()), analyserFactory).subscribe( + (level) => levels.push(level), + ); + // Initial emission (startWith(0)) + expect(levels).toEqual([0]); + expect(analyserFactory).toHaveBeenCalled(); + + // Advance the interval timer + await vi.advanceTimersByTimeAsync(200); + expect(levels).toEqual([0, 0.7]); + }); + + test("cleans up the analyser when unsubscribed", () => { + const sub = observeTrackAudioLevel$( + of(mockAudioTrack()), + analyserFactory, + ).subscribe(); + sub.unsubscribe(); + + expect(cleanup).toHaveBeenCalled(); + }); +}); + +describe("observeSpeakingFromLevel$", () => { + let levels: BehaviorSubject; + let speaking: boolean[]; + let sub: ReturnType; + + function subscribeToSpeaking(options?: Parameters[1]) { + speaking = []; + const s = observeSpeakingFromLevel$(levels, options).subscribe((v) => + speaking.push(v), + ); + return s; + } + + beforeEach(() => { + vi.useFakeTimers(); + levels = new BehaviorSubject(0.01); // below threshold + }); + + afterEach(() => { + sub?.unsubscribe(); + vi.useRealTimers(); + }); + + test("starts as not speaking and stays silent when level is low", () => { + sub = subscribeToSpeaking({ confirmMs: 300, dropOffMs: 1000 }); + expect(speaking).toEqual([false]); + levels.next(0.01); + expect(speaking).toEqual([false]); + }); + + test("brief blip above threshold does not trigger speaking", async () => { + sub = subscribeToSpeaking({ confirmMs: 300, dropOffMs: 1000 }); + levels.next(0.1); // blip above threshold + await vi.advanceTimersByTimeAsync(100); // blip lasts 100ms < confirmMs + levels.next(0.01); // back below threshold + await vi.advanceTimersByTimeAsync(1000); // more than confirmMs + expect(speaking).toEqual([false]); + }); + + test("sustained voice becomes speaking after confirm period", async () => { + sub = subscribeToSpeaking({ confirmMs: 300, dropOffMs: 1000 }); + levels.next(0.1); // above threshold + await vi.advanceTimersByTimeAsync(200); + expect(speaking).toEqual([false]); // not yet confirmed + await vi.advanceTimersByTimeAsync(100); // total 300ms + expect(speaking).toEqual([false, true]); // confirmed speaking + }); + + test("stops speaking after drop-off once level falls below hold threshold", async () => { + sub = subscribeToSpeaking({ confirmMs: 300, dropOffMs: 1000 }); + levels.next(0.1); + await vi.advanceTimersByTimeAsync(300); + expect(speaking).toEqual([false, true]); // confirmed speaking + levels.next(0.01); // below hold threshold + await vi.advanceTimersByTimeAsync(500); + expect(speaking).toEqual([false, true]); // still speaking during drop-off + await vi.advanceTimersByTimeAsync(500); // total 1000ms drop-off + expect(speaking).toEqual([false, true, false]); // stopped speaking + }); + + test("holds speaking through brief dips (hysteresis)", async () => { + sub = subscribeToSpeaking({ confirmMs: 300, dropOffMs: 1000 }); + levels.next(0.1); + await vi.advanceTimersByTimeAsync(300); + expect(speaking).toEqual([false, true]); // confirmed speaking + levels.next(0.01); // brief dip below hold threshold + await vi.advanceTimersByTimeAsync(100); // shorter than drop-off + levels.next(0.1); // resume speaking + await vi.advanceTimersByTimeAsync(1000); + expect(speaking).toEqual([false, true]); // never stopped speaking + }); + + test("hysteresis: requires higher level to start than to keep speaking", async () => { + sub = subscribeToSpeaking({ + threshold: 0.05, + holdThreshold: 0.02, + confirmMs: 300, + dropOffMs: 1000, + }); + // 0.03 is above hold but below threshold: should NOT start speaking + levels.next(0.03); + await vi.advanceTimersByTimeAsync(1000); + expect(speaking).toEqual([false]); + // 0.1 is above threshold: starts speaking after confirm + levels.next(0.1); + await vi.advanceTimersByTimeAsync(300); + expect(speaking).toEqual([false, true]); + // 0.03 is below threshold but above hold: keeps speaking + levels.next(0.03); + await vi.advanceTimersByTimeAsync(500); + expect(speaking).toEqual([false, true]); + }); +}); diff --git a/src/state/media/observeAudioLevel.ts b/src/state/media/observeAudioLevel.ts new file mode 100644 index 0000000000..98f9d659d8 --- /dev/null +++ b/src/state/media/observeAudioLevel.ts @@ -0,0 +1,111 @@ +/* +SableCall +Copyright (C) 2026 TomOdellSheetMusic + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU Affero General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU Affero General Public License for more details. + +You should have received a copy of the GNU Affero General Public License +along with this program. If not, see . +*/ +import { + createAudioAnalyser, + type AudioAnalyserOptions, + type LocalAudioTrack, + type RemoteAudioTrack, + type Track, +} from "livekit-client"; +import { + distinctUntilChanged, + finalize, + interval, + map, + of, + scan, + startWith, + switchMap, + timer, + type Observable, +} from "rxjs"; + +// Constants for audio level detection and debounce +export const AUDIO_LEVEL_SAMPLE_INTERVAL_MS = 100; +export const VOICE_ACTIVITY_THRESHOLD = 0.05; +export const VOICE_ACTIVITY_HOLD_THRESHOLD = 0.02; +export const VOICE_ACTIVITY_CONFIRM_MS = 50; +export const VOICE_ACTIVITY_DROP_OFF_MS = 50; + +export type AudioAnalyserFactory = ( + track: LocalAudioTrack | RemoteAudioTrack, + options?: AudioAnalyserOptions, +) => { calculateVolume: () => number; cleanup: () => Promise }; + +function isAudioTrack( + track: Track, +): track is LocalAudioTrack | RemoteAudioTrack { + return track.kind === "audio" && typeof track.mediaStreamTrack === "object"; +} + + +// Raw audio level (0-1) of a participant's microphone track, sampled continuously. +export function observeTrackAudioLevel$( + track$: Observable, + analyserFactory: AudioAnalyserFactory = createAudioAnalyser, +): Observable { + return track$.pipe( + switchMap((track) => { + if (!track || !isAudioTrack(track)) return of(0); + const { calculateVolume, cleanup } = analyserFactory(track, { + cloneTrack: true, + smoothingTimeConstant: 0.1, + }); + return interval(AUDIO_LEVEL_SAMPLE_INTERVAL_MS).pipe( + map(() => calculateVolume()), + startWith(0), + distinctUntilChanged(), + finalize(() => void cleanup()), + ); + }), + ); +} + +export interface SpeakingOptions { + threshold?: number; + holdThreshold?: number; + confirmMs?: number; + dropOffMs?: number; +} + +// Debounced speaking detection +export function observeSpeakingFromLevel$( + level$: Observable, + { + threshold = VOICE_ACTIVITY_THRESHOLD, + holdThreshold = VOICE_ACTIVITY_HOLD_THRESHOLD, + confirmMs = VOICE_ACTIVITY_CONFIRM_MS, + dropOffMs = VOICE_ACTIVITY_DROP_OFF_MS, + }: SpeakingOptions = {}, +): Observable { + return level$.pipe( + scan( + (speaking, level) => + speaking ? level > holdThreshold : level > threshold, + false, + ), + distinctUntilChanged(), + switchMap((speaking, index) => + index === 0 + ? of(speaking) + : timer(speaking ? confirmMs : dropOffMs).pipe(map(() => speaking)), + ), + distinctUntilChanged(), + ); +} + diff --git a/src/widget.ts b/src/widget.ts index 6bb326e7d7..259f64884b 100644 --- a/src/widget.ts +++ b/src/widget.ts @@ -28,6 +28,7 @@ export enum ElementWidgetActions { JoinCall = "io.element.join", HangupCall = "im.vector.hangup", Close = "io.element.close", + ActiveSpeakers = "io.element.active_speakers", // This can be sent as from or to widget // fromWidget: updates the client about the current device mute state // toWidget: the client requests a specific device mute configuration