feat: add new remuxer to normalize media within milliseconds

Author: bajankristof

lib/ffmpeg.rb

@@ -29,6 +29,7 @@
 require_relative 'ffmpeg/presets/dash/h264'
 require_relative 'ffmpeg/presets/h264'
 require_relative 'ffmpeg/raw_command_args'
+require_relative 'ffmpeg/remuxer'
 require_relative 'ffmpeg/reporters/output'
 require_relative 'ffmpeg/reporters/progress'
 require_relative 'ffmpeg/reporters/silence'
@@ -270,6 +271,40 @@ def ffprobe_popen3(*args, &)
       FFMPEG::IO.popen3(ffprobe_binary, *args, &)
     end
 
+    # Get the path to the exiftool binary.
+    # Returns nil if exiftool is not found in the PATH.
+    #
+    # @return [String, nil]
+    def exiftool_binary
+      return @exiftool_binary if defined?(@exiftool_binary)
+
+      @exiftool_binary = which('exiftool')
+    rescue Errno::ENOENT
+      @exiftool_binary = nil
+    end
+
+    # Set the path to the exiftool binary.
+    #
+    # @param path [String]
+    # @return [String]
+    # @raise [Errno::ENOENT] If the exiftool binary is not an executable.
+    def exiftool_binary=(path)
+      if path.is_a?(String) && !File.executable?(path)
+        raise Errno::ENOENT, "The exiftool binary, '#{path}', is not executable"
+      end
+
+      @exiftool_binary = path
+    end
+
+    # Safely captures the standard output and the standard error of the exiftool command.
+    #
+    # @param args [Array<String>] The arguments to pass to exiftool.
+    # @return [Array<String, Process::Status>] The standard output, the standard error, and the process status.
+    def exiftool_capture3(*args)
+      logger.debug(self) { "exiftool #{Shellwords.join(args)}" }
+      FFMPEG::IO.capture3(exiftool_binary, *args)
+    end
+
     # Cross-platform way of finding an executable in the $PATH.
     # See http://stackoverflow.com/questions/2108727/which-in-ruby-checking-if-program-exists-in-path-from-ruby
     #

lib/ffmpeg/command_args.rb

@@ -27,7 +27,7 @@ class << self
       #
       # @param media [FFMPEG::Media] The media to transcode.
       # @param context [Hash, nil] Additional context for composing the arguments.
-      # # @return [FFMPEG::CommandArgs] The new FFMPEG::CommandArgs object.
+      # @return [FFMPEG::CommandArgs] The new FFMPEG::CommandArgs object.
       def compose(media, context: nil, &block)
         new(media, context:).tap do |args|
           args.instance_exec(&block) if block_given?
@@ -66,7 +66,7 @@ def video_bit_rate(target_value, **kwargs)
       super(adjusted_video_bit_rate(target_value), **kwargs)
     end
 
-    # Sets the audio bit rate to the minimum of the current audio bit rate and the target value.
+    # Sets the minimum video bit rate to the minimum of the current video bit rate and the target value.
     # The target value can be an Integer or a String (e.g.: 128k or 1M).
     #
     # @param target_value [Integer, String] The target bit rate.
@@ -77,7 +77,7 @@ def min_video_bit_rate(target_value)
       super(adjusted_video_bit_rate(target_value))
     end
 
-    # Sets the audio bit rate to the minimum of the current audio bit rate and the target value.
+    # Sets the maximum video bit rate to the minimum of the current video bit rate and the target value.
     # The target value can be an Integer or a String (e.g.: 128k or 1M).
     #
     # @param target_value [Integer, String] The target bit rate.

lib/ffmpeg/filter.rb

@@ -18,7 +18,7 @@ class << self
       # @param filters [Array<Filter>] The filters to join.
       # @return [String] The filter chain.
       def join(*filters)
-        filters.compact.map(&:to_s).join(',')
+        filters.compact.join(',')
       end
     end
 

lib/ffmpeg/io.rb

@@ -10,32 +10,62 @@ module IO
     class << self
       attr_writer :timeout, :encoding
 
+      # Returns the I/O timeout in seconds. Defaults to 30.
+      #
+      # @return [Integer]
       def timeout
         return @timeout if defined?(@timeout)
 
         @timeout = 30
       end
 
+      # Returns the I/O encoding. Defaults to UTF-8.
+      #
+      # @return [Encoding]
       def encoding
         @encoding ||= Encoding::UTF_8
       end
 
+      # Encodes the string in-place using the configured encoding,
+      # replacing invalid and undefined characters.
+      #
+      # @param string [String] The string to encode.
+      # @return [String]
       def encode!(string)
         string.encode!(encoding, invalid: :replace, undef: :replace)
       end
 
+      # Extends the given IO object with the configured timeout, encoding,
+      # and the FFMPEG::IO module.
+      #
+      # @param io [IO] The IO object to extend.
+      # @return [IO]
       def extend!(io)
         io.timeout = timeout
         io.set_encoding(encoding, invalid: :replace, undef: :replace)
         io.extend(FFMPEG::IO)
       end
 
+      # Runs the given command and captures stdout, stderr, and the process status.
+      # Encodes the output using the configured encoding.
+      #
+      # @param cmd [Array<String>] The command to run.
+      # @return [Array<String, Process::Status>] stdout, stderr, and the process status.
       def capture3(*cmd)
         *io, status = Open3.capture3(*cmd)
         io.each(&method(:encode!))
         [*io, status]
       end
 
+      # Starts the given command and yields or returns stdin, stdout, stderr, and the wait thread.
+      # Each IO stream is extended with the configured timeout and encoding.
+      #
+      # @param cmd [Array<String>] The command to run.
+      # @yieldparam stdin [IO]
+      # @yieldparam stdout [FFMPEG::IO]
+      # @yieldparam stderr [FFMPEG::IO]
+      # @yieldparam wait_thr [Thread]
+      # @return [Process::Status, Array<IO, Thread>]
       def popen3(*cmd, &block)
         if block_given?
           Open3.popen3(*cmd) do |*io, wait_thr|
@@ -54,6 +84,10 @@ def popen3(*cmd, &block)
       end
     end
 
+    # Iterates over each line of the IO stream, yielding each line to the block.
+    #
+    # @param chomp [Boolean] Whether to include the line separator in each yielded line.
+    # @yieldparam line [String] Each line from the stream.
     def each(chomp: false, &block)
       # We need to run this loop in a separate thread to avoid
       # errors with exit signals being sent to the main thread.

lib/ffmpeg/media.rb

@@ -80,6 +80,30 @@ def initialize(path, *ffprobe_args, load: true, autoload: true)
       load! if load
     end
 
+    # Remuxes the media file to the given output path via stream copy.
+    # If the initial stream copy fails and the video codec supports Annex B
+    # extraction, it falls back to extracting raw streams and re-muxing with
+    # a corrected frame rate.
+    #
+    # @param output_path [String, Pathname] The output path for the remuxed file.
+    # @param timeout [Integer, nil] Timeout in seconds for each ffmpeg command.
+    # @yield [report] Reports from the ffmpeg command (see FFMPEG::Reporters).
+    # @return [FFMPEG::Transcoder::Status]
+    def remux(output_path, timeout: nil, &block)
+      Remuxer.new(timeout:).process(self, output_path, &block)
+    end
+
+    # Remuxes the media file to the given output path via stream copy,
+    # raising an error if the remux fails.
+    #
+    # @param output_path [String, Pathname] The output path for the remuxed file.
+    # @param timeout [Integer, nil] Timeout in seconds for each ffmpeg command.
+    # @yield [report] Reports from the ffmpeg command (see FFMPEG::Reporters).
+    # @return [FFMPEG::Transcoder::Status]
+    def remux!(output_path, timeout: nil, &block)
+      remux(output_path, timeout:, &block).assert!
+    end
+
     # Load the metadata of the multimedia file.
     #
     # @return [Boolean]

lib/ffmpeg/raw_command_args.rb

@@ -349,7 +349,7 @@ def bitstream_filters(*filters, stream_id: nil, stream_index: nil)
     # @param filters [Array<FFMPEG::Filter, String>] The filters to add.
     # @return [self]
     def filter_complex(*filters)
-      arg('filter_complex', filters.compact.map(&:to_s).join(';'))
+      arg('filter_complex', filters.compact.join(';'))
 
       self
     end

lib/ffmpeg/remuxer.rb

@@ -0,0 +1,154 @@
+# frozen_string_literal: true
+
+require 'fileutils'
+require 'tmpdir'
+
+require_relative 'media'
+
+module FFMPEG
+  # The Remuxer class is responsible for remuxing multimedia files via stream copy.
+  # It attempts a direct stream copy first, and if that fails (e.g. due to corrupted
+  # timestamps), it falls back to extracting raw Annex B streams and re-muxing them
+  # with a corrected frame rate.
+  #
+  # @example
+  #  remuxer = FFMPEG::Remuxer.new
+  #  status = remuxer.process('input.mp4', 'output.mp4')
+  #  status.success? # => true
+  class Remuxer
+    ANNEXB_CODEC_NAMES = %w[h264 hevc].freeze
+
+    # @param name [String, nil] An optional name for the remuxer.
+    # @param metadata [Hash, nil] Optional metadata to associate with the remuxer.
+    # @param checks [Array<Symbol, Proc>] Checks to run on the output to determine success.
+    # @param timeout [Integer, nil] Timeout in seconds for each ffmpeg command.
+    def initialize(name: nil, metadata: nil, checks: %i[exist?], timeout: nil)
+      @name = name
+      @metadata = metadata
+      @checks = checks
+      @timeout = timeout
+    end
+
+    class << self
+      # Returns true if the media has a video codec that supports lossless
+      # Annex B bitstream extraction (H.264 or H.265).
+      #
+      # @param media [FFMPEG::Media]
+      # @return [Boolean]
+      def annexb?(media)
+        media.video? && ANNEXB_CODEC_NAMES.include?(media.video_codec_name)
+      end
+    end
+
+    # Remuxes the media file to the given output path via stream copy.
+    # If the initial stream copy fails and the video codec supports Annex B
+    # extraction, it falls back to extracting raw streams and re-muxing with
+    # a corrected frame rate.
+    #
+    # @param media [String, Pathname, URI, FFMPEG::Media] The media file to remux.
+    # @param output_path [String, Pathname] The output path for the remuxed file.
+    # @yield [report] Reports from the ffmpeg command (see FFMPEG::Reporters).
+    # @return [FFMPEG::Transcoder::Status]
+    def process(media, output_path, &)
+      media = Media.new(media, load: false) unless media.is_a?(Media)
+
+      status = ffmpeg_copy(media, output_path, &)
+      return status if status.success?
+      return status unless self.class.annexb?(media)
+
+      Dir.mktmpdir do |tmpdir|
+        annexb_extname = media.video_codec_name == 'hevc' ? '.h265' : '.h264'
+        annexb_path = File.join(tmpdir, "remux#{annexb_extname}")
+        annexb_filter = annexb_filter(media)
+        annexb_status = ffmpeg_copy(media, '-map', '0:v:0', *annexb_filter, annexb_path, &)
+        return annexb_status unless annexb_status.success?
+
+        mka_path = File.join(tmpdir, 'remux.mka')
+        mka_status = ffmpeg_copy(media, '-vn', mka_path, &)
+        return mka_status unless mka_status.success?
+
+        video = annexb_status.media.first
+        audio = mka_status.media.first
+        frame_rate = detect_frame_rate(video, audio)
+
+        status = ffmpeg_copy(
+          [video, audio, media],
+          '-map', '0:v',
+          '-map', '1:a',
+          '-map_metadata', '2',
+          output_path,
+          inargs: %W[-r #{frame_rate}],
+          &
+        )
+        return status unless status.success?
+        return status unless FFMPEG.exiftool_binary
+
+        FFMPEG.exiftool_capture3(
+          '-overwrite_original',
+          "-rotation=#{media.rotation}",
+          output_path
+        ).tap do |_, stderr, exiftool_status|
+          next if exiftool_status.success?
+
+          status.warn!("ExifTool exited with non-zero status: #{exiftool_status.exitstatus}\n#{stderr.strip}")
+        end
+
+        status
+      end
+    end
+
+    # Remuxes the media file to the given output path via stream copy,
+    # raising an error if the remux fails.
+    #
+    # @param media [String, Pathname, URI, FFMPEG::Media] The media file to remux.
+    # @param output_path [String, Pathname] The output path for the remuxed file.
+    # @yield [report] Reports from the ffmpeg command (see FFMPEG::Reporters).
+    # @return [FFMPEG::Transcoder::Status]
+    def process!(media, output_path, &)
+      process(media, output_path, &).assert!
+    end
+
+    protected
+
+    def ffmpeg_copy(media, *args, inargs: [], &)
+      media = [media] unless media.is_a?(Array)
+
+      FFMPEG.ffmpeg_execute(
+        *inargs.map(&:to_s),
+        *media.map { ['-i', _1.path.to_s] }.flatten,
+        '-c',
+        'copy',
+        *args.map(&:to_s),
+        timeout: @timeout,
+        status: Transcoder::Status.new([args.last], checks: @checks),
+        &
+      )
+    end
+
+    def annexb_filter(media)
+      ['-bsf:v', "#{media.video_codec_name}_mp4toannexb"]
+    end
+
+    def detect_frame_rate(video, audio)
+      stdout, = FFMPEG.ffprobe_capture3(
+        '-v', 'quiet',
+        '-count_packets',
+        '-select_streams', 'v:0',
+        '-show_entries', 'stream=nb_read_packets',
+        '-of', 'csv=p=0',
+        video.path
+      )
+      frame_count = stdout.strip.to_i
+
+      stdout, = FFMPEG.ffprobe_capture3(
+        '-v', 'quiet',
+        '-show_entries', 'format=duration',
+        '-of', 'csv=p=0',
+        audio.path
+      )
+      duration = stdout.strip.to_f
+
+      (frame_count.to_f / duration).round
+    end
+  end
+end

lib/ffmpeg/reporters/output.rb

@@ -4,15 +4,25 @@ module FFMPEG
   module Reporters
     # Represents a raw output line from ffmpeg.
     class Output
+      # Returns true — raw output lines are always logged.
+      #
+      # @return [Boolean]
       def self.log? = true
+
+      # Returns true — this reporter matches every output line.
+      #
+      # @param _line [String]
+      # @return [Boolean]
       def self.match?(_line) = true
 
       attr_reader :output
 
+      # @param output [String] The raw output line from ffmpeg.
       def initialize(output)
         @output = output
       end
 
+      # @return [String]
       def to_s
         output
       end

lib/ffmpeg/reporters/progress.rb

@@ -6,8 +6,15 @@ module FFMPEG
   module Reporters
     # Represents the progress of an encoding operation.
     class Progress < Output
+      # Returns false — progress lines are not logged.
+      #
+      # @return [Boolean]
       def self.log? = false
 
+      # Returns true if the line is a progress line.
+      #
+      # @param line [String]
+      # @return [Boolean]
       def self.match?(line)
         line.match?(/^\s*(?:size|time|frame)=/)
       end