/* Copyright 2026 Lotus Guild SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial Please see LICENSE in the repository root for full details. */ import { type LocalTrackPublication, type Room as LivekitRoom, RoomEvent, Track, } from "livekit-client"; import { Observable, type Subscription, share, switchMap } from "rxjs"; import { distinctUntilChanged, map } from "rxjs/operators"; import { type CallViewModel } from "../state/CallViewModel/CallViewModel"; import { widget } from "../widget"; import { LotusWidgetActions, lotusFlag, lotusSendToHost } from "./lotusWidget"; /** * [lotus #146] One local mic sampler shared by the host's mic level meter * (`io.lotus.mic_level`) and "talking while muted" (#37, lotusMutedSpeech). * It taps a CLONE of the published mic track (the post-processor track when * the in-source denoiser is active, so what's measured is what's sent) and * reads RMS at ~10 Hz while a mic track is published, muted or not. * Local only: nothing here reaches other participants. */ const SAMPLE_MS = 100; export interface LocalMicSample { rms: number; /** The mic is published but muted (the clone still hears it). */ muted: boolean; } const micPublication = ( room: LivekitRoom, ): { track: MediaStreamTrack; muted: boolean } | null => { const pub: LocalTrackPublication | undefined = room.localParticipant.getTrackPublication(Track.Source.Microphone); const track = pub?.track?.mediaStreamTrack; return track && track.readyState === "live" ? { track, muted: pub?.isMuted ?? false } : null; }; /** Pure: the same LiveKit rooms, in the same order (by identity). */ export const sameRooms = (a: readonly T[], b: readonly T[]): boolean => a.length === b.length && a.every((r, i) => r === b[i]); const samplers = new WeakMap< CallViewModel, Observable >(); /** * RMS samples of the local mic, or `null` while no mic track is published. * Shared per call view model, so the meter and the muted-speech detector use * one AudioContext between them. */ export function observeLocalMicSample$( vm: CallViewModel, ): Observable { const cached = samplers.get(vm); if (cached) return cached; const sampler$ = vm.allConnections$.pipe( map((data) => data.getConnections().map((c) => c.livekitRoom)), // [EC #45] allConnections$ re-emits far more often than the set of // LiveKit rooms changes; restarting the tap each time closed the // AudioContext ~1×/s and emitted null, which reset the muted-speech // gate so the "You're muted" nudge could never fire. distinctUntilChanged(sameRooms), switchMap( (rooms) => new Observable((subscriber) => { let ctx: AudioContext | null = null; let clone: MediaStreamTrack | null = null; let timer: ReturnType | undefined; let tapped: MediaStreamTrack | null = null; let muted = false; const stopTap = (): void => { if (timer !== undefined) clearInterval(timer); timer = undefined; clone?.stop(); clone = null; void ctx?.close().catch(() => undefined); ctx = null; tapped = null; subscriber.next(null); }; const startTap = (source: MediaStreamTrack): void => { try { clone = source.clone(); // Muting disables the published track; the clone must still hear. clone.enabled = true; ctx = new AudioContext(); const analyser = ctx.createAnalyser(); analyser.fftSize = 1024; ctx .createMediaStreamSource(new MediaStream([clone])) .connect(analyser); const buf = new Float32Array(analyser.fftSize); tapped = source; timer = setInterval(() => { analyser.getFloatTimeDomainData(buf); let sum = 0; for (let i = 0; i < buf.length; i += 1) sum += buf[i] * buf[i]; subscriber.next({ rms: Math.sqrt(sum / buf.length), muted }); }, SAMPLE_MS); } catch { stopTap(); } }; const reconcile = (): void => { const pub = rooms.map(micPublication).find((p) => p !== null) ?? null; muted = pub?.muted ?? false; const track = pub?.track ?? null; if (track === tapped) return; if (tapped) stopTap(); if (track) startTap(track); }; const events = [ RoomEvent.TrackMuted, RoomEvent.TrackUnmuted, RoomEvent.LocalTrackPublished, RoomEvent.LocalTrackUnpublished, RoomEvent.Reconnected, RoomEvent.Disconnected, ] as const; rooms.forEach((room) => events.forEach((ev) => room.on(ev, reconcile)), ); subscriber.next(null); reconcile(); return (): void => { rooms.forEach((room) => events.forEach((ev) => room.off(ev, reconcile)), ); if (tapped) stopTap(); }; }), ), share(), ); samplers.set(vm, sampler$); return sampler$; } /** RMS at which each bar lights: ≈ −46, −36 and −26 dBFS. */ export const BAR_THRESHOLDS = [0.005, 0.015, 0.05] as const; /** * Quantise RMS to 0–3 bars with a little hysteresis: rises at once, falls one * bar per sample, so the meter doesn't flicker between words. Unit-tested. */ export class MicLevelQuantizer { private bars = 0; public reset(): void { this.bars = 0; } public push(rms: number): number { const target = BAR_THRESHOLDS.filter((t) => rms >= t).length; this.bars = target >= this.bars ? target : this.bars - 1; return this.bars; } } /** * Send the host `io.lotus.mic_level { bars }` (0–3) whenever the quantised * level changes; 0 while muted or with no mic. At most one message per sample * (10 Hz) and none during steady silence. Opt-in with the rest of the host * state stream (`lotusCallState=1`). Returns a teardown function. */ export function startLotusMicLevel(vm: CallViewModel): () => void { if (!lotusFlag("lotusCallState") || !widget) return (): void => undefined; const quantizer = new MicLevelQuantizer(); const sub: Subscription = observeLocalMicSample$(vm) .pipe( map((sample) => { if (!sample || sample.muted) { quantizer.reset(); return 0; } return quantizer.push(sample.rms); }), distinctUntilChanged(), ) .subscribe((bars) => { lotusSendToHost(LotusWidgetActions.MicLevel, { bars }); }); return (): void => sub.unsubscribe(); }