AV1 payloader

examples/send_from_file/example.exs

@@ -16,11 +16,13 @@ defmodule Peer do
     Media.IVFReader,
     PeerConnection,
     RTPCodecParameters,
-    RTP.VP8Payloader,
+    RTP.AV1Payloader,
     RTPTransceiver,
     SessionDescription
   }
 
+  @video_file "./av1_correct_long.ivf"
+
   @max_rtp_timestamp (1 <<< 32) - 1
 
   @ice_servers [
@@ -45,15 +47,11 @@ defmodule Peer do
         {:ok, pc} =
           PeerConnection.start_link(
             ice_servers: @ice_servers,
-            ice_ip_filter: ice_ip_filter,
             video_codecs: [
               %RTPCodecParameters{
-                payload_type: 96,
-                mime_type: "video/VP8",
-                clock_rate: 90_000,
-                channels: nil,
-                sdp_fmtp_line: nil,
-                rtcp_fbs: []
+                payload_type: 45,
+                mime_type: "video/AV1",
+                clock_rate: 90_000
               }
             ]
           )
@@ -117,16 +115,15 @@ defmodule Peer do
 
     case IVFReader.next_frame(state.ivf_reader) do
       {:ok, frame} ->
-        {rtp_packets, payloader} = VP8Payloader.payload(state.payloader, frame.data)
+        {rtp_packets, payloader} = AV1Payloader.payload(state.payloader, frame.data)
+
+        # the video has 30 FPS, VP8 clock rate is 90_000, so we have:
+        # 90_000 / 30 = 3_000
+        last_timestamp = state.last_timestamp + 3_000 &&& @max_rtp_timestamp
+
+        rtp_packets = Enum.map(rtp_packets, fn rtp_packet -> %{rtp_packet | timestamp: last_timestamp} end)
 
-        {rtp_packets, last_timestamp} =
-          Enum.map_reduce(rtp_packets, state.last_timestamp, fn rtp_packet, last_timestamp ->
-            # the video has 30 FPS, VP8 clock rate is 90_000, so we have:
-            # 90_000 / 30 = 3_000
-            last_timestamp = last_timestamp + 3_000 &&& @max_rtp_timestamp
-            rtp_packet = %{packet | timestamp: last_timestamp}
-            {rtp_packet, last_timestamp}
-          end)
+        # dbg(rtp_packets)
 
         Enum.each(rtp_packets, fn rtp_packet ->
           PeerConnection.send_rtp(state.peer_connection, state.track_id, rtp_packet)
@@ -137,7 +134,7 @@ defmodule Peer do
 
       :eof ->
         Logger.info("video.ivf ended. Looping...")
-        {:ok, ivf_reader} = IVFReader.open("./video.ivf")
+        {:ok, ivf_reader} = IVFReader.open(@video_file)
         {:ok, _header} = IVFReader.read_header(ivf_reader)
         state = %{state | ivf_reader: ivf_reader}
         {:noreply, state}
@@ -206,9 +203,9 @@ defmodule Peer do
   defp handle_webrtc_message({:connection_state_change, :connected} = msg, state) do
     Logger.info("#{inspect(msg)}")
     Logger.info("Starting sending video.ivf")
-    {:ok, ivf_reader} = IVFReader.open("./video.ivf")
+    {:ok, ivf_reader} = IVFReader.open(@video_file)
     {:ok, _header} = IVFReader.read_header(ivf_reader)
-    payloader = VP8Payloader.new(800)
+    payloader = AV1Payloader.new()
 
     Process.send_after(self(), :send_frame, 30)
     %{state | ivf_reader: ivf_reader, payloader: payloader}

lib/ex_webrtc/rtp/av1_payloader.ex

@@ -0,0 +1,123 @@
+defmodule ExWebRTC.RTP.AV1Payloader do
+  @moduledoc """
+  Encapsulates AV1 video frames into RTP packets.
+
+  https://norkin.org/research/av1_decoder_model/index.html
+  https://chromium.googlesource.com/external/webrtc/+/HEAD/modules/rtp_rtcp/source/video_rtp_depacketizer_av1.cc
+   AV1 format:
+
+   RTP payload syntax:
+       0 1 2 3 4 5 6 7
+      +-+-+-+-+-+-+-+-+
+      |Z|Y| W |N|-|-|-| (REQUIRED)
+      +=+=+=+=+=+=+=+=+ (REPEATED W-1 times, or any times if W = 0)
+      |1|             |
+      +-+ OBU fragment|
+      |1|             | (REQUIRED, leb128 encoded)
+      +-+    size     |
+      |0|             |
+      +-+-+-+-+-+-+-+-+
+      |  OBU fragment |
+      |     ...       |
+      +=+=+=+=+=+=+=+=+
+      |     ...       |
+      +=+=+=+=+=+=+=+=+ if W > 0, last fragment MUST NOT have size field
+      |  OBU fragment |
+      |     ...       |
+      +=+=+=+=+=+=+=+=+
+
+   OBU syntax:
+       0 1 2 3 4 5 6 7
+      +-+-+-+-+-+-+-+-+
+      |0| type  |X|S|-| (REQUIRED)
+      +-+-+-+-+-+-+-+-+
+   X: | TID |SID|-|-|-| (OPTIONAL)
+      +-+-+-+-+-+-+-+-+
+      |1|             |
+      +-+ OBU payload |
+   S: |1|             | (OPTIONAL, variable length leb128 encoded)
+      +-+    size     |
+      |0|             |
+      +-+-+-+-+-+-+-+-+
+      |  OBU payload  |
+      |     ...       |
+  """
+  import Bitwise
+
+  alias ExWebRTC.RTP.LEB128
+
+  @obu_sequence_header 1
+  @obu_temporal_delimiter 2
+
+  @opaque t() :: %__MODULE__{
+            max_payload_size: non_neg_integer()
+          }
+
+  defstruct [:max_payload_size]
+
+  @spec new(non_neg_integer()) :: t()
+  def new(max_payload_size \\ 1000) when max_payload_size > 100 do
+    %__MODULE__{max_payload_size: max_payload_size}
+  end
+
+  @doc """
+  Packs AV1 frame into one or more RTP packets.
+
+  Fields from RTP header like ssrc, timestamp etc. are set to 0.
+  """
+  @spec payload(t(), frame :: binary()) :: {[ExRTP.Packet.t()], t()}
+  def payload(payloader, frame) when frame != <<>> do
+    # obus = parse_obus(frame)
+    # for obu <- obus do
+    #   <<_::1, type::4, _::3, _rest::binary>> = obu
+    #   dbg(type)
+    # end
+    # dbg(:end)
+
+    # remove temporal delimiter
+    obus =
+      frame
+      |> parse_obus()
+      |> Enum.reject(fn obu ->
+        <<_::1, type::4, _::3, _rest::binary>> = obu
+        type == @obu_temporal_delimiter
+      end)
+
+    rtp_packets =
+      Enum.map(obus, fn obu ->
+        <<_::1, type::4, _::3, _rest::binary>> = obu
+        n_bit = if type == @obu_sequence_header, do: 1, else: 0
+
+        # obu = LEB128.encode(byte_size(obu)) <> obu
+
+        payload = <<0::1, 0::1, 1::2, n_bit::1, 0::3>> <> obu
+        ExRTP.Packet.new(payload, 0, 0, 0, 0)
+      end)
+
+    last_rtp_packet = List.last(rtp_packets)
+    last_rtp_packet = %{last_rtp_packet | marker: true}
+    rtp_packets = List.insert_at(rtp_packets, -1, last_rtp_packet)
+    {rtp_packets, payloader}
+  end
+
+  defp parse_obus(data, obus \\ [])
+  defp parse_obus(<<>>, obus), do: Enum.reverse(obus)
+  # X and S bits set
+  defp parse_obus(<<_::5, 1::1, 1::1, _::1, _::8, rest::binary>> = data, obus) do
+    {leb128_size, obu_payload_size} = LEB128.read(rest)
+    <<obu::binary-size(2 + leb128_size + obu_payload_size), rest::binary>> = data
+    parse_obus(rest, [obu | obus])
+  end
+
+  # X bit unset but S bit set
+  defp parse_obus(<<_::5, 0::1, 1::1, _::1, rest::binary>> = data, obus) do
+    {leb128_size, obu_payload_size} = LEB128.read(rest)
+    <<obu::binary-size(1 + leb128_size + obu_payload_size), rest::binary>> = data
+    parse_obus(rest, [obu | obus])
+  end
+
+  # S bit unset
+  defp parse_obus(<<_::5, _::1, 0::1, _::1, _rest::binary>> = data, obus) do
+    parse_obus(<<>>, [data | obus])
+  end
+end

lib/ex_webrtc/rtp/leb128.ex

@@ -0,0 +1,31 @@
+defmodule ExWebRTC.RTP.LEB128 do
+  import Bitwise
+
+  # see https://chromium.googlesource.com/external/webrtc/+/HEAD/modules/rtp_rtcp/source/rtp_packetizer_av1.cc#61
+  def encode(value, acc \\ [])
+
+  def encode(value, acc) when value < 0x80 do
+    acc = acc ++ [value]
+
+    for group <- acc, into: <<>> do
+      <<group>>
+    end
+  end
+
+  def encode(value, acc) do
+    group = 0x80 ||| (value &&& 0x7F)
+    acc = acc ++ [group]
+    encode(value >>> 7, acc)
+  end
+
+  # see https://source.chromium.org/chromium/chromium/src/+/main:third_party/webrtc/rtc_base/byte_buffer.cc;drc=8e78783dc1f7007bad46d657c9f332614e240fd8;l=107
+  def read(data, read_bits \\ 0, leb128_size \\ 0, value \\ 0)
+
+  def read(<<0::1, group::7, _rest::binary>>, read_bits, leb128_size, value) do
+    {leb128_size + 1, value ||| group <<< read_bits}
+  end
+
+  def read(<<1::1, group::7, rest::binary>>, read_bits, leb128_size, value) do
+    read(rest, read_bits + 7, leb128_size + 1, value ||| group <<< read_bits)
+  end
+end

test/ex_webrtc/rtp/av1_payloader_test.exs

@@ -0,0 +1,33 @@
+defmodule ExWebRTC.RTP.AV1PayloaderTest do
+  use ExUnit.Case, async: true
+
+  alias ExWebRTC.Media.IVFReader
+  alias ExWebRTC.RTP.AV1Payloader
+
+  @tag :debug
+  test "payload av1 video" do
+    # video frames in the fixture are mostly 500+ bytes
+    av1_payloader = AV1Payloader.new(200)
+    {:ok, ivf_reader} = IVFReader.open("test/fixtures/ivf/av1_correct.ivf")
+    {:ok, _header} = IVFReader.read_header(ivf_reader)
+
+    for _i <- 0..28, reduce: av1_payloader do
+      av1_payloader ->
+        {:ok, frame} = IVFReader.next_frame(ivf_reader)
+        {rtp_packets, av1_payloader} = AV1Payloader.payload(av1_payloader, frame.data)
+
+        # assert all packets but last are 200 bytes
+        # rtp_packets
+        # |> Enum.slice(0, length(rtp_packets) - 1)
+        # |> Enum.each(fn rtp_packet ->
+        #   assert byte_size(rtp_packet.payload) == 200
+        # end)
+
+        # last_rtp = List.last(rtp_packets)
+        # assert byte_size(last_rtp.payload) < 200
+        # assert last_rtp.marker == true
+
+        av1_payloader
+    end
+  end
+end

test/ex_webrtc/rtp/leb128_test.exs

@@ -0,0 +1,17 @@
+defmodule ExWebrtc.RTP.LEB128Test do
+  use ExUnit.Case, async: true
+
+  alias ExWebRTC.RTP.LEB128
+
+  test "encode" do
+    assert <<0>> == LEB128.encode(0)
+    assert <<5>> == LEB128.encode(5)
+    assert <<0xBF, 0x84, 0x3D>> == LEB128.encode(999_999)
+  end
+
+  test "read" do
+    assert {1, 0} == LEB128.read(<<0>>)
+    assert {1, 5} == LEB128.read(<<5>>)
+    assert {3, 999_999} == LEB128.read(<<0xBF, 0x84, 0x3D>>)
+  end
+end