Replies: 4 comments 21 replies
-
|
Hey @ndrean, thanks for creating the issue. Is it possible to send the complete ffmpeg command for testing? I'll take a look post work today. |
Beta Was this translation helpful? Give feedback.
-
|
Thanks @akash-akya for looking into this. FFmpeg: path = up_to_you
segment = Path.join(path, "segment_%03d.ts")
playlist = Path.join(path, "stream.m3u8")
ffmpeg = System.find_executable("ffmpeg")
[
# Input from stdin (pipe)
["-i", "pipe:0"],
# sets the input frame rate to 20 frames per second.
# Adjusted to the value used in the browser:"canvas.captureStream(20)"
["-r", "20"],
# Video codec to use (libx264)
["-c:v", "libx264"],
# Duration of each segment in seconds
["-hls_time", "2"],
# Number of segments to keep in the playlist (rolling playlist)
["-hls_list_size", "5"],
# Option to delete old segments
["-hls_flags", "delete_segments"],
# Type of playlist (live for continuous update)
["-hls_playlist_type", "event"],
# Segment file naming pattern
["-hls_segment_filename", segment],
# Playlist file
playlist
]
|> List.flatten() |
Beta Was this translation helpful? Give feedback.
-
|
@akash-akya :ok =
ExCmd.stream!(
~w(#{state.ffmpeg} -i pipe:0 -frames:v 1 -f image2 -vcodec mjpeg -y priv/static/images/test_%02d.jpg),
input: File.stream!(path, [], 65_336)
)
|> Stream.run()Result is :ok ExCmd.stream!(
~w(#{state.ffmpeg} -i pipe:0 -frames:v 1 -f image2 -vcodec mjpeg -y priv/static/images/test_%02d.jpg),
input: File.stream!(frame, [], 65336)
) #=> %ExCmd.Stream{
process: #PID<0.4555.0>,
stream_opts: %{
input: {:enumerable,
%File.Stream{
path: "/var/folders/mz/91hbds1j23125yksdf67dcgm0000gn/T/plug-1719-tBhO/multipart-1719241495-896826904435-1",
modes: [:raw, :read_ahead, :binary],
line_or_bytes: 65336,
raw: true,
node: :nonode@nohost
}},
exit_timeout: :infinity
}
}
|> Stream.run() #=> :okThis works, almost (modulo you exclude "priv/static/images" from being watched by the hot_reload"), but the output file is continuously overwritten. I believe this is a single task, meaning that the FFmpeg is not kept alive, thus it does not auto-increment w/r to the pattern %02d. Is it possible for ExCmd let FFmpeg run as a controlled process, like Porcelain or erlexec does? As a side note, dialyzer gives a warning even if the command works. The function call will not succeed.
File.stream!(_frame :: any(), [], 65336)
breaks the contract
(Path.t(), :line | pos_integer(), [stream_mode()]) :: File.Stream.t() |
Beta Was this translation helpful? Give feedback.
-
|
That's odd. The above command worked for me. What OS are you trying this on? Linux? I hope the ffmpeg command is similar. |
Beta Was this translation helpful? Give feedback.
-
|
Just a wild guess. Can you ensure you are not using |
Beta Was this translation helpful? Give feedback.
-
|
Oh I see the issue. The Dialyzer is actually warning about that |
Beta Was this translation helpful? Give feedback.
-
|
The File.stream! args are incorrect in your code. Yes, it fails with I'm on OSX and the command works when run with Porcelain. At least, a good point. |
Beta Was this translation helpful? Give feedback.
-
|
Looks like the function signature is changed recently 😅 |
Beta Was this translation helpful? Give feedback.
-
|
Hey @ndrean, I was checking this and since in this case, we get chunks from user as separate {:ok, _} = Supervisor.start_link(
[{Task.Supervisor, name: FfmpegTest.TaskSupervisor}],
strategy: :one_for_one
)
defmodule FfmpegTest do
require Logger
@spec start_ffmpeg(String.t()) :: pid()
def start_ffmpeg(base_path) do
{:ok, ffmpeg_proc} =
Task.Supervisor.start_child(FfmpegTest.TaskSupervisor, fn ->
segment_pattern = Path.join(base_path, "temp_segment_%03d.ts")
playlist_path = Path.join(base_path, "temp_stream.m3u8")
cmd = build_ffmpeg_cmd(segment_pattern, playlist_path)
{:ok, ffmpeg_proc} = ExCmd.Process.start_link(cmd)
Logger.info("ffmpeg started. Start sending chunks")
received_to_stdin!(ffmpeg_proc)
Logger.info("All chunks are forwarded, waiting for the ffmpeg to finish")
await_exit!(ffmpeg_proc)
Logger.info("Ffmpeg finished successfully\n#{File.read!(playlist_path)}")
end)
ffmpeg_proc
end
defp received_to_stdin!(ffmpeg_proc) do
receive do
{:chunk, data} ->
:ok = ExCmd.Process.write(ffmpeg_proc, data)
received_to_stdin!(ffmpeg_proc)
:done ->
# indicate that we are done sending all chunks
:ok = ExCmd.Process.close_stdin(ffmpeg_proc)
end
end
# consume everything written to stdout (if there is any)
# and wait for cmd to exit
defp await_exit!(ffmpeg_proc) do
case ExCmd.Process.read(ffmpeg_proc) do
{:ok, data} ->
Logger.info("[stdout] #{inspect(data)}")
await_exit!(ffmpeg_proc)
:eof ->
{:ok, 0} = ExCmd.Process.await_exit(ffmpeg_proc)
end
end
defp build_ffmpeg_cmd(segment_pattern, playlist_path) do
[
"ffmpeg",
# Input from stdin (pipe)
["-i", "pipe:0"],
# sets the input frame rate to 20 frames per second.
# Adjusted to the value used in the browser:"canvas.captureStream(20)"
["-r", "20"],
# Video codec to use (libx264)
["-c:v", "libx264"],
# Duration of each segment in seconds
["-hls_time", "2"],
# Number of segments to keep in the playlist (rolling playlist)
["-hls_list_size", "5"],
# Option to delete old segments
["-hls_flags", "delete_segments"],
# Type of playlist (live for continuous update)
["-hls_playlist_type", "event"],
# Segment file naming pattern
["-hls_segment_filename", segment_pattern],
# Playlist file
playlist_path
]
|> List.flatten()
end
end
##################
ffmpeg_proc = FfmpegTest.start_ffmpeg("/Users/akash/repo/elixir/ex_cmd/temp")
# forward each chunk we receive in the request to ffmpeg
send(ffmpeg_proc, {:chunk, File.read!("/Users/akash/repo/elixir/ex_cmd/segment_000.ts")})
send(ffmpeg_proc, {:chunk, File.read!("/Users/akash/repo/elixir/ex_cmd/segment_001.ts")})
# say we are done when there are no more data
send(ffmpeg_proc, :done)Outputwdyt? |
Beta Was this translation helpful? Give feedback.
-
|
@ndrean can you try this? this is working for me locally Application.put_env(:sample, Example.Endpoint,
http: [ip: {127, 0, 0, 1}, port: 5001],
server: true,
live_view: [signing_salt: "aaaaaaaa"],
secret_key_base: String.duplicate("a", 64)
)
Mix.install([
{:plug_cowboy, "~> 2.5"},
{:plug, "~>1.6"},
{:jason, "~> 1.0"},
{:phoenix, "~> 1.7.0"},
{:phoenix_live_view, "~> 0.20.0"},
{:ex_cmd, "~> 0.12"}
])
defmodule Example.ErrorView do
def render(template, _), do: Phoenix.Controller.status_message_from_template(template)
end
defmodule Example.HomeLive do
use Phoenix.LiveView, layout: {__MODULE__, :live}
def mount(_params, _session, socket) do
{:ok, socket}
end
defp phx_vsn, do: Application.spec(:phoenix, :vsn)
defp lv_vsn, do: Application.spec(:phoenix_live_view, :vsn)
def render("live.html", assigns) do
~H"""
<script src={"https://cdn.jsdelivr.net/npm/phoenix@#{phx_vsn()}/priv/static/phoenix.min.js"}>
</script>
<script
src={"https://cdn.jsdelivr.net/npm/phoenix_live_view@#{lv_vsn()}/priv/static/phoenix_live_view.min.js"}
>
</script>
<script>
let liveSocket = new window.LiveView.LiveSocket("/live", window.Phoenix.Socket)
liveSocket.connect()
function run() {
navigator.mediaDevices.getUserMedia({video: true})
.then((stream)=> {
video.srcObject = stream
let mediaRecorder = new MediaRecorder(stream);
mediaRecorder.ondataavailable = ({data}) => {
if (data.size > 0) {
console.log(data.size)
const file = new File([data], "chunk.webm", {
type: "video/webm",
});
const formData = new FormData();
formData.append("file", file);
fetch(`/upload`, {method: "POST",body: formData})
.then((res) => res.text())
.then(console.log)
}
}
mediaRecorder.start(1000)
})
}
run()
</script>
<%= @inner_content %>
"""
end
def render(assigns) do
~H"""
<video id="video" width="500" height="500" autoplay></video>
"""
end
end
defmodule Example.Router do
use Phoenix.Router
import Phoenix.LiveView.Router
pipeline :browser do
plug(:accepts, ["html"])
plug(Plug.Parsers, parsers: [:urlencoded, :multipart])
end
scope "/", Example do
pipe_through(:browser)
live("/", HomeLive, :index)
post("/upload", Post, :upload)
end
end
defmodule FFmpeger do
require Logger
use GenServer
def start_link(_) do
GenServer.start_link(__MODULE__, nil, name: __MODULE__)
end
def enqueue(path) do
Logger.info("Queue #{path}")
GenServer.cast(__MODULE__, {:process, path})
end
defp playlist_path, do: Path.join(System.tmp_dir(), "stream.m3u8")
defp segment_path, do: Path.join(System.tmp_dir(), "segment_%03d.ts")
def init(_) do
cmd =
~w(ffmpeg -hide_banner -loglevel warning -i pipe:0 -r 20 -c:v libx264 -hls_time 2 -hls_list_size 5 -hls_flags delete_segments+append_list -hls_playlist_type event -hls_segment_filename #{segment_path()} #{playlist_path()})
Logger.info("Playlist path: #{playlist_path()}")
ExCmd.Process.start_link(cmd, log: true)
end
def handle_cast({:process, path}, pid) do
:ok = ExCmd.Process.write(pid, File.read!(path))
Logger.info("Wrote #{path}")
{:noreply, pid}
end
end
defmodule Example.Post do
use Phoenix.Controller
def upload(conn, %{"file" => %Plug.Upload{path: path} = file}) do
# NOTE: the `file` will be kept alive in the temp directory till `FFmpeger` terminates
:ok = Plug.Upload.give_away(file, GenServer.whereis(FFmpeger), self())
:ok = FFmpeger.enqueue(path)
conn |> put_status(201) |> json(%{status: "ok"})
end
end
defmodule Example.Endpoint do
use Phoenix.Endpoint, otp_app: :sample
socket("/live", Phoenix.LiveView.Socket)
plug(Example.Router)
end
{:ok, _} = Supervisor.start_link([Example.Endpoint, FFmpeger], strategy: :one_for_one)
Process.sleep(:infinity) |
Beta Was this translation helpful? Give feedback.
-
Thats correct. It captures it by default, |
Beta Was this translation helpful? Give feedback.
-
|
@akash-akya, of course..., Ok, perfect. Thanks for your time. Correct me if I am wrong. I get the output back into Elixir if any when I express the stream with To kill FFmpeg, you just ExCmd.Process.stop(pid) |
Beta Was this translation helpful? Give feedback.
-
Yes. But only whatever is written to stdout (not |
Beta Was this translation helpful? Give feedback.
-
Not actually needed in this case, it is taken care for you when the Ffmpeg genserver terminates :) |
Beta Was this translation helpful? Give feedback.
-
|
I think this could be another example for your documentation. I just added a stop button to test how FFmpeg is killed. Application.put_env(:sample, Example.Endpoint,
http: [ip: {127, 0, 0, 1}, port: 5001],
server: true,
live_view: [signing_salt: "aaaaaaaa"],
secret_key_base: String.duplicate("a", 64)
)
Mix.install([
{:plug_cowboy, "~> 2.7"},
{:jason, "~> 1.4"},
{:phoenix, "~> 1.7.0"},
{:phoenix_live_view, "~> 0.20.0"},
{:ex_cmd, "~> 0.12"},
])
defmodule Example.ErrorView do
def render(template, _), do: Phoenix.Controller.status_message_from_template(template)
end
defmodule Example.HomeLive do
use Phoenix.LiveView, layout: {__MODULE__, :live}
def mount(_params, _session, socket) do
{:ok, assign(socket, :count, 0)}
end
defp phx_vsn, do: Application.spec(:phoenix, :vsn)
defp lv_vsn, do: Application.spec(:phoenix_live_view, :vsn)
def render("live.html", assigns) do
~H"""
<script src={"https://cdn.jsdelivr.net/npm/phoenix@#{phx_vsn()}/priv/static/phoenix.min.js"}>
</script>
<script
src={"https://cdn.jsdelivr.net/npm/phoenix_live_view@#{lv_vsn()}/priv/static/phoenix_live_view.min.js"}
>
</script>
<script>
let liveSocket = new window.LiveView.LiveSocket("/live", window.Phoenix.Socket)
liveSocket.connect()
function run() {
navigator.mediaDevices.getUserMedia({video: true})
.then((stream)=> {
let video = document.getElementById("hls");
let send = true;
document.getElementById("stop").onclick = () => {
send = false;
}
video.srcObject = stream
let mediaRecorder = new MediaRecorder(stream);
mediaRecorder.ondataavailable = ({data}) => {
if (!send) return;
if (data.size > 0) {
console.log(data.size)
const file = new File([data], "chunk.webm", {
type: "video/webm",
});
const formData = new FormData();
formData.append("file", file);
fetch(`/upload`, {method: "POST",body: formData})
.then((res) => res.text())
.then(console.log)
}
}
mediaRecorder.start(1000)
})
}
run()
</script>
<%= @inner_content %>
"""
end
def render(assigns) do
~H"""
<video id="video" width="300" height="300" controls autoplay></video>
<button type="button" id="stop" phx-click="stop">STOP</button>
"""
end
def handle_event("stop", _, socket) do
FFmpeger.stop()
{:noreply, socket}
end
end
defmodule Example.Router do
use Phoenix.Router
import Phoenix.LiveView.Router
pipeline :browser do
plug(:accepts, ["html"])
plug Plug.Parsers, parsers: [:urlencoded, :multipart]
end
scope "/", Example do
pipe_through(:browser)
live("/", HomeLive, :index)
post("/upload", Post, :upload)
end
end
defmodule FFmpeger do
use GenServer
def start_link(_) do
GenServer.start_link(__MODULE__, {}, name: __MODULE__)
end
def enqueue(path) do
GenServer.call(__MODULE__, {:process, path})
end
def stop(), do: GenServer.call(__MODULE__, :stop)
def init(_) do
dir = System.tmp_dir()
playlist_path = Path.join(dir, "stream.m3u8")
segment_path = Path.join(dir, "segment_%03d.ts")
ffmpeg = System.find_executable("ffmpeg")
cmd = ~w(#{ffmpeg} -loglevel debug -hide_banner -i pipe:0 -r 20 -c:v libx264 -hls_time 2 -hls_list_size 5 -hls_flags delete_segments+append_list -hls_playlist_type event -hls_segment_filename #{segment_path} #{playlist_path})
{:ok, _pid} = ExCmd.Process.start_link(cmd)
end
def handle_call({:process, path}, _from, pid) do
data = File.read!(path)
ExCmd.Process.write(pid, data)
IO.puts "processed-----------------#{byte_size(data)}"
{:reply, :processed, pid}
end
def handle_call(:stop, _from, pid) do
IO.puts "stopping-----------------"
# tell ffmpeg that we are done sending chunks
:ok = ExCmd.Process.close_stdin(pid)
# Ideally this should not be needed, I'll fix soon in the master
:eof = ExCmd.Process.read(pid)
# wait for for ffmpeg to terminate, ensure it is terminated successfully
{:ok, 0} = ExCmd.Process.await_exit(pid)
{:stop, :shutdown, pid}
end
end
defmodule Example.Post do
use Phoenix.Controller
def upload(conn,%{"file" => %Plug.Upload{path: path}} ) do
:processed = FFmpeger.enqueue(path)
conn |> put_status(201) |> json(%{status: "ok"})
end
end
defmodule Example.Endpoint do
use Phoenix.Endpoint, otp_app: :sample
socket("/live", Phoenix.LiveView.Socket)
plug(Example.Router)
end
{:ok, _} = Supervisor.start_link([Example.Endpoint, FFmpeger], strategy: :one_for_one)
Process.sleep(:infinity) |
Beta Was this translation helpful? Give feedback.
-
|
@ndrean thanks for sharing. Minor note, a bit more cleaner way to terminate the external program is to close stdin first and wait for the program to terminate successfully. def handle_call(:stop, _from, pid) do
IO.puts "Stop-----------------"
# tell ffmpeg that we are done sending chunks
:ok = ExCmd.Process.close_stdin(pid)
# Ideally this should not be needed, I'll fix soon in the master
:eof = ExCmd.Process.read(pid)
# wait for for ffmpeg to terminate, ensure it is terminated successfully
{:ok, 0} = ExCmd.Process.await_exit(pid)
{:stop, :shutdown, pid}
end |
Beta Was this translation helpful? Give feedback.
-
|
@akash-akya Thanks again, a nice tool with the stream and the process. I can submit the example if you want, or let you close or not this discussion. Might be helpful to others, modulo the iterations :) |
Beta Was this translation helpful? Give feedback.
-
|
Thanks @ndrean. I'll try to create livebook example based on your code this weekend. With credits to you of course :) |
Beta Was this translation helpful? Give feedback.
-
|
@akash-akya The credits are all yours. I made a (draft) Livebook to may be inspire you and save you some time :) https://github.com/ndrean/RTC-HLS/blob/main/demo/ffmpeg-buffer-excmd-process.livemd
Working with ExCmd.ProcessMix.install([
{:ex_cmd, "~> 0.12.0"},
{:plug_cowboy, "~> 2.7"},
{:corsica, "~> 2.1"},
{:kino, "~> 0.13.1"}
])FFmpeg accepting buffer input with ExCmd.ProcessContextWe want to Live stram ourselves from the webcam using HTTP Live Streaming. Since You can use FFmpeg GenServerThe module below is a GenServer. It starts an The GenServer starts the
defmodule FFmpeger do
use GenServer
def start_link(_) do
GenServer.start_link(__MODULE__, {}, name: __MODULE__)
end
def enqueue(path) do
GenServer.call(__MODULE__, {:process, path})
end
def pid, do: GenServer.call(__MODULE__, :pid)
def stop(), do: GenServer.call(__MODULE__, :stop)
def init(_) do
dir = System.tmp_dir()
playlist_path = Path.join(dir, "stream.m3u8")
segment_path = Path.join(dir, "segment_%03d.ts")
ffmpeg = System.find_executable("ffmpeg") || "/opt/homebrew/bin/ffmpeg"
cmd =
~w(#{ffmpeg} -loglevel debug -hide_banner -i pipe:0 -r 20 -c:v libx264 -hls_time 2 -hls_list_size 5 -hls_flags delete_segments+append_list -hls_playlist_type event -hls_segment_filename #{segment_path} #{playlist_path})
ExCmd.Process.start_link(cmd)
end
def handle_call(:pid,_, pid), do: {:reply, self(), pid}
def handle_call({:process, path}, _from, pid) do
data = File.read!(path)
ExCmd.Process.write(pid, data)
IO.puts("processed-----------------#{byte_size(data)}")
{:reply, :processed, pid}
end
def handle_call(:stop, _from, pid) do
IO.puts("stopping-----------------")
:ok = ExCmd.Process.close_stdin(pid)
:eof = ExCmd.Process.read(pid)
{:ok, 0} = ExCmd.Process.await_exit(pid)
{:stop, :shutdown, pid}
end
endWebserver to handle POST requestsWe will run a (Cowboy) WebServer to listen on port 4002 to handle the data sent by the video stream. We run a multipart HTTP POST request every second to send the binary data packaged into a file to the endpoint http://localhost:4002/upload. defmodule PostRouter do
use Plug.Router
plug Corsica, origins: "*", allow_methods: ["GET", "POST"]
plug(Plug.Parsers,
parsers: [:urlencoded, :multipart],
pass: ["*/*"]
)
plug(:match)
plug(:dispatch)
get "/" do
IO.puts("received")
send_resp(conn, 200, "hello from Cowboy")
end
post "/upload" do
%{"file" => %Plug.Upload{path: path}} = conn.params
:processed = FFmpeger.enqueue(path)
send_resp(conn, 201, "uploaded")
end
endStart the weberser and the FFmpeg runner module: webserver = {Plug.Cowboy, plug: PostRouter, scheme: :http, port: 4002}
Supervisor.start_link([webserver, FFmpeger], strategy: :one_for_one, name: MySup)
Supervisor.which_children(MySup)We test that the Cowboy webserver is serving port 4002 by running a ExCmd.stream!(~w(curl http://localhost:4002)) |> Enum.into("")Launch the video streamThis module runs a You can watch the logs below where the size of the data is displayed. You can stop the process with the button "stop". defmodule VideoLive do
use Kino.JS
use Kino.JS.Live
@html """
<div id="elt">
<video id="video" width="400" height="400" controls muted playsinline autoplay></video>
<button type="button" id="stop" phx-click="stop">STOP</button>
</div>
"""
def new() do
Kino.JS.Live.new(__MODULE__, @html)
end
asset "main.css" do
"""
#elt {
display: flex;
flex-direction: column;
align-items: center
}
button {
margin-top: 1em;
}
"""
end
asset "main.js" do
"""
export function init(ctx, html) {
ctx.importCSS("main.css");
ctx.root.innerHTML = html;
function run() {
navigator.mediaDevices.getUserMedia({video: {width: 400, height: 400}, audio: false})
.then((stream)=> {
let video = document.getElementById("video"),
send = true;
document.getElementById("stop").onclick = () => {
send = false;
ctx.pushEvent("stop", {})
}
video.srcObject = stream
let mediaRecorder = new MediaRecorder(stream);
mediaRecorder.ondataavailable = ({data}) => {
if (!send) return;
if (data.size > 0) {
console.log(data.size)
const file = new File([data], "chunk.webm", {
type: "video/webm",
});
const formData = new FormData();
formData.append("file", file);
fetch(`http://localhost:4002/upload`, {method: "POST",body: formData})
.then((res) => res.text())
.then(console.log)
}
}
mediaRecorder.start(1000)
})
}
run()
}
"""
end
@impl true
def init(html, ctx) do
{:ok, assign(ctx, html: html)}
end
@impl true
def handle_connect(ctx) do
{:ok, ctx.assigns.html, ctx}
end
@impl true
def handle_event("stop", _, ctx) do
FFmpeger.stop()
{:noreply, ctx}
end
endVideoLive.new() |
Beta Was this translation helpful? Give feedback.
-
|
Thanks a lot @ndrean! I appreciate your time :) |
Beta Was this translation helpful? Give feedback.
-
This is not an issue but I would like to open a discussion on how to use ExCmd.
My use case is:
%PlugUpload{path: path}.I currently
data = File.read!(path)and pass the data to a process. It is a GenServer that initsPorcelainwith a command that runs FFmpeg: indeed, I run the FFmpeg using the stdin buffer, so I need it to be running. FFmpeg is building HLS segments and a playlist.My interest in
ExCmdis that it seems I don't need a GenServer.I can
%Plug.Upload.give_away(FFmpegProcess, path, self())and just pass the path.How would you do then? The following does not run. I have no external file to output to since FFmpeg builds the segments and the playlist, passed as arguments in the command.
ExCmd.stream!(__ffmpeg_cmd__, input: File.read!(path))I tried to add
Enum.into(""), but I get{:error, :epipe}.I did not add a Livebook because I am not familiar with Kino, nor how would I run some Javascript code in a Livebook.
Beta Was this translation helpful? Give feedback.
All reactions