write ffmpeg command in pyav

Author: David Eigen

clarifai/client/model.py

@@ -1,7 +1,5 @@
 import json
-import logging
 import os
-import subprocess
 import time
 from typing import Any, Dict, Generator, Iterator, List, Tuple, Union
 
@@ -1207,15 +1205,13 @@ def stream_by_video_file(self,
     # TODO this conversion can offset the start time by a little bit; we should account for this
     # by getting the original start time ffprobe and either sending that to the model so it can adjust
     # with the ts of the first frame (too fragile to do all of this adjustment in the client input stream)
-    command = 'ffmpeg -i FILEPATH -c copy -f mpegts -muxpreload 0 -muxdelay 0 pipe:'.split()
-    command[command.index('FILEPATH')] = filepath  # handles special characters in filepath
-    proc = subprocess.Popen(
-        command,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.DEVNULL if self.logger.level >= logging.INFO else None)
+    # or by adjusting the timestamps in the output stream
+    from clarifai.runners.utils import video_utils
+    stream = video_utils.recontain_as_streamable(filepath)
 
+    # TODO accumulate reads to fill the chunk size
     chunk_size = 1024 * 1024  # 1 MB
-    chunk_iterator = iter(lambda: proc.stdout.read(chunk_size), b'')
+    chunk_iterator = iter(lambda: stream.read(chunk_size), b'')
 
     return self.stream_by_bytes(chunk_iterator, input_type, compute_cluster_id, nodepool_id,
                                 deployment_id, user_id, inference_params, output_config)

clarifai/runners/utils/video_utils.py

@@ -1,5 +1,7 @@
 import io
+import os
 import tempfile
+import threading
 
 import av
 import requests
@@ -55,3 +57,42 @@ def stream_video_from_bytes(bytes_iterator):
   reader = io.BufferedReader(buffer)
   container = av.open(reader)
   yield from container.decode(video=0)
+
+
+def recontain_as_streamable(filepath):
+  return recontain(filepath, "mpegts", {"muxpreload": "0", "muxdelay": "0"})
+
+
+def recontain(input, format, options={}):
+  # pyav-only implementation of "ffmpeg -i filepath -f mpegts -muxpreload 0 -muxdelay 0 pipe:"
+  read_pipe_fd, write_pipe_fd = os.pipe()
+  read_pipe = os.fdopen(read_pipe_fd, "rb")
+  write_pipe = os.fdopen(write_pipe_fd, "wb")
+
+  def _run_av():
+    input_container = output_container = None
+    try:
+      # open input and output containers, using mpegts as output format
+      input_container = av.open(input, options=options)
+      output_container = av.open(write_pipe, mode="w", format=format)
+
+      # Copy streams directly without re-encoding
+      for stream in input_container.streams:
+        output_container.add_stream_from_template(stream)
+
+      # Read packets from input and write them to output
+      for packet in input_container.demux():
+        if not packet.size:
+          break
+        output_container.mux(packet)
+
+    finally:
+      if output_container:
+        output_container.close()
+      if input_container:
+        input_container.close()
+
+  t = threading.Thread(target=_run_av)
+  t.start()
+
+  return read_pipe