Merge pull request #8 from jellyfish-dev/bump_compositor

Update compositor deps

Author: Karolk99

.circleci/config.yml

@@ -5,7 +5,7 @@ orbs:
 executors:
   docker-executor:
     docker:
-      - image: ghcr.io/jellyfish-dev/recording-converter-test:0.0.2
+      - image: ghcr.io/jellyfish-dev/recording-converter-test:0.0.3
         environment:
           MIX_ENV: test
 jobs:

lib/compositor_utils.ex

@@ -1,6 +1,8 @@
 defmodule RecordingConverter.Compositor do
   @moduledoc false
 
+  alias Membrane.LiveCompositor.Request
+
   @text_margin 10
   @letter_width 12
   @output_width 1280
@@ -37,60 +39,40 @@ defmodule RecordingConverter.Compositor do
   @spec video_output_id() :: String.t()
   def video_output_id(), do: @video_output_id
 
-  @spec generate_output_update(map(), number()) :: [tuple()]
+  @spec generate_output_update(map(), number()) :: [struct()]
   def generate_output_update(tracks, timestamp),
     do: [
       generate_video_output_update(tracks, timestamp),
       generate_audio_output_update(tracks, timestamp)
     ]
 
-  @spec schedule_unregister_video_output(number()) :: {:lc_request, map()}
+  @spec schedule_unregister_video_output(number()) :: Request.UnregisterOutput.t() | struct()
   def schedule_unregister_video_output(schedule_time_ns),
-    do: {
-      :lc_request,
-      %{
-        type: :unregister,
-        entity_type: :output_stream,
-        output_id: @video_output_id,
-        schedule_time_ms: from_ns_to_ms(schedule_time_ns)
-      }
+    do: %Request.UnregisterOutput{
+      output_id: @video_output_id,
+      schedule_time: Membrane.Time.nanoseconds(schedule_time_ns)
     }
 
-  @spec schedule_unregister_audio_output(number()) :: {:lc_request, map()}
+  @spec schedule_unregister_audio_output(number()) :: Request.UnregisterOutput.t() | struct()
   def schedule_unregister_audio_output(schedule_time_ns),
-    do: {
-      :lc_request,
-      %{
-        type: :unregister,
-        entity_type: :output_stream,
-        output_id: @audio_output_id,
-        schedule_time_ms: from_ns_to_ms(schedule_time_ns)
-      }
+    do: %Request.UnregisterOutput{
+      output_id: @audio_output_id,
+      schedule_time: Membrane.Time.nanoseconds(schedule_time_ns)
     }
 
-  @spec schedule_unregister_input(number(), binary()) :: {:lc_request, map()}
+  @spec schedule_unregister_input(number(), binary()) :: Request.UnregisterInput.t() | struct()
   def schedule_unregister_input(schedule_time_ns, input_id),
-    do: {
-      :lc_request,
-      %{
-        type: :unregister,
-        entity_type: :input_stream,
-        input_id: input_id,
-        schedule_time_ms: from_ns_to_ms(schedule_time_ns)
-      }
+    do: %Request.UnregisterInput{
+      input_id: input_id,
+      schedule_time: Membrane.Time.nanoseconds(schedule_time_ns)
     }
 
-  @spec register_image_action(String.t()) :: {:lc_request, map()}
+  @spec register_image_action(String.t()) :: Request.RegisterImage.t() | struct()
   def register_image_action(image_url) do
-    {
-      :lc_request,
-      %{
-        type: "register",
-        entity_type: "image",
-        asset_type: "png",
-        image_id: "avatar_png",
-        url: image_url
-      }
+    %Request.RegisterImage{
+      asset_type: "png",
+      image_id: "avatar_png",
+      url: image_url
     }
   end
 
@@ -105,29 +87,19 @@ defmodule RecordingConverter.Compositor do
     avatars_config = Enum.map(avatar_tracks, &avatar_view/1)
     video_tracks_config = Enum.map(video_tracks, &video_input_source_view/1)
 
-    {
-      :lc_request,
-      %{
-        type: :update_output,
-        output_id: @video_output_id,
-        schedule_time_ms: from_ns_to_ms(timestamp),
-        video: scene(video_tracks_config ++ avatars_config)
-      }
+    %Request.UpdateVideoOutput{
+      output_id: @video_output_id,
+      schedule_time: Membrane.Time.nanoseconds(timestamp),
+      root: scene(video_tracks_config ++ avatars_config)
     }
   end
 
   defp generate_audio_output_update(%{"audio" => audio_tracks}, timestamp)
        when is_list(audio_tracks) do
-    {
-      :lc_request,
-      %{
-        type: :update_output,
-        output_id: @audio_output_id,
-        audio: %{
-          inputs: Enum.map(audio_tracks, &%{input_id: &1.id})
-        },
-        schedule_time_ms: from_ns_to_ms(timestamp)
-      }
+    %Request.UpdateAudioOutput{
+      output_id: @audio_output_id,
+      inputs: Enum.map(audio_tracks, &%{input_id: &1.id}),
+      schedule_time: Membrane.Time.nanoseconds(timestamp)
     }
   end
 
@@ -139,7 +111,7 @@ defmodule RecordingConverter.Compositor do
           # TODO: fix after compositor update
           # unnecessary rescaler
           %{
-            type: "rescaler",
+            type: :rescaler,
             mode: "fit",
             child: %{
               type: :input_stream,
@@ -152,16 +124,16 @@ defmodule RecordingConverter.Compositor do
 
   defp avatar_view(track) do
     %{
-      type: "view",
+      type: :view,
       children:
         [
           # TODO: fix after compositor update
           # unnecessary rescaler
           %{
-            type: "rescaler",
+            type: :rescaler,
             mode: "fit",
             child: %{
-              type: "image",
+              type: :image,
               image_id: "avatar_png"
             }
           }
@@ -174,31 +146,26 @@ defmodule RecordingConverter.Compositor do
 
     [
       %{
-        type: "view",
+        type: :view,
         bottom: 20,
         right: 20,
         width: label_width,
         height: 20,
         background_color_rgba: "#000000FF",
         children: [
+          %{type: :view},
           %{
-            type: "text",
+            type: :text,
             text: label,
             align: "center",
             width: label_width,
             font_size: 20.0
-          }
+          },
+          %{type: :view}
         ]
       }
     ]
   end
 
   defp text_view(_metadata), do: []
-
-  defp from_ns_to_ms(timestamp_ns) do
-    rounded_ts =
-      timestamp_ns |> Membrane.Time.nanoseconds() |> Membrane.Time.as_milliseconds(:round)
-
-    max(0, rounded_ts - 10)
-  end
 end

lib/pipeline.ex

@@ -6,7 +6,7 @@ defmodule RecordingConverter.Pipeline do
 
   alias Membrane.AWS.S3.Source
   alias Membrane.HTTPAdaptiveStream.{SinkBin, Storages}
-  alias Membrane.Time
+  alias Membrane.LiveCompositor
   alias RecordingConverter.{Compositor, ReportParser}
 
   @segment_duration 3
@@ -17,7 +17,12 @@ defmodule RecordingConverter.Pipeline do
 
   @impl true
   def handle_init(_ctx, opts) do
-    output_directory = opts.output_directory
+    {[], opts}
+  end
+
+  @impl true
+  def handle_setup(_ctx, state) do
+    output_directory = state.output_directory
 
     with {:ok, files} when files != [] <- File.ls(output_directory) do
       Logger.warning(
@@ -29,19 +34,6 @@ defmodule RecordingConverter.Pipeline do
 
     File.mkdir_p!(output_directory)
 
-    main_spec =
-      [
-        generate_sink_bin(output_directory),
-        generate_output_audio_branch(opts),
-        generate_output_video_branch(opts)
-      ]
-      |> Enum.reject(&is_nil(&1))
-
-    {[spec: main_spec], opts}
-  end
-
-  @impl true
-  def handle_setup(_ctx, state) do
     report_path = s3_file_path(@report_file, state)
 
     tracks = ReportParser.get_tracks(state.bucket_name, report_path)
@@ -52,15 +44,21 @@ defmodule RecordingConverter.Pipeline do
 
     tracks_spec = Enum.map(tracks, &create_branch(&1, state))
 
-    track_actions =
-      tracks
-      |> ReportParser.get_all_track_actions()
-      |> Enum.map(&notify_compositor/1)
+    track_actions = ReportParser.get_all_track_actions(tracks)
 
-    register_image_action =
-      state.image_url |> Compositor.register_image_action() |> notify_compositor()
+    register_image_action = Compositor.register_image_action(state.image_url)
 
-    actions = [{:spec, tracks_spec}, register_image_action | track_actions]
+    all_compositor_actions = [register_image_action | track_actions]
+
+    main_spec =
+      [
+        generate_sink_bin(output_directory),
+        generate_output_audio_branch(state, all_compositor_actions),
+        generate_output_video_branch(state)
+      ]
+      |> Enum.reject(&is_nil(&1))
+
+    actions = [{:spec, main_spec ++ tracks_spec}]
 
     {actions,
      %{
@@ -90,24 +88,39 @@ defmodule RecordingConverter.Pipeline do
 
   @impl true
   def handle_child_notification(
-        {:lc_request_response, req, %Req.Response{status: response_code, body: response_body},
-         _lc_ctx},
-        _child,
+        {:request_result, _req, {:error, error}},
+        :video_compositor,
         _membrane_ctx,
         state
       ) do
-    if response_code != 200 do
-      raise """
-      Request failed.
-      Request: `#{inspect(req)}.
-      Response code: #{response_code}.
-      Response body: #{inspect(response_body)}.
-      """
-    end
+    raise """
+      Request failed with error: #{inspect(error)}
+    """
 
     {[], state}
   end
 
+  @impl true
+  def handle_child_notification(
+        {atom, _pad, _lc_ctx},
+        :video_compositor,
+        _membrane_ctx,
+        state
+      )
+      when atom in [:input_delivered, :input_playing, :input_eos] do
+    {[], state}
+  end
+
+  @impl true
+  def handle_child_notification(:start_of_stream, :hls_sink_bin, _ctx, state) do
+    {[], state}
+  end
+
+  @impl true
+  def handle_child_notification({:track_playable, _track}, :hls_sink_bin, _ctx, state) do
+    {[], state}
+  end
+
   @impl true
   def handle_child_notification(:end_of_stream, :hls_sink_bin, _ctx, state) do
     {[terminate: :normal], state}
@@ -143,14 +156,12 @@ defmodule RecordingConverter.Pipeline do
     get_child(:video_compositor)
     |> via_out(Pad.ref(:video_output, Compositor.video_output_id()),
       options: [
-        encoder_preset: :slow,
         width: @output_width,
         height: @output_height,
-        initial:
-          Compositor.scene([
-            %{type: :input_stream, input_id: "video_input_0", id: "child_0"}
-          ]),
-        send_eos_when: :all_inputs
+        encoder: %LiveCompositor.Encoder.FFmpegH264{preset: :slow},
+        initial: %{
+          root: Compositor.scene([])
+        }
       ]
     )
     |> child(:output_video_parser, %Membrane.H264.Parser{
@@ -160,27 +171,27 @@ defmodule RecordingConverter.Pipeline do
     |> via_in(Pad.ref(:input, :video),
       options: [
         encoding: :H264,
-        segment_duration: Time.seconds(@segment_duration)
+        segment_duration: Membrane.Time.seconds(@segment_duration)
       ]
     )
     |> get_child(:hls_sink_bin)
   end
 
-  defp generate_output_audio_branch(state) do
+  defp generate_output_audio_branch(state, compositor_actions) do
     child(:video_compositor, %Membrane.LiveCompositor{
       framerate: {30, 1},
       composing_strategy: :offline_processing,
-      server_setup: Compositor.server_setup(state.compositor_path)
+      server_setup: Compositor.server_setup(state.compositor_path),
+      init_requests: compositor_actions
     })
     |> via_out(Pad.ref(:audio_output, Compositor.audio_output_id()),
       options: [
-        channels: :stereo,
-        initial: %{
-          inputs: [
-            %{input_id: "audio_input_0", volume: 0.2}
-          ]
+        encoder: %LiveCompositor.Encoder.Opus{
+          channels: :stereo
         },
-        send_eos_when: :all_inputs
+        initial: %{
+          inputs: []
+        }
       ]
     )
     |> child(:opus_output_parser, Membrane.Opus.Parser)
@@ -190,7 +201,7 @@ defmodule RecordingConverter.Pipeline do
     |> via_in(Pad.ref(:input, :audio),
       options: [
         encoding: :AAC,
-        segment_duration: Time.seconds(@segment_duration)
+        segment_duration: Membrane.Time.seconds(@segment_duration)
       ]
     )
     |> get_child(:hls_sink_bin)