Handle TDEN frame in timed ID3v2 stream

lib/ex_hls/demuxing_engine/mpeg_ts.ex

@@ -3,12 +3,13 @@ defmodule ExHLS.DemuxingEngine.MPEGTS do
   @behaviour ExHLS.DemuxingEngine
 
   use Bunch.Access
+  use Bunch
 
   require Logger
   alias Membrane.{AAC, H264, RemoteStream}
   alias MPEG.TS.Demuxer
 
-  @enforce_keys [:demuxer]
+  @enforce_keys [:demuxer, :last_tden_tag]
   defstruct @enforce_keys ++ [track_timestamps_data: %{}]
 
   # using it a boundary expressed in nanoseconds, instead of the usual 90kHz clock ticks,
@@ -17,7 +18,8 @@ defmodule ExHLS.DemuxingEngine.MPEGTS do
   @timestamp_range_size_ns div(2 ** 33 * 1_000_000_000, 90_000)
 
   @type t :: %__MODULE__{
-          demuxer: Demuxer.t()
+          demuxer: Demuxer.t(),
+          last_tden_tag: String.t() | nil
         }
 
   @impl true
@@ -32,7 +34,7 @@ defmodule ExHLS.DemuxingEngine.MPEGTS do
     # TODO - figure out how to do it properly
     demuxer = %{demuxer | waiting_random_access_indicator: false}
 
-    %__MODULE__{demuxer: demuxer}
+    %__MODULE__{demuxer: demuxer, last_tden_tag: nil}
   end
 
   @impl true
@@ -79,8 +81,11 @@ defmodule ExHLS.DemuxingEngine.MPEGTS do
   @impl true
   def pop_chunk(%__MODULE__{} = demuxing_engine, track_id) do
     with {[packet], demuxer} <- Demuxer.take(demuxing_engine.demuxer, track_id) do
+      {maybe_tden_tag, demuxer} = maybe_read_tden_tag(demuxer, packet.pts)
+      tden_tag = maybe_tden_tag || demuxing_engine.last_tden_tag
+
       {demuxing_engine, packet} =
-        %{demuxing_engine | demuxer: demuxer}
+        %{demuxing_engine | demuxer: demuxer, last_tden_tag: tden_tag}
         |> handle_possible_timestamps_rollover(track_id, packet)
 
       chunk = %ExHLS.Chunk{
@@ -90,7 +95,8 @@ defmodule ExHLS.DemuxingEngine.MPEGTS do
         track_id: track_id,
         metadata: %{
           discontinuity: packet.discontinuity,
-          is_aligned: packet.is_aligned
+          is_aligned: packet.is_aligned,
+          tden_tag: tden_tag
         }
       }
 
@@ -101,6 +107,37 @@ defmodule ExHLS.DemuxingEngine.MPEGTS do
     end
   end
 
+  defp maybe_read_tden_tag(demuxer, packet_pts) do
+    withl no_id3_stream:
+            {id3_track_id, _stream_description} <-
+              demuxer.pmt.streams
+              |> Enum.find(fn {_pid, stream_description} ->
+                stream_description.stream_type == :METADATA_IN_PES
+              end),
+          no_id3_data: {[id3], demuxer} <- Demuxer.take(demuxer, id3_track_id),
+          id3_not_in_timerange: true <- id3.pts <= packet_pts do
+      {parse_tden_tag(id3.data), demuxer}
+    else
+      no_id3_stream: nil -> {nil, demuxer}
+      no_id3_data: {[], updated_demuxer} -> {nil, updated_demuxer}
+      id3_not_in_timerange: false -> {nil, demuxer}
+    end
+  end
+
+  defp parse_tden_tag(payload) do
+    # UTF-8 encoding
+    encoding = 3
+
+    with {pos, _len} <- :binary.match(payload, "TDEN"),
+         <<_skip::binary-size(pos), "TDEN", tden::binary>> <- payload,
+         <<size::integer-size(4)-unit(8), _flags::16, ^encoding::8, text::binary-size(size - 2),
+           0::8, _rest::binary>> <- tden do
+      text
+    else
+      _error -> nil
+    end
+  end
+
   # value returned by Demuxer is represented in nanoseconds
   defp packet_ts_to_millis(ts), do: div(ts, 1_000_000)
 

mix.exs

@@ -40,9 +40,12 @@ defmodule ExHLS.Mixfile do
       {:ex_m3u8, "~> 0.15.4"},
       {:req, "~> 0.5.10"},
       {:qex, "~> 0.5.1"},
+      {:bunch, "~> 1.6"},
       {:membrane_mp4_plugin, "~> 0.36.0"},
       {:membrane_h26x_plugin, "~> 0.10.2"},
-      {:mpeg_ts, "~> 2.0.0"},
+      {:mpeg_ts,
+       github: "membraneframework-labs/kim_mpeg_ts",
+       branch: "varsill/fix_pes_optional_header_resolving"},
       {: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

@@ -32,8 +32,8 @@
   "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", "c8c770e0e7714c72b3faa7f20088fdbd76f5bade", [branch: "varsill/fix_pes_optional_header_resolving"]},
   "mock": {:hex, :mock, "0.3.9", "10e44ad1f5962480c5c9b9fa779c6c63de9bd31997c8e04a853ec990a9d841af", [:mix], [{:meck, "~> 0.9.2", [hex: :meck, repo: "hexpm", optional: false]}], "hexpm", "9e1b244c4ca2551bb17bb8415eed89e40ee1308e0fbaed0a4fdfe3ec8a4adbd3"},
-  "mpeg_ts": {:hex, :mpeg_ts, "2.0.2", "87f7d3b38c962fc367edbb4b1419f5f314be41a0d512b95437f11c4f60c931f4", [:mix], [], "hexpm", "5b7f1245a945de647c29abc9453e3d9d7eca1b0001d3d582f4feb11fc09b2792"},
   "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

@@ -4,10 +4,11 @@ defmodule ExHLS.Client.Test do
   alias ExHLS.Client
   alias Membrane.{AAC, H264, RemoteStream}
 
-  @fixtures "https://raw.githubusercontent.com/membraneframework-labs/ex_hls/refs/heads/master/test/fixtures/"
+  @fixtures "https://raw.githubusercontent.com/membraneframework/ex_hls/refs/heads/master/test/fixtures/"
   @fmp4_url @fixtures <> "fmp4/output.m3u8"
   @fmp4_only_video_url @fixtures <> "fmp4_only_video/output.m3u8"
   @mpegts_only_video_url @fixtures <> "mpeg_ts_only_video/output_playlist.m3u8"
+  @mpegts_with_tden_url "test/fixtures/mpeg_ts_with_tden/output_playlist.m3u8"
   @mpegts_url "https://test-streams.mux.dev/x36xhzz/x36xhzz.m3u8"
   @mpegts_live_url "./test/fixtures/mpeg_ts_live/output_playlist.m3u8"
 
@@ -129,7 +130,35 @@ defmodule ExHLS.Client.Test do
              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
 
-    assert video_chunk.metadata == %{discontinuity: false, is_aligned: false}
+    assert video_chunk.metadata == %{discontinuity: false, is_aligned: false, tden_tag: nil}
+  end
+
+  test "(MPEGTS) stream with ID3v2.4 TDEN tag" do
+    client = Client.new(@mpegts_with_tden_url)
+
+    assert Client.get_variants(client) == %{}
+
+    chunks = Client.generate_stream(client) |> Enum.take(381)
+
+    first_audio_chunk_after_tden =
+      Enum.find(
+        chunks,
+        &(&1.metadata.tden_tag != nil and &1.media_type == :audio)
+      )
+
+    first_video_chunk_after_tden =
+      Enum.find(
+        chunks,
+        &(&1.metadata.tden_tag != nil and &1.media_type == :video)
+      )
+
+    assert first_audio_chunk_after_tden.pts_ms == 3328
+    assert first_audio_chunk_after_tden.dts_ms == 3328
+    assert first_audio_chunk_after_tden.metadata.tden_tag == "2025-10-21T08:07:50"
+
+    assert first_video_chunk_after_tden.pts_ms == 3233
+    assert first_video_chunk_after_tden.dts_ms == 3233
+    assert first_video_chunk_after_tden.metadata.tden_tag == "2025-10-21T08:07:50"
   end
 
   test "(fMP4) stream with only video" do
@@ -223,7 +252,7 @@ defmodule ExHLS.Client.Test do
              12, 2, 13, 110, 0, 0, 9, 154, 0, 1, 224, 0, 30, 44, 91, 44, 0, 0, 0, 1, 104, 234,
              225, 178, 200, 176, 0, 0>> <> _rest = video_chunk.payload
 
-    assert video_chunk.metadata == %{discontinuity: false, is_aligned: false}
+    assert video_chunk.metadata == %{discontinuity: false, is_aligned: false, tden_tag: nil}
   end
 
   defp assert_chunks_are_in_proper_order(chunks) do

test/demuxing_engine_mpegts_test.exs

@@ -26,7 +26,8 @@ defmodule ExHLS.DemuxingEngine.MPEGTS.Test do
 
     demuxer = %{
       waiting_random_access_indicator: nil,
-      packet_buffers: %{1 => packets, 2 => packets}
+      packet_buffers: %{1 => packets, 2 => packets},
+      pmt: %{streams: %{}}
     }
 
     new = fn -> demuxer end

test/fixtures/mpeg_ts_with_tden/output_playlist.m3u8

@@ -0,0 +1,18 @@
+#EXTM3U
+#EXT-X-VERSION:3
+#EXT-X-TARGETDURATION:2
+#EXT-X-MEDIA-SEQUENCE:0
+#EXT-X-PLAYLIST-TYPE:VOD
+#EXTINF:2.000000,
+output_playlist0.ts
+#EXTINF:2.000000,
+output_playlist1.ts
+#EXTINF:2.000000,
+output_playlist2.ts
+#EXTINF:2.000000,
+output_playlist3.ts
+#EXTINF:2.000000,
+output_playlist4.ts
+#EXTINF:0.033333,
+output_playlist5.ts
+#EXT-X-ENDLIST