SimplePipeline exploration

simple_pipeline.ex

defmodule Membrane.Demo.SimplePipeline do
  @moduledoc """
  Sample Membrane pipeline that will play an `.mp3` file.

  ## Example

      iex> Membrane.Demo.SimplePipeline.start_link(path_to_mp3: "demo.mp3")
      iex> Membrane.Pipeline.terminate(Membrane.Demo.SimplePipeline)
  """

  use Membrane.Pipeline
  alias Membrane.Clock
  alias Membrane.Pipeline
  alias Membrane.Time

  @interval 5000

  def start_link(opts) do
    Pipeline.start_link(__MODULE__, opts, name: Keyword.get(opts, :name, __MODULE__))
  end

  @doc """
  In order to play `.mp3` file we need to read it first.

  In membrane every entry point to data flow is called `Source`. Since we want
  to play a `file`, we will use `File.Source`.

  Next problem that arises is the fact that we are reading MPEG Layer 3 frames
  not raw audio. To deal with that we need to use `Filter` called decoder. It
  takes `.mp3` frames and yields RAW audio data.

  There is one tiny problem here though. Decoder returns `%Raw{format: :s24le}`
  data, but PortAudio (module that actually talks with the audio driver of your
  computer) wants `%Raw{format: :s16le, sample_rate: 48000, channels: 2}`.

  That's where `SWResample.Converter` comes into play. It will consume data that
  doesn't suite our needs and will yield data in format we want.
  """
  @impl true
  def handle_init(_context, opts) do
    path_to_mp3 = Keyword.get(opts, :path_to_mp3)
    IO.puts("filename: #{path_to_mp3}")

    children = [
      # Stream from file
      child(:file, %Membrane.File.Source{location: path_to_mp3}),
      # Decode frames
      child(:decoder, Membrane.MP3.MAD.Decoder),
      # Convert Raw :s24le to Raw :s16le
      child(:converter, %Membrane.FFmpeg.SWResample.Converter{
        output_stream_format: %Membrane.RawAudio{
          sample_format: :s16le,
          sample_rate: 48000,
          channels: 2
        }
      }),
      # Stream data into PortAudio to play it on speakers.
      child(:portaudio, Membrane.PortAudio.Sink)
    ]

    # Setup the flow of the data
    links = [
      get_child(:file) |> get_child(:decoder) |> get_child(:converter) |> get_child(:portaudio)
    ]

    structure = children ++ links

    state = %{
      last_tick: Time.monotonic_time()
    }

    {[spec: structure], state}
  end

  # def playback(name, mode) do
  #   Pipeline.call(name, {:playback, mode})
  # end

  # def handle_call({:playback, mode}, _context, state) do
  #   {[reply: :ok, playback: mode], state}
  # end

  def handle_call(message, _context, state) do
    IO.puts("handle_call:")
    IO.inspect(message, pretty: true)
    IO.puts("===============================")
    {[], state}
  end

  def handle_playing(context, state) do
    IO.puts("handle_playing, activating timer!")
    IO.puts("===============================")
    Clock.subscribe(context.clock, self())
    {[start_timer: {:tick, Membrane.Time.milliseconds(@interval)}], state}
  end

  def handle_child_notification(notification, element, _context, state) do
    IO.puts("handle_child_notification:")
    IO.inspect(notification, pretty: true)
    IO.inspect(element, pretty: true)
    IO.puts("===============================")
    {[], state}
  end

  def handle_element_start_of_stream(child, pad, _context, state) do
    IO.puts("handle_element_start_of_stream:")
    IO.inspect(child, pretty: true)
    IO.inspect(pad, pretty: true)
    IO.puts("===============================")
    {[], state}
  end

  def handle_element_end_of_stream(child, pad, _context, state) do
    IO.puts("handle_element_end_of_stream:")
    IO.inspect(child, pretty: true)
    IO.inspect(pad, pretty: true)
    IO.puts("===============================")
    {[], state}
  end

  @impl Membrane.Pipeline
  def handle_tick(tick, _context, %{last_tick: last_tick} = state) do
    now = Time.monotonic_time()
    IO.puts("handle_tick:")
    IO.inspect(tick, pretty: true)
    IO.puts("last_tick: #{last_tick}")
    # ns to ms
    measured = (now - last_tick) / 1_000_000
    IO.puts("interval: #{measured} (off #{measured / @interval})")
    IO.puts("===============================")

    state = %{state | last_tick: now}
    {[], state}
  end

  # def handle_info({:membrane_clock_ratio, _from, ratio}, _context, state) do
  #   IO.puts("ratio: #{ratio}")
  #   IO.puts("===============================")
  #   {[], state}
  # end

  def handle_info(message, _context, state) do
    IO.puts("handle_info:")
    IO.inspect(message, pretty: true)
    IO.puts("===============================")
    {[], state}
  end

  # def handle{:membrane_clock_ratio, clock :: pid(), ratio()}
end