Ogg writer (#50)

examples/send_from_file/example.exs

@@ -161,7 +161,7 @@ defmodule Peer do
         Process.send_after(self(), :send_audio_packet, duration)
         # values set to 0 are handled by PeerConnection.set_rtp
         rtp_packet = OpusPayloader.payload(packet)
-        rtp_packet = %{rtp_packet | timestamp: state.last_audio_timestamp}
+        rtp_packet = %{rtp_packet | timestamp: trunc(state.last_audio_timestamp)}
         PeerConnection.send_rtp(state.peer_connection, state.audio_track_id, rtp_packet)
 
         # OggReader.next_packet/1 returns duration in ms

lib/ex_webrtc/media/ogg/header.ex

@@ -0,0 +1,47 @@
+defmodule ExWebRTC.Media.Ogg.Header do
+  @moduledoc false
+  # based on RFC 7845, sec. 5
+
+  @id_signature "OpusHead"
+  @comment_signature "OpusTags"
+
+  @vendor "elixir-webrtc"
+
+  @default_preskip 3840
+  @default_gain 0
+  # mono or stereo
+  @channel_mapping 0
+
+  # for now, we ignore the Ogg/Opus header when decoding
+  @spec decode_id(binary()) :: :ok | {:error, term()}
+  def decode_id(<<@id_signature, _rest::binary>>), do: :ok
+  def decode_id(_packet), do: {:error, :invalid_id_header}
+
+  @spec decode_id(binary()) :: :ok | {:error, term()}
+  def decode_comment(<<@comment_signature, _rest::binary>>), do: :ok
+  def decode_commend(_packet), do: {:error, :invalid_comment_header}
+
+  @spec create_id(non_neg_integer(), non_neg_integer()) :: binary()
+  def create_id(sample_rate, channel_count) do
+    <<
+      @id_signature,
+      1,
+      channel_count,
+      @default_preskip::little-16,
+      sample_rate::little-32,
+      @default_gain::little-16,
+      @channel_mapping
+    >>
+  end
+
+  @spec create_comment() :: binary()
+  def create_comment do
+    <<
+      @comment_signature,
+      byte_size(@vendor)::little-32,
+      @vendor,
+      # no additional user comments
+      0
+    >>
+  end
+end

lib/ex_webrtc/media/ogg/page.ex

@@ -0,0 +1,162 @@
+defmodule ExWebRTC.Media.Ogg.Page do
+  @moduledoc false
+  # see RFC 3553, sec. 6 for description of the Ogg Page
+
+  import Bitwise
+
+  @crc_params %{
+    extend: :crc_32,
+    poly: 0x04C11DB7,
+    init: 0x0,
+    xorout: 0x0,
+    refin: false,
+    refout: false
+  }
+
+  @signature "OggS"
+  @version 0
+
+  @type t() :: %__MODULE__{
+          continued?: boolean(),
+          first?: boolean(),
+          last?: boolean(),
+          granule_pos: non_neg_integer(),
+          serial_no: non_neg_integer(),
+          sequence_no: non_neg_integer(),
+          packets: [binary()],
+          rest: binary()
+        }
+
+  @enforce_keys [:granule_pos, :serial_no, :sequence_no]
+  defstruct @enforce_keys ++
+              [
+                continued?: false,
+                first?: false,
+                last?: false,
+                packets: [],
+                rest: <<>>
+              ]
+
+  @spec read(File.io_device()) :: {:ok, t()} | {:error, term()}
+  def read(file) do
+    with <<@signature, @version, type, granule_pos::little-64, serial_no::little-32,
+           sequence_no::little-32, _checksum::little-32,
+           segment_no>> = header <- IO.binread(file, 27),
+         raw_segment_table when is_binary(raw_segment_table) <- IO.binread(file, segment_no),
+         segment_table <- :binary.bin_to_list(raw_segment_table),
+         payload_length <- Enum.sum(segment_table),
+         payload when is_binary(payload) <- IO.binread(file, payload_length),
+         :ok <- verify_checksum(header <> raw_segment_table <> payload) do
+      {packets, rest} = split_packets(segment_table, payload)
+
+      page = %__MODULE__{
+        continued?: (type &&& 0x01) != 0,
+        first?: (type &&& 0x02) != 0,
+        last?: (type &&& 0x04) != 0,
+        granule_pos: granule_pos,
+        serial_no: serial_no,
+        sequence_no: sequence_no,
+        packets: packets,
+        rest: rest
+      }
+
+      {:ok, page}
+    else
+      data when is_binary(data) -> {:error, :invalid_page_header}
+      :eof -> :eof
+      {:error, _res} = err -> err
+    end
+  end
+
+  @spec write(File.io_device(), t()) :: :ok | {:error, term()}
+  def write(file, %__MODULE__{} = page) do
+    with {:ok, segment_table} <- create_segment_table(page.packets, page.rest) do
+      continued = if page.continued?, do: 0x01, else: 0
+      first = if page.first?, do: 0x02, else: 0
+      last = if page.last?, do: 0x04, else: 0
+      type = first ||| continued ||| last
+
+      before_crc = <<
+        @signature,
+        @version,
+        type,
+        page.granule_pos::little-64,
+        page.serial_no::little-32,
+        page.sequence_no::little-32
+      >>
+
+      after_crc =
+        <<length(segment_table)>> <>
+          :binary.list_to_bin(segment_table) <>
+          :binary.list_to_bin(page.packets) <>
+          page.rest
+
+      checksum = CRC.calculate(<<before_crc::binary, 0::32, after_crc::binary>>, @crc_params)
+      packet = <<before_crc::binary, checksum::little-32, after_crc::binary>>
+
+      IO.binwrite(file, packet)
+    end
+  end
+
+  defp verify_checksum(<<start::binary-22, checksum::little-32, rest::binary>>) do
+    actual_checksum =
+      <<start::binary, 0::32, rest::binary>>
+      |> CRC.calculate(@crc_params)
+
+    if checksum == actual_checksum do
+      :ok
+    else
+      {:error, :invalid_checksum}
+    end
+  end
+
+  defp split_packets(segment_table, payload, packets \\ [], packet \\ <<>>)
+  defp split_packets([], <<>>, packets, packet), do: {Enum.reverse(packets), packet}
+
+  defp split_packets([segment_len | segment_table], payload, packets, packet) do
+    <<segment::binary-size(segment_len), rest::binary>> = payload
+    packet = packet <> segment
+
+    case segment_len do
+      255 -> split_packets(segment_table, rest, packets, packet)
+      _len -> split_packets(segment_table, rest, [packet | packets], <<>>)
+    end
+  end
+
+  defp create_segment_table(packets, rest) when rem(byte_size(rest), 255) == 0 do
+    # normally packet of length that is a multiple of 255 would end with 0-lenght segment
+    # for the rest (split packet) we don't want that
+    rest_segments =
+      case segment_packet(rest) do
+        [0 | segments] -> segments
+        [] -> []
+      end
+
+    segment_table =
+      packets
+      |> Enum.reduce([], fn packet, segments ->
+        segment_packet(packet) ++ segments
+      end)
+      |> then(&Enum.concat(rest_segments, &1))
+      |> Enum.reverse()
+
+    if length(segment_table) > 255 do
+      {:error, :too_many_segments}
+    else
+      {:ok, segment_table}
+    end
+  end
+
+  defp create_segment_table(_packets, _rest), do: {:error, :rest_too_short}
+
+  # returned segment table for the packet is reversed
+  # thus the Enum.reverse/1 call in create_segment_table/2
+  defp segment_packet(packet, acc \\ [])
+  defp segment_packet(<<>>, [255 | _rest] = acc), do: [0 | acc]
+  defp segment_packet(<<>>, acc), do: acc
+
+  defp segment_packet(<<_seg::binary-255, rest::binary>>, acc),
+    do: segment_packet(rest, [255 | acc])
+
+  defp segment_packet(packet, acc) when is_binary(packet), do: [byte_size(packet) | acc]
+end

lib/ex_webrtc/media/ogg_reader.ex

@@ -1,6 +1,6 @@
 defmodule ExWebRTC.Media.OggReader do
   @moduledoc """
-  Defines Ogg reader.
+  Reads Opus packets from an Ogg container file.
 
   For now, works only with single Opus stream in the container.
 
@@ -10,21 +10,8 @@ defmodule ExWebRTC.Media.OggReader do
   * [RFC 6716: Definition of the Opus Audio Codec](https://www.rfc-editor.org/rfc/rfc6716.txt)
   """
 
-  import Bitwise
-
-  @crc_params %{
-    extend: :crc_32,
-    poly: 0x04C11DB7,
-    init: 0x0,
-    xorout: 0x0,
-    refin: false,
-    refout: false
-  }
-
-  @signature "OggS"
-  @id_signature "OpusHead"
-  @comment_signature "OpusTags"
-  @version 0
+  alias ExWebRTC.Media.Ogg.{Header, Page}
+  alias ExWebRTC.Media.Opus
 
   @opaque t() :: %{
             file: File.io_device(),
@@ -38,17 +25,18 @@ defmodule ExWebRTC.Media.OggReader do
   For now, works only with single Opus stream in the container.
   This function reads the ID and Comment Headers (and, for now, ignores them).
   """
-  @spec open(Path.t()) :: {:ok, t()} | {:error, File.posix() | :invalid_header}
+  @spec open(Path.t()) :: {:ok, t()} | {:error, term()}
   def open(path) do
     with {:ok, file} <- File.open(path),
          reader <- %{file: file, packets: [], rest: <<>>},
-         # for now, we ignore ID Header and Comment Header
-         {:ok, <<@id_signature, _rest::binary>>, reader} <- do_next_packet(reader),
-         {:ok, <<@comment_signature, _rest::binary>>, reader} <- do_next_packet(reader) do
+         {:ok, id_header, reader} <- do_next_packet(reader),
+         {:ok, comment_header, reader} <- do_next_packet(reader),
+         :ok <- Header.decode_id(id_header),
+         :ok <- Header.decode_comment(comment_header) do
       {:ok, reader}
     else
+      :eof -> {:error, :invalid_file}
       {:error, _res} = err -> err
-      _other -> {:error, :invalid_header}
     end
   end
 
@@ -59,13 +47,10 @@ defmodule ExWebRTC.Media.OggReader do
   This function also returns the duration of the audio in milliseconds, based on Opus packet TOC sequence (see RFC 6716, sec. 3).
   It assumes that all of the Ogg packets belong to the same stream.
   """
-  @spec next_packet(t()) ::
-          {:ok, {binary(), non_neg_integer()}, t()}
-          | {:error, :invalid_page_header | :not_enough_data}
-          | :eof
+  @spec next_packet(t()) :: {:ok, {binary(), non_neg_integer()}, t()} | {:error, term()} | :eof
   def next_packet(reader) do
     with {:ok, packet, reader} <- do_next_packet(reader),
-         {:ok, duration} <- get_packet_duration(packet) do
+         {:ok, duration} <- Opus.duration(packet) do
       {:ok, {packet, duration}, reader}
     end
   end
@@ -75,7 +60,7 @@ defmodule ExWebRTC.Media.OggReader do
   end
 
   defp do_next_packet(%{packets: []} = reader) do
-    with {:ok, _header, packets, rest} <- read_page(reader.file) do
+    with {:ok, %Page{packets: packets, rest: rest}} <- Page.read(reader.file) do
       case packets do
         [] ->
           do_next_packet(%{reader | packets: [], rest: reader.rest <> rest})
@@ -87,82 +72,4 @@ defmodule ExWebRTC.Media.OggReader do
       end
     end
   end
-
-  defp read_page(file) do
-    with <<@signature, @version, type, granule_pos::little-64, serial_no::little-32,
-           sequence_no::little-32, _checksum::little-32,
-           segment_no>> = header <- IO.binread(file, 27),
-         raw_segment_table when is_binary(raw_segment_table) <- IO.binread(file, segment_no),
-         segment_table <- :binary.bin_to_list(raw_segment_table),
-         payload_length <- Enum.sum(segment_table),
-         payload when is_binary(payload) <- IO.binread(file, payload_length),
-         :ok <- verify_checksum(header <> raw_segment_table <> payload) do
-      {packets, rest} = split_packets(segment_table, payload)
-
-      type = %{
-        fresh?: (type &&& 0x01) != 0,
-        first?: (type &&& 0x02) != 0,
-        last?: (type &&& 0x04) != 0
-      }
-
-      {:ok,
-       %{
-         type: type,
-         granule_pos: granule_pos,
-         serial_no: serial_no,
-         sequence_no: sequence_no
-       }, packets, rest}
-    else
-      data when is_binary(data) -> {:error, :invalid_page_header}
-      :eof -> :eof
-      {:error, _res} = err -> err
-    end
-  end
-
-  defp verify_checksum(<<start::binary-22, checksum::little-32, rest::binary>>) do
-    actual_checksum =
-      <<start::binary, 0::32, rest::binary>>
-      |> CRC.calculate(@crc_params)
-
-    if checksum == actual_checksum do
-      :ok
-    else
-      {:error, :invalid_checksum}
-    end
-  end
-
-  defp split_packets(segment_table, payload, packets \\ [], packet \\ <<>>)
-  defp split_packets([], <<>>, packets, packet), do: {Enum.reverse(packets), packet}
-
-  defp split_packets([segment_len | segment_table], payload, packets, packet) do
-    <<segment::binary-size(segment_len), rest::binary>> = payload
-    packet = packet <> segment
-
-    case segment_len do
-      255 -> split_packets(segment_table, rest, packets, packet)
-      _len -> split_packets(segment_table, rest, [packet | packets], <<>>)
-    end
-  end
-
-  # computes how much audio Opus packet contains (in ms), based on the TOC sequence
-  # RFC 6716, sec. 3
-  defp get_packet_duration(<<config::5, rest::bitstring>>) do
-    with {:ok, frame_count} <- get_frame_count(rest) do
-      {:ok, trunc(frame_count * get_frame_duration(config))}
-    end
-  end
-
-  defp get_packet_duration(_other), do: {:error, :not_enough_data}
-
-  defp get_frame_count(<<_s::1, 0::2, _rest::binary>>), do: {:ok, 1}
-  defp get_frame_count(<<_s::1, c::2, _rest::binary>>) when c in 1..2, do: {:ok, 2}
-  defp get_frame_count(<<_s::1, 3::2, _vp::2, frame_no::5, _rest::binary>>), do: {:ok, frame_no}
-  defp get_frame_count(_other), do: {:error, :not_enough_data}
-
-  defp get_frame_duration(config) when config in [16, 20, 24, 28], do: 2.5
-  defp get_frame_duration(config) when config in [17, 21, 25, 29], do: 5
-  defp get_frame_duration(config) when config in [0, 4, 8, 12, 14, 18, 22, 26, 30], do: 10
-  defp get_frame_duration(config) when config in [1, 5, 9, 13, 15, 19, 23, 27, 31], do: 20
-  defp get_frame_duration(config) when config in [2, 6, 10], do: 40
-  defp get_frame_duration(config) when config in [3, 7, 11], do: 60
 end