From 43d057369328ff51bf9b29389258f0217df7be78 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 20 Apr 2022 23:59:51 +0530 Subject: [PATCH] Moving smoothing into the audio level module. (#636) --- pkg/rtc/mediatrackreceiver.go | 5 +- pkg/rtc/participant.go | 19 ++--- pkg/rtc/room.go | 53 +------------- pkg/rtc/room_test.go | 6 +- pkg/rtc/types/interfaces.go | 4 +- .../typesfakes/fake_local_media_track.go | 20 ++--- .../typesfakes/fake_local_participant.go | 20 ++--- pkg/sfu/audio/audiolevel.go | 73 ++++++++++++------- pkg/sfu/audio/audiolevel_test.go | 4 +- pkg/sfu/buffer/buffer.go | 4 +- pkg/sfu/receiver.go | 9 ++- 11 files changed, 93 insertions(+), 124 deletions(-) diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index 44b904752..a80cedc75 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -15,7 +15,6 @@ import ( "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu" - "github.com/livekit/livekit-server/pkg/sfu/audio" "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/livekit-server/pkg/telemetry" ) @@ -304,10 +303,10 @@ func (t *MediaTrackReceiver) GetQualityForDimension(width, height uint32) liveki return quality } -func (t *MediaTrackReceiver) GetAudioLevel() (uint8, bool) { +func (t *MediaTrackReceiver) GetAudioLevel() (float64, bool) { receiver := t.Receiver() if receiver == nil { - return audio.SilentAudioLevel, false + return 0, false } return receiver.GetAudioLevel() diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index c4322b819..f8384792e 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -23,7 +23,6 @@ import ( "github.com/livekit/livekit-server/pkg/routing" "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu" - "github.com/livekit/livekit-server/pkg/sfu/audio" "github.com/livekit/livekit-server/pkg/sfu/connectionquality" "github.com/livekit/livekit-server/pkg/sfu/twcc" "github.com/livekit/livekit-server/pkg/telemetry" @@ -689,15 +688,15 @@ func (p *ParticipantImpl) ICERestart() error { // signal connection methods // -func (p *ParticipantImpl) GetAudioLevel() (level uint8, active bool) { - level = audio.SilentAudioLevel +func (p *ParticipantImpl) GetAudioLevel() (level float64, active bool) { + level = 0 for _, pt := range p.GetPublishedTracks() { mediaTrack := pt.(types.LocalMediaTrack) if mediaTrack.Source() == livekit.TrackSource_MICROPHONE { tl, ta := mediaTrack.GetAudioLevel() if ta { active = true - if tl < level { + if tl > level { level = tl } } @@ -975,11 +974,13 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w publishedTrack, isNewTrack := p.mediaTrackReceived(track, rtpReceiver) - p.params.Logger.Infow("mediaTrack published", - "kind", track.Kind().String(), - "trackID", publishedTrack.ID(), - "rid", track.RID(), - "SSRC", track.SSRC()) + if publishedTrack != nil { + p.params.Logger.Infow("mediaTrack published", + "kind", track.Kind().String(), + "trackID", publishedTrack.ID(), + "rid", track.RID(), + "SSRC", track.SSRC()) + } if !isNewTrack && publishedTrack != nil && p.IsReady() && p.onTrackUpdated != nil { p.onTrackUpdated(p, publishedTrack) } diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 661610b63..ec091bad5 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -2,7 +2,6 @@ package rtc import ( "context" - "math" "sort" "sync" "time" @@ -16,7 +15,6 @@ import ( "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/routing" "github.com/livekit/livekit-server/pkg/rtc/types" - "github.com/livekit/livekit-server/pkg/sfu/audio" "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/livekit-server/pkg/telemetry" "github.com/livekit/livekit-server/pkg/telemetry/prometheus" @@ -131,7 +129,7 @@ func (r *Room) GetActiveSpeakers() []*livekit.SpeakerInfo { } speakers = append(speakers, &livekit.SpeakerInfo{ Sid: string(p.ID()), - Level: audio.ConvertAudioLevel(level), + Level: float32(level), Active: active, }) } @@ -762,16 +760,6 @@ func (r *Room) sendSpeakerChanges(speakers []*livekit.SpeakerInfo) { } func (r *Room) audioUpdateWorker() { - var smoothValues map[livekit.ParticipantID]float32 - var smoothFactor float32 - var activeThreshold float32 - if ss := r.audioConfig.SmoothIntervals; ss > 1 { - smoothValues = make(map[livekit.ParticipantID]float32) - // exponential moving average (EMA), same center of mass with simple moving average (SMA) - smoothFactor = 2 / float32(ss+1) - activeThreshold = audio.ConvertAudioLevel(r.audioConfig.ActiveLevel) - } - lastActiveMap := make(map[livekit.ParticipantID]*livekit.SpeakerInfo) for { if r.IsClosed() { @@ -779,44 +767,6 @@ func (r *Room) audioUpdateWorker() { } activeSpeakers := r.GetActiveSpeakers() - if smoothValues != nil { - for _, speaker := range activeSpeakers { - sid := livekit.ParticipantID(speaker.Sid) - level := smoothValues[sid] - delete(smoothValues, sid) - // exponential moving average (EMA) - level += (speaker.Level - level) * smoothFactor - speaker.Level = level - } - - // ensure that previous active speakers are also included - for sid, level := range smoothValues { - delete(smoothValues, sid) - level += -level * smoothFactor - if level > activeThreshold { - activeSpeakers = append(activeSpeakers, &livekit.SpeakerInfo{ - Sid: string(sid), - Level: level, - Active: true, - }) - } - } - - // smoothValues map is drained, now repopulate it back - for _, speaker := range activeSpeakers { - smoothValues[livekit.ParticipantID(speaker.Sid)] = speaker.Level - } - - sort.Slice(activeSpeakers, func(i, j int) bool { - return activeSpeakers[i].Level > activeSpeakers[j].Level - }) - } - - const invAudioLevelQuantization = 1.0 / AudioLevelQuantization - for _, speaker := range activeSpeakers { - speaker.Level = float32(math.Ceil(float64(speaker.Level*AudioLevelQuantization)) * invAudioLevelQuantization) - } - changedSpeakers := make([]*livekit.SpeakerInfo, 0, len(activeSpeakers)) nextActiveMap := make(map[livekit.ParticipantID]*livekit.SpeakerInfo, len(activeSpeakers)) for _, speaker := range activeSpeakers { @@ -826,6 +776,7 @@ func (r *Room) audioUpdateWorker() { } nextActiveMap[livekit.ParticipantID(speaker.Sid)] = speaker } + // changedSpeakers need to include previous speakers that are no longer speaking for sid, speaker := range lastActiveMap { if nextActiveMap[sid] == nil { diff --git a/pkg/rtc/room_test.go b/pkg/rtc/room_test.go index 223015ff1..8a3138330 100644 --- a/pkg/rtc/room_test.go +++ b/pkg/rtc/room_test.go @@ -306,8 +306,8 @@ func TestActiveSpeakers(t *testing.T) { participants := rm.GetParticipants() p := participants[0].(*typesfakes.FakeLocalParticipant) p2 := participants[1].(*typesfakes.FakeLocalParticipant) - p.GetAudioLevelReturns(10, true) - p2.GetAudioLevelReturns(20, true) + p.GetAudioLevelReturns(20, true) + p2.GetAudioLevelReturns(10, true) speakers := rm.GetActiveSpeakers() require.Len(t, speakers, 2) @@ -361,7 +361,7 @@ func TestActiveSpeakers(t *testing.T) { p := participants[0].(*typesfakes.FakeLocalParticipant) op := participants[1].(*typesfakes.FakeLocalParticipant) p.GetAudioLevelReturns(30, true) - convertedLevel := audio.ConvertAudioLevel(30) + convertedLevel := float32(audio.ConvertAudioLevel(30)) testutils.WithTimeout(t, func() string { updates := getActiveSpeakerUpdates(op) diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 6ea99b931..229b51b70 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -127,7 +127,7 @@ type LocalParticipant interface { // returns list of participant identities that the current participant is subscribed to GetSubscribedParticipants() []livekit.ParticipantID - GetAudioLevel() (level uint8, active bool) + GetAudioLevel() (smoothedLevel float64, active bool) GetConnectionQuality() *livekit.ConnectionQualityInfo // server sent messages @@ -220,7 +220,7 @@ type LocalMediaTrack interface { SignalCid() string SdpCid() string - GetAudioLevel() (level uint8, active bool) + GetAudioLevel() (level float64, active bool) GetConnectionScore() float32 SetRTT(rtt uint32) diff --git a/pkg/rtc/types/typesfakes/fake_local_media_track.go b/pkg/rtc/types/typesfakes/fake_local_media_track.go index cd027a4d1..f52efe5ed 100644 --- a/pkg/rtc/types/typesfakes/fake_local_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_local_media_track.go @@ -36,16 +36,16 @@ type FakeLocalMediaTrack struct { getAllSubscribersReturnsOnCall map[int]struct { result1 []livekit.ParticipantID } - GetAudioLevelStub func() (uint8, bool) + GetAudioLevelStub func() (float64, bool) getAudioLevelMutex sync.RWMutex getAudioLevelArgsForCall []struct { } getAudioLevelReturns struct { - result1 uint8 + result1 float64 result2 bool } getAudioLevelReturnsOnCall map[int]struct { - result1 uint8 + result1 float64 result2 bool } GetConnectionScoreStub func() float32 @@ -403,7 +403,7 @@ func (fake *FakeLocalMediaTrack) GetAllSubscribersReturnsOnCall(i int, result1 [ }{result1} } -func (fake *FakeLocalMediaTrack) GetAudioLevel() (uint8, bool) { +func (fake *FakeLocalMediaTrack) GetAudioLevel() (float64, bool) { fake.getAudioLevelMutex.Lock() ret, specificReturn := fake.getAudioLevelReturnsOnCall[len(fake.getAudioLevelArgsForCall)] fake.getAudioLevelArgsForCall = append(fake.getAudioLevelArgsForCall, struct { @@ -427,34 +427,34 @@ func (fake *FakeLocalMediaTrack) GetAudioLevelCallCount() int { return len(fake.getAudioLevelArgsForCall) } -func (fake *FakeLocalMediaTrack) GetAudioLevelCalls(stub func() (uint8, bool)) { +func (fake *FakeLocalMediaTrack) GetAudioLevelCalls(stub func() (float64, bool)) { fake.getAudioLevelMutex.Lock() defer fake.getAudioLevelMutex.Unlock() fake.GetAudioLevelStub = stub } -func (fake *FakeLocalMediaTrack) GetAudioLevelReturns(result1 uint8, result2 bool) { +func (fake *FakeLocalMediaTrack) GetAudioLevelReturns(result1 float64, result2 bool) { fake.getAudioLevelMutex.Lock() defer fake.getAudioLevelMutex.Unlock() fake.GetAudioLevelStub = nil fake.getAudioLevelReturns = struct { - result1 uint8 + result1 float64 result2 bool }{result1, result2} } -func (fake *FakeLocalMediaTrack) GetAudioLevelReturnsOnCall(i int, result1 uint8, result2 bool) { +func (fake *FakeLocalMediaTrack) GetAudioLevelReturnsOnCall(i int, result1 float64, result2 bool) { fake.getAudioLevelMutex.Lock() defer fake.getAudioLevelMutex.Unlock() fake.GetAudioLevelStub = nil if fake.getAudioLevelReturnsOnCall == nil { fake.getAudioLevelReturnsOnCall = make(map[int]struct { - result1 uint8 + result1 float64 result2 bool }) } fake.getAudioLevelReturnsOnCall[i] = struct { - result1 uint8 + result1 float64 result2 bool }{result1, result2} } diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 1a2a475fe..0bf3a1cbc 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -131,16 +131,16 @@ type FakeLocalParticipant struct { getAdaptiveStreamReturnsOnCall map[int]struct { result1 bool } - GetAudioLevelStub func() (uint8, bool) + GetAudioLevelStub func() (float64, bool) getAudioLevelMutex sync.RWMutex getAudioLevelArgsForCall []struct { } getAudioLevelReturns struct { - result1 uint8 + result1 float64 result2 bool } getAudioLevelReturnsOnCall map[int]struct { - result1 uint8 + result1 float64 result2 bool } GetConnectionQualityStub func() *livekit.ConnectionQualityInfo @@ -1253,7 +1253,7 @@ func (fake *FakeLocalParticipant) GetAdaptiveStreamReturnsOnCall(i int, result1 }{result1} } -func (fake *FakeLocalParticipant) GetAudioLevel() (uint8, bool) { +func (fake *FakeLocalParticipant) GetAudioLevel() (float64, bool) { fake.getAudioLevelMutex.Lock() ret, specificReturn := fake.getAudioLevelReturnsOnCall[len(fake.getAudioLevelArgsForCall)] fake.getAudioLevelArgsForCall = append(fake.getAudioLevelArgsForCall, struct { @@ -1277,34 +1277,34 @@ func (fake *FakeLocalParticipant) GetAudioLevelCallCount() int { return len(fake.getAudioLevelArgsForCall) } -func (fake *FakeLocalParticipant) GetAudioLevelCalls(stub func() (uint8, bool)) { +func (fake *FakeLocalParticipant) GetAudioLevelCalls(stub func() (float64, bool)) { fake.getAudioLevelMutex.Lock() defer fake.getAudioLevelMutex.Unlock() fake.GetAudioLevelStub = stub } -func (fake *FakeLocalParticipant) GetAudioLevelReturns(result1 uint8, result2 bool) { +func (fake *FakeLocalParticipant) GetAudioLevelReturns(result1 float64, result2 bool) { fake.getAudioLevelMutex.Lock() defer fake.getAudioLevelMutex.Unlock() fake.GetAudioLevelStub = nil fake.getAudioLevelReturns = struct { - result1 uint8 + result1 float64 result2 bool }{result1, result2} } -func (fake *FakeLocalParticipant) GetAudioLevelReturnsOnCall(i int, result1 uint8, result2 bool) { +func (fake *FakeLocalParticipant) GetAudioLevelReturnsOnCall(i int, result1 float64, result2 bool) { fake.getAudioLevelMutex.Lock() defer fake.getAudioLevelMutex.Unlock() fake.GetAudioLevelStub = nil if fake.getAudioLevelReturnsOnCall == nil { fake.getAudioLevelReturnsOnCall = make(map[int]struct { - result1 uint8 + result1 float64 result2 bool }) } fake.getAudioLevelReturnsOnCall[i] = struct { - result1 uint8 + result1 float64 result2 bool }{result1, result2} } diff --git a/pkg/sfu/audio/audiolevel.go b/pkg/sfu/audio/audiolevel.go index 986113782..15b549c9c 100644 --- a/pkg/sfu/audio/audiolevel.go +++ b/pkg/sfu/audio/audiolevel.go @@ -7,38 +7,46 @@ import ( ) const ( - // duration of audio frames for observe window - SilentAudioLevel = 127 + silentAudioLevel = 127 + negInv20 = -1.0 / 20 ) type AudioLevelParams struct { ActiveLevel uint8 MinPercentile uint8 ObserveDuration uint32 + SmoothIntervals uint32 } // keeps track of audio level for a participant type AudioLevel struct { params AudioLevelParams - - currentLevel *atomic.Uint32 - // min duration to be considered active + // min duration within an observe duration window to be considered active minActiveDuration uint32 + smoothFactor float64 + activeThreshold float64 - // for Observe goroutine use - // keeps track of current activity - observeLevel uint8 - activeDuration uint32 // ms - observedDuration uint32 // ms + smoothedLevel atomic.Float64 + + loudestObservedLevel uint8 + activeDuration uint32 // ms + observedDuration uint32 // ms } func NewAudioLevel(params AudioLevelParams) *AudioLevel { l := &AudioLevel{ - params: params, - minActiveDuration: uint32(params.MinPercentile) * params.ObserveDuration / 100, - currentLevel: atomic.NewUint32(SilentAudioLevel), - observeLevel: SilentAudioLevel, + params: params, + minActiveDuration: uint32(params.MinPercentile) * params.ObserveDuration / 100, + smoothFactor: 1, + activeThreshold: ConvertAudioLevel(float64(params.ActiveLevel)), + loudestObservedLevel: silentAudioLevel, } + + if l.params.SmoothIntervals > 0 { + // exponential moving average (EMA), same center of mass with simple moving average (SMA) + l.smoothFactor = float64(2) / (float64(l.params.SmoothIntervals + 1)) + } + return l } @@ -48,34 +56,43 @@ func (l *AudioLevel) Observe(level uint8, durationMs uint32) { if level <= l.params.ActiveLevel { l.activeDuration += durationMs - if l.observeLevel > level { - l.observeLevel = level + if l.loudestObservedLevel > level { + l.loudestObservedLevel = level } } if l.observedDuration >= l.params.ObserveDuration { // compute and reset if l.activeDuration >= l.minActiveDuration { - level := uint32(l.observeLevel) - uint32(20*math.Log10(float64(l.activeDuration)/float64(l.params.ObserveDuration))) - l.currentLevel.Store(level) + // adjust loudest observed level by how much of the window was active. + // Weight will be 0 if active the entire duration + // > 0 if active for longer than observe duration + // < 0 if active for less than observe duration + activityWeight := 20 * math.Log10(float64(l.activeDuration)/float64(l.params.ObserveDuration)) + adjustedLevel := float64(l.loudestObservedLevel) - activityWeight + linearLevel := ConvertAudioLevel(adjustedLevel) + + // exponential smoothing to dampen transients + smoothedLevel := l.smoothedLevel.Load() + smoothedLevel += (linearLevel - smoothedLevel) * l.smoothFactor + l.smoothedLevel.Store(smoothedLevel) } else { - l.currentLevel.Store(SilentAudioLevel) + l.smoothedLevel.Store(0) } - l.observeLevel = SilentAudioLevel + l.loudestObservedLevel = silentAudioLevel l.activeDuration = 0 l.observedDuration = 0 } } -// returns current audio level, 0 (loudest) to 127 (silent) -func (l *AudioLevel) GetLevel() (uint8, bool) { - level := uint8(l.currentLevel.Load()) - active := level != SilentAudioLevel - return level, active +// returns current soothed audio level +func (l *AudioLevel) GetLevel() (float64, bool) { + smoothedLevel := l.smoothedLevel.Load() + active := smoothedLevel >= l.activeThreshold + return smoothedLevel, active } // convert decibel back to linear -func ConvertAudioLevel(level uint8) float32 { - const negInv20 = -1.0 / 20 - return float32(math.Pow(10, float64(level)*negInv20)) +func ConvertAudioLevel(level float64) float64 { + return math.Pow(10, level*negInv20) } diff --git a/pkg/sfu/audio/audiolevel_test.go b/pkg/sfu/audio/audiolevel_test.go index 8c6da378e..aadd4ca29 100644 --- a/pkg/sfu/audio/audiolevel_test.go +++ b/pkg/sfu/audio/audiolevel_test.go @@ -53,8 +53,8 @@ func TestAudioLevel(t *testing.T) { level, noisy := a.GetLevel() require.True(t, noisy) - require.Less(t, level, uint8(defaultActiveLevel)) - require.Greater(t, level, uint8(25)) + require.Greater(t, level, ConvertAudioLevel(float64(defaultActiveLevel))) + require.Less(t, level, ConvertAudioLevel(float64(25))) }) } diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index a8b0f088f..4f585cab3 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -628,12 +628,12 @@ func (b *Buffer) GetDeltaStats() *StreamStatsWithLayers { } } -func (b *Buffer) GetAudioLevel() (uint8, bool) { +func (b *Buffer) GetAudioLevel() (float64, bool) { b.RLock() defer b.RUnlock() if b.audioLevel == nil { - return audio.SilentAudioLevel, false + return 0, false } return b.audioLevel.GetLevel() diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 798b74cab..b0754d386 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -40,7 +40,7 @@ type TrackReceiver interface { ReadRTP(buf []byte, layer uint8, sn uint16) (int, error) GetBitrateTemporalCumulative() Bitrates - GetAudioLevel() (uint8, bool) + GetAudioLevel() (float64, bool) SendPLI(layer int32) @@ -275,6 +275,7 @@ func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buff ActiveLevel: w.audioConfig.ActiveLevel, MinPercentile: w.audioConfig.MinPercentile, ObserveDuration: w.audioConfig.UpdateInterval, + SmoothIntervals: w.audioConfig.SmoothIntervals, }) buff.OnFeedback(w.sendRTCP) @@ -450,9 +451,9 @@ func (w *WebRTCReceiver) GetTrackStats() *livekit.RTPStats { return buffer.AggregateRTPStats(stats) } -func (w *WebRTCReceiver) GetAudioLevel() (uint8, bool) { +func (w *WebRTCReceiver) GetAudioLevel() (float64, bool) { if w.Kind() == webrtc.RTPCodecTypeVideo { - return audio.SilentAudioLevel, false + return 0, false } w.bufferMu.RLock() @@ -466,7 +467,7 @@ func (w *WebRTCReceiver) GetAudioLevel() (uint8, bool) { return buff.GetAudioLevel() } - return audio.SilentAudioLevel, false + return 0, false } func (w *WebRTCReceiver) getQualityParams() *buffer.ConnectionQualityParams {