Add how_much_to_skip_ms option (#5)

* Add `how_much_to_skip_ms` option
* Add `Client.get_how_much_truly_skipped_ms/1`
* Fix timestamp calculation in MPEG-TS demuxing engine

Author: varsill

lib/ex_hls/client.ex

@@ -21,6 +21,8 @@ defmodule ExHLS.Client do
     :queues,
     :timestamp_offsets,
     :last_timestamps,
+    :how_much_to_skip_ms,
+    :how_much_truly_skipped_ms,
     :end_stream_executed?
   ]
 
@@ -44,8 +46,8 @@ defmodule ExHLS.Client do
   By default, it uses `DemuxingEngine.MPEGTS` as the demuxing engine implementation.
   """
 
-  @spec new(String.t()) :: client()
-  def new(url) do
+  @spec new(String.t(), non_neg_integer()) :: client()
+  def new(url, how_much_to_skip_ms \\ 0) do
     %{status: 200, body: request_body} = Req.get!(url)
     multivariant_playlist = request_body |> ExM3U8.deserialize_multivariant_playlist!([])
 
@@ -61,6 +63,8 @@ defmodule ExHLS.Client do
       queues: %{audio: Qex.new(), video: Qex.new()},
       timestamp_offsets: %{audio: nil, video: nil},
       last_timestamps: %{audio: nil, video: nil},
+      how_much_to_skip_ms: how_much_to_skip_ms,
+      how_much_truly_skipped_ms: nil,
       end_stream_executed?: false
     }
   end
@@ -87,14 +91,16 @@ defmodule ExHLS.Client do
   defp ensure_media_playlist_loaded(client), do: client
 
   defp read_media_playlist_without_variant(%__MODULE__{media_playlist: nil} = client) do
-    deserialized_media_playlist =
+    {deserialized_media_playlist, how_much_truly_skipped_ms} =
       client.root_playlist_string
       |> ExM3U8.deserialize_media_playlist!([])
+      |> skip_to_how_much_to_skip(client.how_much_to_skip_ms)
 
     %{
       client
       | media_playlist: deserialized_media_playlist,
-        media_base_url: client.base_url
+        media_base_url: client.base_url,
+        how_much_truly_skipped_ms: how_much_truly_skipped_ms
     }
   end
 
@@ -121,15 +127,17 @@ defmodule ExHLS.Client do
 
     media_playlist = Path.join(client.base_url, chosen_variant.uri) |> Req.get!()
 
-    deserialized_media_playlist =
+    {deserialized_media_playlist, how_much_truly_skipped_ms} =
       ExM3U8.deserialize_media_playlist!(media_playlist.body, [])
+      |> skip_to_how_much_to_skip(client.how_much_to_skip_ms)
 
     media_base_url = Path.join(client.base_url, Path.dirname(chosen_variant.uri))
 
     %{
       client
       | media_playlist: deserialized_media_playlist,
-        media_base_url: media_base_url
+        media_base_url: media_base_url,
+        how_much_truly_skipped_ms: how_much_truly_skipped_ms
     }
   end
 
@@ -296,7 +304,8 @@ defmodule ExHLS.Client do
     %{
       client
       | demuxing_engine_impl: demuxing_engine_impl,
-        demuxing_engine: demuxing_engine_impl.new()
+        # how_much_truly_skipped_ms is the timestamps offset of the first non-discarded sample
+        demuxing_engine: get_how_much_truly_skipped_ms(client) |> demuxing_engine_impl.new()
     }
   end
 
@@ -316,4 +325,48 @@ defmodule ExHLS.Client do
       {:error, _reason} -> :error
     end
   end
+
+  defp skip_to_how_much_to_skip(media_playlist, how_much_to_skip_ms) do
+    {discarded, timeline_with_cumulative_duration} =
+      Enum.map_reduce(
+        media_playlist.timeline,
+        0,
+        fn
+          %ExM3U8.Tags.Segment{} = chunk, cumulative_duration_ms ->
+            chunk_end_ms = cumulative_duration_ms + 1000 * chunk.duration
+            {{chunk, chunk_end_ms}, chunk_end_ms}
+
+          other_tag, cumulative_duration_ms ->
+            {{other_tag, cumulative_duration_ms}, cumulative_duration_ms}
+        end
+      )
+      |> elem(0)
+      |> Enum.split_with(fn
+        {%ExM3U8.Tags.Segment{}, chunk_end_ms} -> chunk_end_ms < how_much_to_skip_ms
+        _other -> false
+      end)
+
+    how_much_truly_skipped_ms =
+      case List.last(discarded) do
+        nil -> 0
+        {_discarded_timeline, cumulative_duration_ms} -> cumulative_duration_ms
+      end
+      |> round()
+
+    timeline = Enum.map(timeline_with_cumulative_duration, &elem(&1, 0))
+
+    {put_in(media_playlist.timeline, timeline), how_much_truly_skipped_ms}
+  end
+
+  @spec get_how_much_truly_skipped_ms(client()) :: non_neg_integer() | no_return()
+  def get_how_much_truly_skipped_ms(%{how_much_truly_skipped_ms: nil}) do
+    raise """
+    `how_much_truly_skipped_ms` is not yet available.
+    Please call `read_audio_chunk/1`, `read_video_chunk/1`
+    or `choose_variant/2` before calling this function.
+    """
+  end
+
+  def get_how_much_truly_skipped_ms(%{how_much_truly_skipped_ms: how_much_truly_skipped_ms}),
+    do: how_much_truly_skipped_ms
 end

lib/ex_hls/demuxing_engine.ex

@@ -3,7 +3,7 @@ defmodule ExHLS.DemuxingEngine do
 
   @type t :: any()
 
-  @callback new() :: t()
+  @callback new(base_timestamps_ms :: non_neg_integer()) :: t()
 
   @callback feed!(t(), binary()) :: t()
 

lib/ex_hls/demuxing_engine/cmaf.ex

@@ -3,19 +3,22 @@ defmodule ExHLS.DemuxingEngine.CMAF do
   @behaviour ExHLS.DemuxingEngine
 
   alias Membrane.MP4.Demuxer.CMAF
+  alias Membrane.MP4.Demuxer.Sample
 
-  @enforce_keys [:demuxer]
+  @enforce_keys [:demuxer, :timestamps_offset_ms]
   defstruct @enforce_keys ++ [tracks_to_chunks: %{}]
 
   @type t :: %__MODULE__{
           demuxer: CMAF.Engine.t(),
-          tracks_to_chunks: map()
+          tracks_to_chunks: map(),
+          timestamps_offset_ms: non_neg_integer()
         }
 
   @impl true
-  def new() do
+  def new(timestamps_offset_ms) do
     %__MODULE__{
-      demuxer: CMAF.Engine.new()
+      demuxer: CMAF.Engine.new(),
+      timestamps_offset_ms: timestamps_offset_ms
     }
   end
 
@@ -30,7 +33,7 @@ defmodule ExHLS.DemuxingEngine.CMAF do
       chunks
       |> Enum.group_by(
         fn chunk -> chunk.track_id end,
-        fn %CMAF.Engine.Sample{} = chunk ->
+        fn %Sample{} = chunk ->
           %ExHLS.Chunk{
             payload: chunk.payload,
             pts_ms: chunk.pts,
@@ -66,6 +69,7 @@ defmodule ExHLS.DemuxingEngine.CMAF do
     with qex when qex != nil <- demuxing_engine.tracks_to_chunks[track_id],
          {{:value, chunk}, popped_qex} <- Qex.pop(qex) do
       demuxing_engine = put_in(demuxing_engine.tracks_to_chunks[track_id], popped_qex)
+      chunk = normalize_timestamps(chunk, demuxing_engine.timestamps_offset_ms)
       {:ok, chunk, demuxing_engine}
     else
       nil -> {:error, :unknown_track, demuxing_engine}
@@ -75,4 +79,12 @@ defmodule ExHLS.DemuxingEngine.CMAF do
 
   @impl true
   def end_stream(demuxing_engine), do: demuxing_engine
+
+  defp normalize_timestamps(chunk, timestamps_offset_ms) do
+    %{
+      chunk
+      | pts_ms: round(chunk.pts_ms + timestamps_offset_ms),
+        dts_ms: round(chunk.dts_ms + timestamps_offset_ms)
+    }
+  end
 end

lib/ex_hls/demuxing_engine/mpeg_ts.ex

@@ -14,7 +14,9 @@ defmodule ExHLS.DemuxingEngine.MPEGTS do
         }
 
   @impl true
-  def new() do
+  # We can ignore handling timestamps_offset_ms since timestamps in the
+  # MPEG-TS container already include global timestamps.
+  def new(_timestamps_offset_ms) do
     demuxer = Demuxer.new()
 
     # we need to explicitly override that `waiting_random_access_indicator` as otherwise Demuxer
@@ -44,10 +46,19 @@ defmodule ExHLS.DemuxingEngine.MPEGTS do
             [{id, %RemoteStream{content_format: H264}}]
 
           {id, unsupported_stream_info} ->
-            Logger.warning("""
-            #{__MODULE__ |> inspect()}: dropping unsupported stream with id #{id |> inspect()}.\
-            Stream info: #{unsupported_stream_info |> inspect(pretty: true)}
-            """)
+            unsupported_streams = Process.get(:unsupported_streams) || MapSet.new()
+
+            if unsupported_stream_info.stream_type not in unsupported_streams do
+              Logger.debug("""
+              #{__MODULE__ |> inspect()}: dropping unsupported stream with id #{id |> inspect()}.\
+              Stream info: #{unsupported_stream_info |> inspect(pretty: true)}
+              """)
+
+              unsupported_streams =
+                MapSet.put(unsupported_streams, unsupported_stream_info.stream_type)
+
+              Process.put(:unsupported_streams, unsupported_streams)
+            end
 
             []
         end)
@@ -80,8 +91,8 @@ defmodule ExHLS.DemuxingEngine.MPEGTS do
     end
   end
 
-  @mpegts_clock_rate 90
-  defp packet_ts_to_millis(ts), do: div(ts, @mpegts_clock_rate)
+  # value returned by Demuxer is represented in nanoseconds
+  defp packet_ts_to_millis(ts), do: div(ts, 1_000_000)
 
   @impl true
   def end_stream(%__MODULE__{} = demuxing_engine) do

mix.exs

@@ -40,10 +40,9 @@ defmodule ExHLS.Mixfile do
       {:ex_m3u8, "~> 0.15.3"},
       {:req, "~> 0.5.10"},
       {:qex, "~> 0.5.1"},
-      {:membrane_mp4_plugin, "~> 0.35.3"},
+      {:membrane_mp4_plugin, "~> 0.36.0"},
       {:membrane_h26x_plugin, "~> 0.10.2"},
-      # {:mpeg_ts, github: "kim-company/kim_mpeg_ts"},
-      {:mpeg_ts, github: "membraneframework-labs/kim_mpeg_ts", branch: "backport-v1.0.3"},
+      {:mpeg_ts, github: "kim-company/kim_mpeg_ts"},
       {:ex_doc, ">= 0.0.0", only: :dev, runtime: false},
       {:dialyxir, ">= 0.0.0", only: :dev, runtime: false},
       {:credo, ">= 0.0.0", only: :dev, runtime: false}

mix.lock

@@ -26,12 +26,12 @@
   "membrane_h265_format": {:hex, :membrane_h265_format, "0.2.0", "1903c072cf7b0980c4d0c117ab61a2cd33e88782b696290de29570a7fab34819", [:mix], [], "hexpm", "6df418bdf242c0d9f7dbf2e5aea4c2d182e34ac9ad5a8b8cef2610c290002e83"},
   "membrane_h26x_plugin": {:hex, :membrane_h26x_plugin, "0.10.5", "e9fa1ee9cda944259c4d2728c8b279bfe0152a3a6c1af187b07fa8411ca4e25e", [:mix], [{:bunch, "~> 1.4", [hex: :bunch, repo: "hexpm", optional: false]}, {:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}, {:membrane_h264_format, "~> 0.6.0", [hex: :membrane_h264_format, repo: "hexpm", optional: false]}, {:membrane_h265_format, "~> 0.2.0", [hex: :membrane_h265_format, repo: "hexpm", optional: false]}], "hexpm", "dd0287a6b6223e47bba30a8952d6ec53db35f6a3e33203b7ad786e995711f098"},
   "membrane_mp4_format": {:hex, :membrane_mp4_format, "0.8.0", "8c6e7d68829228117d333b4fbb030e7be829aab49dd8cb047fdc664db1812e6a", [:mix], [], "hexpm", "148dea678a1f82ccfd44dbde6f936d2f21255f496cb45a22cc6eec427f025522"},
-  "membrane_mp4_plugin": {:hex, :membrane_mp4_plugin, "0.35.3", "80228f4332eeef4fce4d90184a82bd5869d184f78438419660da7dc91871a238", [:mix], [{:bunch, "~> 1.5", [hex: :bunch, repo: "hexpm", optional: false]}, {:membrane_aac_format, "~> 0.8.0", [hex: :membrane_aac_format, repo: "hexpm", optional: false]}, {:membrane_cmaf_format, "~> 0.7.0", [hex: :membrane_cmaf_format, repo: "hexpm", optional: false]}, {:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}, {:membrane_file_plugin, "~> 0.17.0", [hex: :membrane_file_plugin, repo: "hexpm", optional: false]}, {:membrane_h264_format, "~> 0.6.1", [hex: :membrane_h264_format, repo: "hexpm", optional: false]}, {:membrane_h265_format, "~> 0.2.0", [hex: :membrane_h265_format, repo: "hexpm", optional: false]}, {:membrane_mp4_format, "~> 0.8.0", [hex: :membrane_mp4_format, repo: "hexpm", optional: false]}, {:membrane_opus_format, "~> 0.3.0", [hex: :membrane_opus_format, repo: "hexpm", optional: false]}, {:membrane_timestamp_queue, "~> 0.2.1", [hex: :membrane_timestamp_queue, repo: "hexpm", optional: false]}], "hexpm", "c6d8b20e49540329f246e9a9c69adae330d424802fdfa1e6485d76a5257e6169"},
+  "membrane_mp4_plugin": {:hex, :membrane_mp4_plugin, "0.36.0", "ef4fc6fc72cc439974f2ae0ce579f3dbf8674f02ab00f7f524a6600b786a9ca8", [:mix], [{:bunch, "~> 1.5", [hex: :bunch, repo: "hexpm", optional: false]}, {:membrane_aac_format, "~> 0.8.0", [hex: :membrane_aac_format, repo: "hexpm", optional: false]}, {:membrane_cmaf_format, "~> 0.7.0", [hex: :membrane_cmaf_format, repo: "hexpm", optional: false]}, {:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}, {:membrane_file_plugin, "~> 0.17.0", [hex: :membrane_file_plugin, repo: "hexpm", optional: false]}, {:membrane_h264_format, "~> 0.6.1", [hex: :membrane_h264_format, repo: "hexpm", optional: false]}, {:membrane_h265_format, "~> 0.2.0", [hex: :membrane_h265_format, repo: "hexpm", optional: false]}, {:membrane_mp4_format, "~> 0.8.0", [hex: :membrane_mp4_format, repo: "hexpm", optional: false]}, {:membrane_opus_format, "~> 0.3.0", [hex: :membrane_opus_format, repo: "hexpm", optional: false]}, {:membrane_timestamp_queue, "~> 0.2.1", [hex: :membrane_timestamp_queue, repo: "hexpm", optional: false]}], "hexpm", "84f55a42c69cb557b73d6272f958812f607abaaa6a3473f301d22393f2a62808"},
   "membrane_opus_format": {:hex, :membrane_opus_format, "0.3.0", "3804d9916058b7cfa2baa0131a644d8186198d64f52d592ae09e0942513cb4c2", [:mix], [], "hexpm", "8fc89c97be50de23ded15f2050fe603dcce732566fe6fdd15a2de01cb6b81afe"},
   "membrane_timestamp_queue": {:hex, :membrane_timestamp_queue, "0.2.2", "1c831b2273d018a6548654aa9f7fa7c4b683f71d96ffe164934ef55f9d11f693", [:mix], [{:heap, "~> 2.0", [hex: :heap, repo: "hexpm", optional: false]}, {:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}], "hexpm", "7c830e760baaced0988421671cd2c83c7cda8d1bd2b61fd05332711675d1204f"},
   "mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"},
   "mint": {:hex, :mint, "1.7.1", "113fdb2b2f3b59e47c7955971854641c61f378549d73e829e1768de90fc1abf1", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1 or ~> 0.2.0 or ~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "fceba0a4d0f24301ddee3024ae116df1c3f4bb7a563a731f45fdfeb9d39a231b"},
-  "mpeg_ts": {:git, "https://github.com/membraneframework-labs/kim_mpeg_ts.git", "8c036fca6558a4339033a5a8697ebf147728f36b", [branch: "backport-v1.0.3"]},
+  "mpeg_ts": {:git, "https://github.com/kim-company/kim_mpeg_ts.git", "ebde017bf639a9d57d068f029f2e43562ae78746", []},
   "nimble_options": {:hex, :nimble_options, "1.1.1", "e3a492d54d85fc3fd7c5baf411d9d2852922f66e69476317787a7b2bb000a61b", [:mix], [], "hexpm", "821b2470ca9442c4b6984882fe9bb0389371b8ddec4d45a9504f00a66f650b44"},
   "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"},
   "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"},

test/client_test.exs

@@ -146,4 +146,46 @@ defmodule Client.Test do
              72, 183, 150, 44, 216, 32, 217, 35, 238, 239, 120, 50, 54, 52, 32, 45, 32, 99, 111,
              114, 101, 32, 49, 54, 52, 32, 114>> <> _rest = video_chunk.payload
   end
+
+  test "(MPEGTS) stream with how_much_to_skip_ms" do
+    how_much_to_skip_ms = 44_000
+    client = Client.new(@mpegts_url, how_much_to_skip_ms)
+
+    variant_720 =
+      Client.get_variants(client)
+      |> Map.values()
+      |> Enum.find(&(&1.resolution == {1280, 720}))
+
+    assert variant_720 != nil
+
+    client = client |> Client.choose_variant(variant_720.id)
+    {:ok, tracks_info, client} = Client.get_tracks_info(client)
+
+    tracks_info = tracks_info |> Map.values()
+
+    assert tracks_info |> length() == 2
+    assert %RemoteStream{content_format: AAC, type: :bytestream} in tracks_info
+    assert %RemoteStream{content_format: H264, type: :bytestream} in tracks_info
+
+    {video_chunk, client} = client |> Client.read_video_chunk()
+
+    # segments in the fixture are 10s long and
+    # the timestamps offset is 10s, so the first
+    # video pts after skipping initial 44 seconds should be div(10+44, 10) = 50s
+    assert %{pts_ms: 50_033, dts_ms: 50_000} = video_chunk
+    assert byte_size(video_chunk.payload) == 135_298
+
+    assert <<0, 0, 0, 1, 9, 240, 0, 0, 0, 1, 103, 100, 0, 31, 172, 217, 128, 80, 5, 187, 1, 16, 0,
+             0, 3, 0, 16, 0, 0, 7, 128, 241, 131, 25, 160, 0, 0, 0, 1, 104, 233, 121, 203, 34,
+             192, 0, 0, 1, 101, 136>> <> _rest = video_chunk.payload
+
+    {audio_chunk, _client} = Client.read_audio_chunk(client)
+
+    assert %{pts_ms: 50_018, dts_ms: 50_018} = audio_chunk
+    assert byte_size(audio_chunk.payload) == 6020
+
+    assert <<255, 241, 80, 128, 47, 63, 252, 33, 10, 204, 43, 253, 251, 213, 30, 152, 129, 48, 80,
+             38, 22, 18, 5, 130, 129, 113, 34, 92, 36, 20, 25, 9, 2, 193, 64, 144, 68, 36, 17, 75,
+             215, 198, 77, 184, 229, 170, 157, 115, 169, 223>> <> _rest = audio_chunk.payload
+  end
 end