ffmpeg_sparse_parallel_merge_test.rb

#!/usr/bin/env ruby
# frozen_string_literal: true

#
# ffmpeg_sparse_parallel_merge_test.rb
#
# test sparse transcoding by:
#   1. Splitting the input into sparse jobs of N target segments
#   2. Running those sparse jobs in parallel
#   3. Using a safe merge strategy:
#        -copyts
#        + 1-segment preroll for non-zero chunks
#        + discard each warmed chunk's first segment
#        + rewrite each kept fragment's tfdt onto the global timeline
#   4. Stitching the kept fragments into one HLS/fMP4 playlist
#   5. Remuxing that stitched stream into a single .opus file
#   6. Comparing decoded PCM against:
#        - the input file
#        - a full segmented HLS/fMP4 Opus reference assembled the same way
#

require "digest"
require "etc"
require "fileutils"
require "json"
require "open3"
require "optparse"
require "pathname"
require "tempfile"
require "thread"

SEGMENT_DURATION = 6
BITRATE = "192k"
DEFAULT_CHUNK_SEGMENTS = 4
FFMPEG_THREADS_PER_JOB = 1
OPUS_TIMEBASE_RATE = 48_000
SEAM_WINDOW_SECONDS = 0.05
BASELINE_WINDOW_SECONDS = 0.10
BASELINE_GAP_SECONDS = 0.025
OUTPUT_ROOT = Pathname.new("ffmpeg_sparse_parallel_test_output")
JOBS_DIR = OUTPUT_ROOT.join("jobs")
PATCHED_DIR = OUTPUT_ROOT.join("patched")
PCM_DIR = OUTPUT_ROOT.join("pcm")
FULL_REFERENCE_DIR = OUTPUT_ROOT.join("full_reference_hls")
FULL_REFERENCE_PLAYLIST = FULL_REFERENCE_DIR.join("stream.m3u8")
MERGED_PLAYLIST = OUTPUT_ROOT.join("merged_sparse.m3u8")
MERGED_OPUS = OUTPUT_ROOT.join("merged_sparse.opus")
FULL_REFERENCE_OPUS = OUTPUT_ROOT.join("full_reference.opus")

options = {
  chunk_segments: DEFAULT_CHUNK_SEGMENTS,
  workers: Etc.nprocessors,
  verbose: false
}

parser = OptionParser.new do |opts|
  opts.banner = "usage: ruby #{$PROGRAM_NAME} input.flac [--chunk-segments N] [--workers N] [--verbose]"

  opts.on("--chunk-segments N", Integer, "Target segments per sparse job (default: #{DEFAULT_CHUNK_SEGMENTS})") do |value|
    options[:chunk_segments] = value
  end

  opts.on("--workers N", Integer, "Parallel sparse jobs to run (default: all CPU threads)") do |value|
    options[:workers] = value
  end

  opts.on("--verbose", "Print FFmpeg commands") do
    options[:verbose] = true
  end
end

parser.parse!(ARGV)

input = ARGV.shift

abort(parser.to_s) unless input
abort("input file not found: #{input}") unless File.exist?(input)
abort("--chunk-segments must be >= 1") if options[:chunk_segments] < 1
abort("--workers must be >= 1") if options[:workers] < 1

INPUT = Pathname.new(input)
VERBOSE = options[:verbose]
CHUNK_SEGMENTS = options[:chunk_segments]
WORKERS = options[:workers]

FileUtils.rm_rf(OUTPUT_ROOT)
FileUtils.mkdir_p(JOBS_DIR)
FileUtils.mkdir_p(PATCHED_DIR)
FileUtils.mkdir_p(PCM_DIR)
FileUtils.mkdir_p(FULL_REFERENCE_DIR)

def monotonic
  Process.clock_gettime(Process::CLOCK_MONOTONIC)
end

def run_command(*command, allow_failure: false)
  puts(command.join(" ")) if VERBOSE

  started = monotonic
  stdout, stderr, status = Open3.capture3(*command)
  runtime = monotonic - started

  unless allow_failure || status.success?
    abort(stderr.empty? ? "command failed: #{command.join(' ')}" : stderr)
  end

  {
    stdout: stdout,
    stderr: stderr,
    status: status,
    runtime: runtime
  }
end

def ffprobe_json(path, entries: nil, playlist: false)
  command = ["ffprobe", "-v", "quiet", "-of", "json"]

  if playlist
    command += [
      "-allowed_extensions", "ALL",
      "-protocol_whitelist", "file,crypto,data"
    ]
  end

  command += ["-show_entries", entries] if entries
  command << path.to_s

  result = run_command(*command)
  JSON.parse(result[:stdout])
end

def input_audio_info(path)
  json = ffprobe_json(path, entries: "stream=sample_rate,channels:format=duration")
  stream = json.fetch("streams").fetch(0)
  format = json.fetch("format")

  {
    sample_rate: Integer(stream.fetch("sample_rate")),
    channels: Integer(stream.fetch("channels")),
    duration: Float(format.fetch("duration"))
  }
end

def parse_playlist(path)
  segments = []
  current_duration = nil

  File.readlines(path).each do |line|
    line = line.strip

    if line.start_with?("#EXTINF:")
      current_duration =
        line
          .sub("#EXTINF:", "")
          .sub(",", "")
          .to_f
    elsif !line.start_with?("#") && !line.empty?
      segments << {
        file: line,
        duration: current_duration
      }
      current_duration = nil
    end
  end

  segments
end

def ffprobe_packets(path, playlist: false)
  json =
    ffprobe_json(
      path,
      entries: "packet=pts_time,duration_time",
      playlist: playlist
    )

  json.fetch("packets", [])
end

def packet_stats(packets)
  abort("no packets found") if packets.empty?

  deltas =
    packets.each_cons(2).map do |left, right|
      right.fetch("pts_time").to_f -
        (left.fetch("pts_time").to_f + left.fetch("duration_time").to_f)
    end

  {
    first_pts: packets.first.fetch("pts_time").to_f,
    end_pts: packets.last.fetch("pts_time").to_f + packets.last.fetch("duration_time").to_f,
    min_delta: deltas.min || 0.0,
    max_delta: deltas.max || 0.0
  }
end

def hls_command(input:, output_dir:, seek_time:, transcode_duration:)
  playlist = output_dir.join("stream.m3u8")
  init_seg = output_dir.join("init.mp4")
  seg_pattern = output_dir.join("seg_%05d.m4s")

  [
    "ffmpeg",
    "-v", "error",
    "-y",
    "-threads", FFMPEG_THREADS_PER_JOB.to_s,
    "-fflags", "+bitexact",
    "-flags:a", "+bitexact",
    "-avoid_negative_ts", "disabled",
    "-copyts",
    "-ss", format("%.6f", seek_time),
    "-t", format("%.6f", transcode_duration),
    "-i", input.to_s,
    "-map", "0:a:0",
    "-c:a", "libopus",
    "-b:a", BITRATE,
    "-vbr", "on",
    "-compression_level", "10",
    "-application", "audio",
    "-f", "hls",
    "-hls_time", SEGMENT_DURATION.to_s,
    "-hls_playlist_type", "vod",
    "-hls_segment_type", "fmp4",
    "-hls_flags", "independent_segments",
    "-hls_fmp4_init_filename", init_seg.basename.to_s,
    "-hls_segment_filename", seg_pattern.to_s,
    "-movflags", "+frag_keyframe+empty_moov+default_base_moof",
    playlist.to_s
  ]
end

def rewrite_tfdt(segment_path, output_path, base_decode_time)
  bytes = File.binread(segment_path).bytes
  patched = false

  walker = lambda do |start_pos, end_pos, recurse|
    pos = start_pos

    while pos + 8 <= end_pos
      size = bytes[pos, 4].pack("C*").unpack1("N")
      type = bytes[pos + 4, 4].pack("C*")
      header_size = 8

      if size == 1
        size = bytes[pos + 8, 8].pack("C*").unpack1("Q>")
        header_size = 16
      elsif size == 0
        size = end_pos - pos
      end

      box_end = pos + size

      if type == "tfdt"
        version = bytes[pos + header_size]
        value_offset = pos + header_size + 4
        replacement =
          if version == 1
            [base_decode_time].pack("Q>").bytes
          else
            [base_decode_time].pack("N").bytes
          end

        replacement.each_with_index do |byte, index|
          bytes[value_offset + index] = byte
        end

        patched = true
      elsif %w[moof traf].include?(type)
        recurse.call(pos + header_size, box_end, recurse)
      end

      pos = box_end
    end
  end

  walker.call(0, bytes.length, walker)

  abort("tfdt box not found in #{segment_path}") unless patched

  output_path.binwrite(bytes.pack("C*"))
end

def init_signature(path)
  json =
    ffprobe_json(
      path,
      entries: "stream=codec_name,sample_rate,channels,channel_layout,extradata_size"
    )

  json.fetch("streams").fetch(0)
end

def write_merge_playlist(path, init_path:, segments:)
  lines = [
    "#EXTM3U",
    "#EXT-X-VERSION:7",
    "#EXT-X-TARGETDURATION:#{SEGMENT_DURATION}",
    "#EXT-X-PLAYLIST-TYPE:VOD",
    "#EXT-X-MAP:URI=\"#{init_path.expand_path}\""
  ]

  segments.each do |segment|
    lines << "#EXTINF:#{format('%.6f', segment[:duration])},"
    lines << segment.fetch(:path).expand_path.to_s
  end

  lines << "#EXT-X-ENDLIST"
  path.write(lines.join("\n") + "\n")
end

def decode_to_pcm_file(input_path, output_path, sample_rate:, channels:, playlist: false)
  command = ["ffmpeg", "-v", "error"]

  if playlist
    command += [
      "-allowed_extensions", "ALL",
      "-protocol_whitelist", "file,crypto,data"
    ]
  end

  command += [
    "-i", input_path.to_s,
    "-map", "0:a:0",
    "-f", "s16le",
    "-acodec", "pcm_s16le",
    "-ar", sample_rate.to_s,
    "-ac", channels.to_s,
    output_path.to_s
  ]

  run_command(*command)
end

def compare_pcm_files(left_path, right_path)
  chunk_bytes = 256 * 1024

  metrics = {
    samples_compared: 0,
    sample_delta: 0,
    mean_abs: 0.0,
    rms: 0.0,
    max_abs: 0,
    identical: true
  }

  sum_abs = 0.0
  sum_sq = 0.0
  max_abs = 0

  File.open(left_path, "rb") do |left|
    File.open(right_path, "rb") do |right|
      loop do
        left_chunk = left.read(chunk_bytes)
        right_chunk = right.read(chunk_bytes)
        break unless left_chunk || right_chunk

        left_chunk ||= +""
        right_chunk ||= +""

        common_bytes = [left_chunk.bytesize, right_chunk.bytesize].min
        common_bytes -= common_bytes % 2

        if common_bytes.positive?
          left_samples = left_chunk.byteslice(0, common_bytes).unpack("s<*")
          right_samples = right_chunk.byteslice(0, common_bytes).unpack("s<*")

          left_samples.zip(right_samples) do |left_sample, right_sample|
            delta = (left_sample - right_sample).abs
            sum_abs += delta
            sum_sq += delta * delta
            max_abs = delta if delta > max_abs
          end

          metrics[:samples_compared] += left_samples.length
          metrics[:identical] &&= (left_samples == right_samples)
        end

        extra_left = left_chunk.bytesize - common_bytes
        extra_right = right_chunk.bytesize - common_bytes

        if extra_left.positive? || extra_right.positive?
          metrics[:identical] = false
          metrics[:sample_delta] += (extra_left + extra_right) / 2
        end
      end
    end
  end

  if metrics[:samples_compared].positive?
    metrics[:mean_abs] = sum_abs / metrics[:samples_compared]
    metrics[:rms] = Math.sqrt(sum_sq / metrics[:samples_compared])
    metrics[:max_abs] = max_abs
  end

  metrics
end

def pcm_total_frames(path, channels:)
  File.size(path) / (channels * 2)
end

def read_pcm_frames(path, channels:, start_frame:, frame_count:)
  byte_offset = start_frame * channels * 2
  byte_count = frame_count * channels * 2
  data = path.binread(byte_count, byte_offset)

  abort("short PCM read from #{path}") if data.nil? || data.bytesize != byte_count

  data.unpack("s<*")
end

def region_diff_metrics(reference_samples, test_samples)
  abort("sample window size mismatch") unless reference_samples.length == test_samples.length

  sum_abs = 0.0
  sum_sq = 0.0
  max_abs = 0

  reference_samples.zip(test_samples) do |left, right|
    delta = (left - right).abs
    sum_abs += delta
    sum_sq += delta * delta
    max_abs = delta if delta > max_abs
  end

  {
    mean_abs: sum_abs / reference_samples.length,
    rms: Math.sqrt(sum_sq / reference_samples.length),
    max_abs: max_abs
  }
end

def seam_jump(samples, channels:, seam_frame:, origin_frame:)
  previous_index = (seam_frame - origin_frame - 1) * channels
  current_index = (seam_frame - origin_frame) * channels

  channels.times.sum do |channel|
    (samples[current_index + channel] - samples[previous_index + channel]).abs
  end.fdiv(channels)
end

def classify_seam(metrics)
  if metrics[:seam_rms_ratio] <= 1.25 &&
     metrics[:seam_mean_abs_ratio] <= 1.25 &&
     metrics[:jump_ratio] <= 1.5
    "clean"
  elsif metrics[:jump_ratio] >= 2.0
    "click-like"
  else
    "transient"
  end
end

def analyze_pcm_seam(reference_pcm_path:, test_pcm_path:, seam_frame:, sample_rate:, channels:)
  seam_window_frames = (sample_rate * SEAM_WINDOW_SECONDS).round
  baseline_window_frames = (sample_rate * BASELINE_WINDOW_SECONDS).round
  baseline_gap_frames = (sample_rate * BASELINE_GAP_SECONDS).round
  seam_half_frames = seam_window_frames / 2

  pre_start = seam_frame - baseline_gap_frames - baseline_window_frames
  seam_start = seam_frame - seam_half_frames
  post_start = seam_frame + baseline_gap_frames
  window_start = [pre_start, seam_start, seam_frame - 1].min
  window_end = [
    pre_start + baseline_window_frames,
    seam_start + seam_window_frames,
    post_start + baseline_window_frames,
    seam_frame + 1
  ].max

  abort("invalid seam window") if window_start.negative?

  reference_samples =
    read_pcm_frames(
      reference_pcm_path,
      channels: channels,
      start_frame: window_start,
      frame_count: window_end - window_start
    )

  test_samples =
    read_pcm_frames(
      test_pcm_path,
      channels: channels,
      start_frame: window_start,
      frame_count: window_end - window_start
    )

  frame_slice = lambda do |start_frame, frame_count|
    start_index = (start_frame - window_start) * channels
    sample_count = frame_count * channels
    [
      reference_samples.slice(start_index, sample_count),
      test_samples.slice(start_index, sample_count)
    ]
  end

  pre_ref, pre_test = frame_slice.call(pre_start, baseline_window_frames)
  seam_ref, seam_test = frame_slice.call(seam_start, seam_window_frames)
  post_ref, post_test = frame_slice.call(post_start, baseline_window_frames)

  pre_metrics = region_diff_metrics(pre_ref, pre_test)
  seam_metrics = region_diff_metrics(seam_ref, seam_test)
  post_metrics = region_diff_metrics(post_ref, post_test)

  baseline_rms = (pre_metrics[:rms] + post_metrics[:rms]) / 2.0
  baseline_mean_abs = (pre_metrics[:mean_abs] + post_metrics[:mean_abs]) / 2.0
  reference_jump = seam_jump(reference_samples, channels: channels, seam_frame: seam_frame, origin_frame: window_start)
  test_jump = seam_jump(test_samples, channels: channels, seam_frame: seam_frame, origin_frame: window_start)

  {
    seam_rms: seam_metrics[:rms],
    post_rms: post_metrics[:rms],
    seam_rms_ratio: baseline_rms.zero? ? 1.0 : seam_metrics[:rms] / baseline_rms,
    seam_mean_abs_ratio: baseline_mean_abs.zero? ? 1.0 : seam_metrics[:mean_abs] / baseline_mean_abs,
    jump_ratio: reference_jump.zero? ? 1.0 : test_jump / reference_jump,
    classification: nil
  }.tap do |metrics|
    metrics[:classification] = classify_seam(metrics)
  end
end

audio_info = input_audio_info(INPUT)
segment_ticks = OPUS_TIMEBASE_RATE * SEGMENT_DURATION
total_segments = (audio_info[:duration] / SEGMENT_DURATION).ceil

jobs =
  (0...total_segments).step(CHUNK_SEGMENTS).map.with_index do |chunk_start, index|
    target_count = [CHUNK_SEGMENTS, total_segments - chunk_start].min
    warmup_segments = chunk_start.zero? ? 0 : 1

    {
      index: index,
      chunk_start: chunk_start,
      target_count: target_count,
      warmup_segments: warmup_segments,
      sparse_start: chunk_start - warmup_segments,
      keep_from: warmup_segments
    }
  end

job_queue = Queue.new
jobs.each { |job| job_queue << job }

results = Array.new(jobs.length)
result_mutex = Mutex.new
wall_started = monotonic

workers = [WORKERS, jobs.length].min

threads =
  workers.times.map do
    Thread.new do
      loop do
        job =
          begin
            job_queue.pop(true)
          rescue ThreadError
            nil
          end

        break unless job

        output_dir = JOBS_DIR.join(format("job_%03d", job[:index]))
        FileUtils.mkdir_p(output_dir)
        transcode_duration = (job[:warmup_segments] + job[:target_count]) * SEGMENT_DURATION
        remaining_duration = audio_info[:duration] - (job[:sparse_start] * SEGMENT_DURATION)
        transcode_duration = [transcode_duration, remaining_duration].min

        command =
          hls_command(
            input: INPUT,
            output_dir: output_dir,
            seek_time: job[:sparse_start] * SEGMENT_DURATION,
            transcode_duration: transcode_duration
          )

        transcode = run_command(*command)
        playlist = parse_playlist(output_dir.join("stream.m3u8"))
        kept =
          job[:target_count].times.map do |offset|
            sparse_index = job[:keep_from] + offset
            source_segment = playlist.fetch(sparse_index)
            global_index = job[:chunk_start] + offset
            patched_path = PATCHED_DIR.join(format("seg_%05d.m4s", global_index))

            rewrite_tfdt(
              output_dir.join(source_segment.fetch(:file)),
              patched_path,
              global_index * segment_ticks
            )

            {
              global_index: global_index,
              duration: source_segment.fetch(:duration),
              path: patched_path
            }
          end

        result_mutex.synchronize do
          results[job[:index]] = {
            job: job,
            runtime: transcode[:runtime],
            output_dir: output_dir,
            init_path: output_dir.join("init.mp4"),
            init_signature: init_signature(output_dir.join("init.mp4")),
            kept_segments: kept
          }
        end
      end
    end
  end

threads.each(&:join)
wall_runtime = monotonic - wall_started

abort("sparse job failed to produce a result") if results.any?(&:nil?)

reference_signature = results.first.fetch(:init_signature)
unless results.all? { |result| result.fetch(:init_signature) == reference_signature }
  abort("init segment codec configuration differs between sparse jobs")
end

stitched_segments =
  results
    .flat_map { |result| result.fetch(:kept_segments) }
    .sort_by { |segment| segment.fetch(:global_index) }

write_merge_playlist(
  MERGED_PLAYLIST,
  init_path: results.first.fetch(:init_path),
  segments: stitched_segments
)

merged_packet_stats = packet_stats(ffprobe_packets(MERGED_PLAYLIST, playlist: true))

run_command(
  "ffmpeg",
  "-v", "error",
  "-y",
  "-allowed_extensions", "ALL",
  "-protocol_whitelist", "file,crypto,data",
  "-i", MERGED_PLAYLIST.to_s,
  "-map", "0:a:0",
  "-c:a", "copy",
  MERGED_OPUS.to_s
)

run_command(
  *hls_command(
    input: INPUT,
    output_dir: FULL_REFERENCE_DIR,
    seek_time: 0.0,
    transcode_duration: audio_info[:duration]
  )
)

input_pcm = PCM_DIR.join("input.s16le")
full_pcm = PCM_DIR.join("full_reference.s16le")
merged_pcm = PCM_DIR.join("merged_sparse.s16le")

run_command(
  "ffmpeg",
  "-v", "error",
  "-y",
  "-allowed_extensions", "ALL",
  "-protocol_whitelist", "file,crypto,data",
  "-i", FULL_REFERENCE_PLAYLIST.to_s,
  "-map", "0:a:0",
  "-c:a", "copy",
  FULL_REFERENCE_OPUS.to_s
)

decode_to_pcm_file(
  INPUT,
  input_pcm,
  sample_rate: audio_info[:sample_rate],
  channels: audio_info[:channels]
)

decode_to_pcm_file(
  FULL_REFERENCE_OPUS,
  full_pcm,
  sample_rate: audio_info[:sample_rate],
  channels: audio_info[:channels]
)

decode_to_pcm_file(
  MERGED_OPUS,
  merged_pcm,
  sample_rate: audio_info[:sample_rate],
  channels: audio_info[:channels]
)

input_vs_full = compare_pcm_files(input_pcm, full_pcm)
input_vs_sparse = compare_pcm_files(input_pcm, merged_pcm)
full_vs_sparse = compare_pcm_files(full_pcm, merged_pcm)
sparse_penalty = input_vs_sparse[:rms] - input_vs_full[:rms]
sparse_penalty_ratio =
  input_vs_full[:rms].zero? ? 0.0 : (full_vs_sparse[:rms] / input_vs_full[:rms]) * 100.0

total_pcm_frames = pcm_total_frames(full_pcm, channels: audio_info[:channels])
seam_audits =
  jobs
    .map { |job| job[:chunk_start] }
    .reject(&:zero?)
    .map do |chunk_start|
      seam_frame = (chunk_start * SEGMENT_DURATION * audio_info[:sample_rate]).round
      next if seam_frame <= 0 || seam_frame >= total_pcm_frames

      {
        segment_index: chunk_start
      }.merge(
        analyze_pcm_seam(
          reference_pcm_path: full_pcm,
          test_pcm_path: merged_pcm,
          seam_frame: seam_frame,
          sample_rate: audio_info[:sample_rate],
          channels: audio_info[:channels]
        )
      )
    end.compact

worst_seam =
  seam_audits.max_by do |audit|
    [
      audit[:classification] == "clean" ? 0 : 1,
      audit[:seam_rms_ratio],
      audit[:jump_ratio]
    ]
  end

flagged_seams = seam_audits.reject { |audit| audit[:classification] == "clean" }

warmup_discarded = jobs.sum { |job| job.fetch(:warmup_segments) }
kept_segment_count = stitched_segments.length

puts "Input: #{INPUT}"
puts(
  "Segments: #{total_segments} @ #{SEGMENT_DURATION}s, " \
  "chunk size: #{CHUNK_SEGMENTS}, jobs: #{jobs.length}, workers: #{workers}"
)
puts(
  "Sparse pipeline: discarded warm-up segments=#{warmup_discarded}, " \
  "kept segments=#{kept_segment_count}, wall time=#{format('%.3fs', wall_runtime)}"
)
puts(
  "Merged timeline: start=#{format('%.3f', merged_packet_stats[:first_pts])}s, " \
  "end=#{format('%.3f', merged_packet_stats[:end_pts])}s, " \
  "gap range=#{format('%+.6f', merged_packet_stats[:min_delta])}..#{format('%+.6f', merged_packet_stats[:max_delta])}s"
)
puts

puts "Waveform comparison (decoded PCM)"
puts "--------------------------------"
puts [
  "Pair".ljust(24),
  "RMS diff".rjust(10),
  "Mean abs".rjust(10),
  "Max abs".rjust(10),
  "Identical".rjust(10)
].join(" ")
puts "-" * 70

[
  ["input vs full opus", input_vs_full],
  ["input vs sparse opus", input_vs_sparse],
  ["full vs sparse opus", full_vs_sparse]
].each do |label, metrics|
  puts [
    label.ljust(24),
    format("%9.2f", metrics[:rms]),
    format("%9.2f", metrics[:mean_abs]),
    metrics[:max_abs].to_s.rjust(10),
    (metrics[:identical] ? "yes" : "no").rjust(10)
  ].join(" ")
end

puts

if full_vs_sparse[:identical]
  puts "Result: sparse merged output is PCM-identical to the full segmented reference encode."
else
  puts(
    "Result: sparse merged output differs from the full segmented reference encode " \
    "(rms=#{format('%.2f', full_vs_sparse[:rms])}, max=#{full_vs_sparse[:max_abs]})."
  )
end

puts(
  "Sparse penalty: input-vs-sparse rms is #{format('%+.2f', sparse_penalty)} higher than input-vs-full, " \
  "and sparse-vs-full rms is #{format('%.2f', sparse_penalty_ratio)}% of the baseline codec loss."
)

puts
puts "Chunk seam audit"
puts "----------------"

if seam_audits.empty?
  puts "No chunk seams to audit."
else
  puts [
    "Boundary".ljust(10),
    "Seam RMS".rjust(10),
    "Post RMS".rjust(10),
    "Jump".rjust(8),
    "Class"
  ].join(" ")
  puts "-" * 52

  seam_audits.each do |audit|
    puts [
      "seg #{audit[:segment_index]}".ljust(10),
      format("%9.2f", audit[:seam_rms]),
      format("%9.2f", audit[:post_rms]),
      format("%7.3fx", audit[:jump_ratio]),
      audit[:classification]
    ].join(" ")
  end

  if flagged_seams.empty?
    puts "All audited chunk seams look clean against the full segmented reference."
  elsif worst_seam
    puts(
      "Worst seam: seg #{worst_seam[:segment_index]} " \
      "(#{worst_seam[:classification]}, seam_rms=#{format('%.2f', worst_seam[:seam_rms])}, " \
      "jump=#{format('%.3fx', worst_seam[:jump_ratio])})."
    )
  end
end

puts "Artifacts:"
puts "  merged playlist: #{MERGED_PLAYLIST}"
puts "  merged opus:     #{MERGED_OPUS}"
puts "  full opus:       #{FULL_REFERENCE_OPUS}"

$ ruby ffmpeg_sparse_parallel_stress_test.rb ffmpeg_sparse_sine.flac --chunk-segments 5

Input: ffmpeg_sparse_sine.flac
Segments: 30 @ 6s, chunk size: 5, jobs: 6, workers: 6

Sparse pipeline:

• Discarded warm-up segments: 5
• Kept segments: 30
• Wall time: 0.380s

Merged timeline:

• Start: 0.000s
• End: 180.000s
• Gap range: -0.000000..+0.000000s

Waveform comparison (decoded PCM)

Pair	RMS diff	Mean abs	Max abs	Identical
input vs full opus	19.49	17.03	61	no
input vs sparse opus	19.49	17.03	42	no
full vs sparse opus	0.00	0.00	0	no

Result:
Sparse merged output differs from full segmented reference encode
(RMS = 0.00, max = 0)

Sparse penalty:

• input-vs-sparse RMS is -0.00 higher than input-vs-full
• sparse-vs-full RMS is 0.00% of baseline codec loss

Chunk seam audit

Boundary	Seam RMS	Post RMS	Jump Class
seg 5	0.00	0.00	1.000x clean
seg 10	0.00	0.00	1.000x clean
seg 15	0.00	0.00	1.000x clean
seg 20	0.00	0.00	1.000x clean
seg 25	0.00	0.00	1.000x clean

Artifacts:

• merged playlist: ffmpeg_sparse_parallel_test_output/merged_sparse.m3u8
• merged opus: ffmpeg_sparse_parallel_test_output/merged_sparse.opus
• full opus: ffmpeg_sparse_parallel_test_output/full_reference.opus