dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-27 22:34:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

Moving smoothing into the audio level module. (#636)

Browse Source
This commit is contained in:
Raja Subramanian
2022-04-20 23:59:51 +05:30
committed by GitHub
parent 6a53891f9f
commit 43d0573693
11 changed files with 93 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/mediatrackreceiver.go
+2 -3
View File
@@ -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()
pkg/rtc/participant.go
+10 -9
View File
@@ -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)
}
pkg/rtc/room.go
+2 -51
View File
@@ -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 {
pkg/rtc/room_test.go
+3 -3
View File
@@ -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)
pkg/rtc/types/interfaces.go
+2 -2
View File
@@ -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)
pkg/rtc/types/typesfakes/fake_local_media_track.go
+10 -10
View File
@@ -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}
}
pkg/rtc/types/typesfakes/fake_local_participant.go
+10 -10
View File
@@ -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}
}
pkg/sfu/audio/audiolevel.go
+45 -28
View File
@@ -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)
}
pkg/sfu/audio/audiolevel_test.go
+2 -2
View File
@@ -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)))
})
}
pkg/sfu/buffer/buffer.go
+2 -2
View File
@@ -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()
pkg/sfu/receiver.go
+5 -4
View File
@@ -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 {