Mv 309 implement vad (from master branch) (#148)

* VAD implementation based on a paper

* Cleaning old vad algorithm code

* `is_speaking_esimator` and its tests

* Writing VAD tests for new implementation

* Module docs

* Adding AudioLevelsQueue and its tests

---------

Co-authored-by: Bartosz Błaszków <[email protected]>

Author: pkrucz00

.vscode/settings.json

@@ -0,0 +1,6 @@
+{
+  "cSpell.words": [
+    "dbov",
+    "immediates"
+  ]
+}

config/config.exs

@@ -1,3 +1,25 @@
 import Config
 
-config :membrane_rtp_plugin, :fir_throttle_duration_ms, 500
+config :membrane_rtp_plugin,
+  fir_throttle_duration_ms: 500,
+  vad_estimation_parameters: %{
+    immediate: %{
+      subunits: 1,
+      score_threshold: 0,
+      lambda: 1
+    },
+    medium: %{
+      subunits: 10,
+      score_threshold: 20,
+      subunit_threshold: 1,
+      lambda: 24
+    },
+    long: %{
+      subunits: 7,
+      score_threshold: 20,
+      subunit_threshold: 3,
+      lambda: 47
+    }
+  }
+
+import_config "#{config_env()}.exs"

config/dev.exs

@@ -0,0 +1,3 @@
+import Config
+
+# place for dev compile time env variables

config/prod.exs

@@ -0,0 +1,3 @@
+import Config
+
+# place for production compile time env variables

config/test.exs

@@ -0,0 +1,22 @@
+import Config
+
+config :membrane_rtp_plugin,
+  vad_estimation_parameters: %{
+    immediate: %{
+      subunits: 2,
+      score_threshold: 0.1,
+      lambda: 1
+    },
+    medium: %{
+      subunits: 2,
+      score_threshold: 0.1,
+      subunit_threshold: 2,
+      lambda: 1
+    },
+    long: %{
+      subunits: 2,
+      score_threshold: 0.1,
+      subunit_threshold: 1,
+      lambda: 1
+    }
+  }

lib/membrane/rtp/vad.ex

@@ -1,21 +1,16 @@
 defmodule Membrane.RTP.VAD do
   @moduledoc """
-  Simple vad based on audio level sent in RTP header.
+  Vad based on audio level sent in RTP header.
 
-  To make this module work appropriate RTP header extension has to be set in SDP offer/answer.
+  To make this module work appropriate RTP header extension has to be set in the SDP offer/answer.
 
-  If avg of audio level in packets in `time_window` exceeds `vad_threshold` it emits `Membrane.RTP.VadEvent`
-  on its output pad.
+  Sends `Membrane.RTP.VadEvent` when a score from `Membrane.RTP.Vad.IsSpeakingEstimator` changes.
 
-  When avg falls below `vad_threshold` and doesn't exceed it in the next `vad_silence_timer`
-  it also emits the event.
+  A more detailed explanation of how the VAD algorithm can be found in the `Membrane.RTP.Vad.IsSpeakingEstimator` module.
 
   Buffers that are processed by this element may or may not have been processed by
   a depayloader and passed through a jitter buffer. If they have not, then the only timestamp
-  available for time comparison is the RTP timestamp. The delta between RTP timestamps is
-  dependent on the clock rate used by the encoding. For `OPUS` the clock rate is `48kHz` and
-  packets are sent every `20ms`, so the RTP timestamp delta between sequential packets should
-  be `48000 / 1000 * 20`, or `960`.
+  available for time comparison is the RTP timestamp.
 
   When calculating the epoch of the timestamp, we need to account for 32bit integer wrapping.
   * `:current` - the difference between timestamps is low: the timestamp has not wrapped around.
@@ -26,6 +21,7 @@ defmodule Membrane.RTP.VAD do
   use Membrane.Filter
 
   alias Membrane.RTP.{Header, Utils, VadEvent}
+  alias Membrane.RTP.Vad.{AudioLevelQueue, IsSpeakingEstimator}
 
   def_input_pad :input, availability: :always, accepted_format: _any, demand_mode: :auto
 
@@ -35,59 +31,24 @@ defmodule Membrane.RTP.VAD do
                 spec: 1..14,
                 description: "ID of VAD header extension."
               ],
-              clock_rate: [
-                spec: Membrane.RTP.clock_rate_t(),
-                default: 48_000,
-                description: "Clock rate (in `Hz`) for the encoding."
-              ],
-              time_window: [
-                spec: pos_integer(),
-                default: 2_000,
-                description: "Time window (in `ms`) in which avg audio level is measured."
-              ],
-              min_packet_num: [
-                spec: pos_integer(),
-                default: 50,
-                description: """
-                Minimal number of packets to count avg audio level from.
-                Speech won't be detected until there are enough packets.
-                """
-              ],
               vad_threshold: [
                 spec: -127..0,
-                default: -50,
+                default: -32,
                 description: """
                 Audio level in dBov representing vad threshold.
                 Values above are considered to represent voice activity.
                 Value -127 represents digital silence.
                 """
-              ],
-              vad_silence_time: [
-                spec: pos_integer(),
-                default: 300,
-                description: """
-                Time to wait before emitting `Membrane.RTP.VadEvent` after audio track is
-                no longer considered to represent speech.
-                If at this time audio track is considered to represent speech again the event will not be sent.
-                """
               ]
 
   @impl true
   def handle_init(_ctx, opts) do
     state = %{
       vad_id: opts.vad_id,
-      audio_levels: Qex.new(),
-      clock_rate: opts.clock_rate,
+      audio_levels: AudioLevelQueue.new(),
       vad: :silence,
-      vad_silence_timestamp: 0,
       current_timestamp: nil,
-      rtp_timestamp_increment: opts.time_window * opts.clock_rate / 1000,
-      min_packet_num: opts.min_packet_num,
-      time_window: opts.time_window,
-      vad_threshold: opts.vad_threshold,
-      vad_silence_time: opts.vad_silence_time,
-      audio_levels_sum: 0,
-      audio_levels_count: 0
+      vad_threshold: opts.vad_threshold + 127
     }
 
     {[], state}
@@ -130,87 +91,34 @@ defmodule Membrane.RTP.VAD do
     end
   end
 
-  defp handle_vad(buffer, rtp_timestamp, level, state) do
-    state = %{state | current_timestamp: rtp_timestamp}
-    state = filter_old_audio_levels(state)
-    state = add_new_audio_level(state, level)
-    audio_levels_vad = get_audio_levels_vad(state)
-    actions = [buffer: {:output, buffer}] ++ maybe_send_event(audio_levels_vad, state)
-    state = update_vad_state(audio_levels_vad, state)
-    {actions, state}
-  end
+  defp handle_vad(buffer, rtp_timestamp, level_in_dbov, state) do
+    level_in_db = 127 - level_in_dbov
+    updated_audio_levels = AudioLevelQueue.add(state.audio_levels, level_in_db)
 
-  defp filter_old_audio_levels(state) do
-    Enum.reduce_while(state.audio_levels, state, fn {level, timestamp}, state ->
-      if Ratio.sub(state.current_timestamp, timestamp)
-         |> Ratio.gt?(state.rtp_timestamp_increment) do
-        {_level, audio_levels} = Qex.pop(state.audio_levels)
-
-        state = %{
-          state
-          | audio_levels_sum: state.audio_levels_sum - level,
-            audio_levels_count: state.audio_levels_count - 1,
-            audio_levels: audio_levels
-        }
-
-        {:cont, state}
-      else
-        {:halt, state}
-      end
-    end)
-  end
+    vad_estimation =
+      updated_audio_levels
+      |> AudioLevelQueue.to_list()
+      |> IsSpeakingEstimator.estimate_is_speaking(state.vad_threshold)
 
-  defp add_new_audio_level(state, level) do
-    audio_levels = Qex.push(state.audio_levels, {-level, state.current_timestamp})
+    actions = [buffer: {:output, buffer}] ++ maybe_send_event(vad_estimation, state)
 
-    %{
+    state = %{
       state
-      | audio_levels: audio_levels,
-        audio_levels_sum: state.audio_levels_sum + -level,
-        audio_levels_count: state.audio_levels_count + 1
+      | current_timestamp: rtp_timestamp,
+        audio_levels: updated_audio_levels,
+        vad: vad_estimation
     }
-  end
 
-  defp get_audio_levels_vad(state) do
-    if state.audio_levels_count >= state.min_packet_num and avg(state) >= state.vad_threshold,
-      do: :speech,
-      else: :silence
+    {actions, state}
   end
 
-  defp avg(state), do: state.audio_levels_sum / state.audio_levels_count
-
   defp maybe_send_event(audio_levels_vad, state) do
-    if vad_silence?(audio_levels_vad, state) or vad_speech?(audio_levels_vad, state) do
+    if vad_state_has_changed(state.vad, audio_levels_vad) do
       [event: {:output, %VadEvent{vad: audio_levels_vad}}]
     else
       []
     end
   end
 
-  defp update_vad_state(audio_levels_vad, state) do
-    cond do
-      vad_maybe_silence?(audio_levels_vad, state) ->
-        Map.merge(state, %{vad: :maybe_silence, vad_silence_timestamp: state.current_timestamp})
-
-      vad_silence?(audio_levels_vad, state) or vad_speech?(audio_levels_vad, state) ->
-        Map.merge(state, %{vad: audio_levels_vad})
-
-      true ->
-        state
-    end
-  end
-
-  defp vad_silence?(audio_levels_vad, state),
-    do: state.vad == :maybe_silence and audio_levels_vad == :silence and timer_expired?(state)
-
-  defp vad_speech?(audio_levels_vad, state) do
-    (state.vad == :maybe_silence and audio_levels_vad == :speech) or
-      (state.vad == :silence and audio_levels_vad == :speech)
-  end
-
-  defp vad_maybe_silence?(audio_levels_vad, state),
-    do: state.vad == :speech and audio_levels_vad == :silence
-
-  defp timer_expired?(state),
-    do: state.current_timestamp - state.vad_silence_timestamp > state.vad_silence_time
+  defp vad_state_has_changed(old_value, new_value), do: old_value != new_value
 end

lib/membrane/rtp/vad/audio_level_queue.ex

@@ -0,0 +1,64 @@
+defmodule Membrane.RTP.Vad.AudioLevelQueue do
+  @moduledoc false
+
+  # The queue contains audio levels for VAD implementation. It is used as an input of IsSpeakingEstimator.estimate_is_speaking.
+  # This structure builds on top of a simple FIFO Erlang queue by having a fixed max number of elements.
+
+  # The newest element in always appended to the front and popped out from its rear, so `to_list/1` returns the most recent element as the head of a list.
+  # The length of a list can be obtained in O(1) time.
+
+  alias Membrane.RTP.Vad.VadParams
+
+  @target_audio_level_length VadParams.target_levels_length()
+
+  @enforce_keys [:levels, :length]
+  defstruct [:levels, :length]
+
+  @typedoc """
+  A type for storing information about a fixed number of recent audio levels.
+
+  `:levels` - erlang queue which stores at most @target_audio_level_length elements
+  `:length` - number of elements
+  """
+
+  @type t() :: %__MODULE__{
+          levels: :queue.queue(non_neg_integer()),
+          length: non_neg_integer()
+        }
+
+  @doc """
+  Creates new AudioLevelQueue.
+  """
+  @spec new(Enum.t()) :: t()
+  def new(init_data \\ []) do
+    levels =
+      init_data
+      |> Enum.take(@target_audio_level_length)
+      |> Enum.to_list()
+      |> :queue.from_list()
+
+    %__MODULE__{levels: levels, length: :queue.len(levels)}
+  end
+
+  @doc """
+  Given a AudioLevelQueue and level value it returns a queue with the level value on front
+
+  The function also reduces the size of the queue if the maximum size has been reached.
+  It does so by dropping the oldest level.
+  """
+  @spec add(t(), non_neg_integer) :: t()
+  def add(%__MODULE__{length: @target_audio_level_length} = old_queue, level) do
+    levels = :queue.in_r(level, :queue.drop_r(old_queue.levels))
+    %__MODULE__{old_queue | levels: levels}
+  end
+
+  def add(%__MODULE__{levels: old_levels, length: length}, level) do
+    %__MODULE__{levels: :queue.in_r(level, old_levels), length: length + 1}
+  end
+
+  @doc """
+  Given an AudioLevelQueue it returns a list.
+  """
+  @spec to_list(t()) :: [non_neg_integer()]
+  def to_list(%__MODULE__{levels: levels}), do: :queue.to_list(levels)
+end