Skip to main content

rff/
transcode.rs

1//! Transcoding — the engine side of `ffmpeg`.
2//!
3//! The pipeline is the classic FFmpeg shape:
4//!
5//! ```text
6//!   demuxer ─▶ decoder ─▶ [filters] ─▶ encoder ─▶ muxer
7//!   (input)    (raw frames pass through the middle)      (output)
8//! ```
9//!
10//! [`TranscodeSpec`] is the declarative description of a job — inputs, an
11//! output, and the codecs to use. [`run`] resolves that spec against an
12//! [`Engine`] (finding the demuxer, decoders, encoders, and muxer) and then
13//! drives the loop to completion.
14//!
15//! Each input stream is either **transcoded** (decode → re-encode) when an
16//! output codec is requested for its media type, or **stream-copied**
17//! (remuxed packet-for-packet) when none is — the same `-c:v copy` distinction
18//! FFmpeg draws. Video runs through the `-vf` filter graph (scale/crop); audio
19//! is automatically resampled to a rate the target encoder accepts (FFmpeg's
20//! implicit `aresample`).
21
22use std::io::Read;
23use std::path::PathBuf;
24
25use rff_codec::{CodecParams, Decoder, Encoder};
26use rff_core::{
27    AudioFrame, CodecId, ColorRange, Dictionary, Error, Frame, MediaType, Packet, PixelFormat,
28    Result, SampleFormat, VideoFrame,
29};
30use rff_filter::{FilterChain, FilterComplex};
31
32use rff_format::{Muxer, Stream};
33use rff_resample::Resampler;
34
35use crate::Engine;
36
37/// Codecs whose decoded output is inherently full-range (they decode to RGB, or
38/// are defined on 0-255 samples). JPEG is the canonical case: the standard has
39/// no limited-range mode, which is why FFmpeg calls its pixel formats `yuvj*`.
40fn is_full_range_codec(codec: CodecId) -> bool {
41    matches!(
42        codec,
43        CodecId::Jpeg | CodecId::Png | CodecId::Gif | CodecId::Webp
44    )
45}
46
47/// What range this stream's decoded frames are actually in.
48///
49/// An `Unspecified` stream is not "unknown, do nothing" — every consumer has to
50/// pick something, so pick it here, once, explicitly:
51/// full-range for codecs that are defined that way, limited-range otherwise
52/// (the video convention, and what an untagged y4m means).
53fn effective_color_range(stream: &Stream) -> ColorRange {
54    match stream.color_range {
55        ColorRange::Unspecified if is_full_range_codec(stream.codec_id) => ColorRange::Full,
56        ColorRange::Unspecified => ColorRange::Limited,
57        explicit => explicit,
58    }
59}
60
61/// One input file for a job.
62#[derive(Debug, Clone)]
63pub struct InputSpec {
64    pub path: PathBuf,
65    /// Force a container format instead of guessing from the extension
66    /// (`ffmpeg -f <fmt> -i ...`).
67    pub format: Option<String>,
68}
69
70/// Per-stream codec selection for the output.
71#[derive(Debug, Clone)]
72pub struct StreamCodec {
73    pub codec: CodecId,
74    /// Codec options (`-b:v 2M`, `-crf 23`, ...).
75    pub options: Dictionary,
76    /// Output sample format pinned by the codec NAME (`-c:a pcm_s16le`) or by
77    /// `-sample_fmt`. `None` = take whatever the decoder emits. Without this,
78    /// raw-PCM output silently ignored the format half of its own codec name.
79    pub sample_format: Option<SampleFormat>,
80}
81
82/// Which input stream(s) a `-map` entry selects.
83#[derive(Debug, Clone)]
84pub enum MapSelector {
85    /// All streams of the input (`-map 0`).
86    All,
87    /// All streams of a media kind (`-map 0:v` / `0:a`).
88    Kind(MediaType),
89    /// One stream by index within the input (`-map 0:2`).
90    Index(usize),
91}
92
93/// One `-map` entry: pick stream(s) from input `input`.
94#[derive(Debug, Clone)]
95pub struct MapSpec {
96    pub input: usize,
97    pub selector: MapSelector,
98}
99
100/// The output file for a job.
101#[derive(Debug, Clone, Default)]
102pub struct OutputSpec {
103    pub path: PathBuf,
104    /// Force a container format instead of guessing from the extension
105    /// (`ffmpeg -f <fmt> ...`).
106    pub format: Option<String>,
107    /// Video codec for the output, if a video stream is produced.
108    pub video_codec: Option<StreamCodec>,
109    /// Audio codec for the output, if an audio stream is produced.
110    pub audio_codec: Option<StreamCodec>,
111    /// Subtitle codec (`-c:s subrip|webvtt`). Text-subtitle packets share one
112    /// contract (plain text + ms timing), so this relabels the stream for the
113    /// muxer rather than re-encoding anything.
114    pub subtitle_codec: Option<CodecId>,
115    /// Video filter graph (`-vf`), e.g. `scale=320:240,crop=...`. Applied to
116    /// decoded video frames before re-encoding (transcode streams only).
117    pub video_filters: Option<String>,
118    /// Multi-input filter graph (`-filter_complex`). Currently models `overlay`:
119    /// the last input is composited over input #0's video.
120    pub filter_complex: Option<String>,
121    /// Explicit stream selection (`-map`). Empty = default (all video + audio,
122    /// in input/stream order).
123    pub maps: Vec<MapSpec>,
124    /// Overwrite the output if it exists (`-y`); otherwise fail (`-n`).
125    pub overwrite: bool,
126    /// Stop after this many VIDEO frames (`-frames:v N`). `None` = whole input.
127    ///
128    /// This is a measurement-critical option: without it a harness that thinks
129    /// it is encoding a 50-frame prefix silently encodes the whole clip, and any
130    /// rate-vs-quality pairing computed against a prefix is then wrong.
131    pub max_video_frames: Option<u64>,
132    /// Output trim window start, in seconds (`-ss`). Decoded frames (or copied
133    /// packets) before this point are dropped; timestamps are shifted so the
134    /// output starts at zero.
135    pub trim_start: Option<f64>,
136    /// Output trim window end, in seconds (`-to`, or `-ss` + `-t`). Exclusive.
137    pub trim_end: Option<f64>,
138    /// Constant output frame rate (`-r`), as `(num, den)` — e.g. `(30000, 1001)`.
139    /// Frames are duplicated/dropped to hit it (FFmpeg's `fps` filter).
140    pub frame_rate: Option<(u32, u32)>,
141    /// Target audio sample rate (`-ar`). The encoder's accepted-rate list still
142    /// wins if it cannot take this exact rate (nearest accepted is used).
143    pub audio_rate: Option<u32>,
144    /// Target audio channel count (`-ac`): 1↔2 downmix/upmix.
145    pub audio_channels: Option<u16>,
146    /// Audio filter chain (`-af` / `-filter:a`), e.g. `volume=-6dB,atrim=...`.
147    pub audio_filters: Option<String>,
148    /// Output metadata (`-metadata title=...`), handed to muxers that carry it.
149    pub metadata: Dictionary,
150    /// Muxer options (`-hls_time 6` → `{"hls_time": "6"}`), for formats that
151    /// take them (HLS/DASH segmenting).
152    pub format_options: Dictionary,
153}
154
155/// A complete, declarative transcoding job.
156#[derive(Debug, Clone, Default)]
157pub struct TranscodeSpec {
158    pub inputs: Vec<InputSpec>,
159    pub outputs: Vec<OutputSpec>,
160}
161
162/// A successful run's summary (frames/packets moved, etc.). Fields will grow as
163/// the pipeline does.
164#[derive(Debug, Clone, Default)]
165pub struct TranscodeReport {
166    pub packets_written: u64,
167    pub frames_decoded: u64,
168}
169
170/// Output trim window (`-ss` / `-t` / `-to`), in seconds.
171#[derive(Clone, Copy)]
172struct Trim {
173    start: f64,
174    end: Option<f64>,
175}
176
177impl Trim {
178    fn from_spec(o: &OutputSpec) -> Option<Trim> {
179        if o.trim_start.is_none() && o.trim_end.is_none() {
180            return None;
181        }
182        Some(Trim {
183            start: o.trim_start.unwrap_or(0.0).max(0.0),
184            end: o.trim_end,
185        })
186    }
187
188    fn keeps(&self, t: f64) -> bool {
189        t >= self.start - 1e-9 && self.end.map_or(true, |e| t < e - 1e-9)
190    }
191}
192
193/// CFR conversion state (`-r`): input frames are snapped onto a fixed
194/// `num/den` fps grid, duplicating the previous frame across gaps and dropping
195/// frames that land on an already-filled slot. Emitted frames carry their slot
196/// index as pts (time base `den/num`).
197struct FpsConv {
198    num: u32,
199    den: u32,
200    next_slot: i64,
201    last: Option<VideoFrame>,
202}
203
204impl FpsConv {
205    fn new(num: u32, den: u32) -> FpsConv {
206        FpsConv {
207            num: num.max(1),
208            den: den.max(1),
209            next_slot: 0,
210            last: None,
211        }
212    }
213
214    fn slot_time(&self, slot: i64) -> f64 {
215        slot as f64 * self.den as f64 / self.num as f64
216    }
217
218    /// Feed one input frame at time `t` (seconds); returns the frames due
219    /// strictly before `t`, each stamped with its slot pts.
220    fn push(&mut self, frame: VideoFrame, t: f64) -> Vec<VideoFrame> {
221        let mut out = Vec::new();
222        while let Some(last) = &self.last {
223            if self.slot_time(self.next_slot) < t - 1e-9 {
224                let mut f = last.clone();
225                f.pts = Some(self.next_slot);
226                out.push(f);
227                self.next_slot += 1;
228            } else {
229                break;
230            }
231        }
232        self.last = Some(frame);
233        out
234    }
235
236    /// End of stream: emit the held frame in its slot.
237    fn finish(&mut self) -> Option<VideoFrame> {
238        self.last.take().map(|mut f| {
239            f.pts = Some(self.next_slot);
240            self.next_slot += 1;
241            f
242        })
243    }
244}
245
246/// What to do with one input stream on its way to the output.
247enum StreamOp {
248    /// Not selected (or a media type we can't route): drop its packets.
249    Skip,
250    /// Remux packets unchanged into the output.
251    Copy(CopyOp),
252    /// Decode, filter, re-encode.
253    Transcode(Box<TranscodeOp>),
254}
255
256/// Stream-copy: forward packets, applying the trim window at packet level.
257/// Video cuts on the first keyframe inside the window (a copy can't split a
258/// GOP); timestamps shift so the output starts at zero.
259struct CopyOp {
260    out_index: usize,
261    is_video: bool,
262    /// Trim window in stream ticks (`None` = no trimming).
263    start_ticks: i64,
264    end_ticks: Option<i64>,
265    trimming: bool,
266    /// Video: set once the first in-window keyframe has passed.
267    started: bool,
268}
269
270impl CopyOp {
271    /// Apply trim/shift; `None` = drop the packet.
272    fn process(&mut self, mut packet: Packet) -> Option<Packet> {
273        if self.trimming {
274            let ts = packet.pts.or(packet.dts).unwrap_or(0);
275            if ts < self.start_ticks {
276                return None;
277            }
278            if let Some(end) = self.end_ticks {
279                if ts >= end {
280                    return None;
281                }
282            }
283            if self.is_video && !self.started {
284                if !packet.flags.keyframe {
285                    return None; // wait for a clean random-access point
286                }
287                self.started = true;
288            }
289            packet.pts = packet.pts.map(|p| p - self.start_ticks);
290            packet.dts = packet.dts.map(|d| d - self.start_ticks);
291        }
292        packet.stream_index = self.out_index;
293        Some(packet)
294    }
295}
296
297/// Decode → trim → fps → filters/overlay → audio conform → encode → mux.
298struct TranscodeOp {
299    decoder: Box<dyn Decoder>,
300    encoder: Box<dyn Encoder>,
301    filters: FilterChain,
302    /// `-filter_complex` overlay: a pre-decoded frame composited onto each
303    /// of this stream's frames at `(x, y)` (after `filters`). Video only.
304    overlay: Option<(VideoFrame, u32, u32)>,
305    /// The input stream's time base (frame pts are in these ticks).
306    in_time_base: rff_core::Rational,
307    is_video: bool,
308    /// Output trim window; pts are shifted by `shift_ticks` after it.
309    trim: Option<Trim>,
310    shift_ticks: i64,
311    /// Fallback clock for trimming audio whose frames carry no pts.
312    audio_clock_samples: u64,
313    /// `-r`: CFR duplicate/drop stage.
314    fps: Option<FpsConv>,
315    /// `-af`: the audio filter chain (volume/atrim/...).
316    audio_chain: rff_filter::AudioFilterChain,
317    /// `-ac`: mix to this many channels before everything else (0 = keep).
318    target_channels: u16,
319    /// Sample rate the encoder needs (0 = no audio resampling required).
320    target_rate: u32,
321    /// Sample format the output pins (`-c:a pcm_s16le`), if any.
322    target_sample_format: Option<SampleFormat>,
323    /// Lazily built once the first audio frame reveals the input rate.
324    resampler: Option<Resampler>,
325    /// Pixel formats the encoder accepts (`None` = anything).
326    accepted_formats: Option<Vec<PixelFormat>>,
327    /// Lazily built conversion into an accepted format.
328    pixel_converter: Option<FilterChain>,
329    /// Range of the frames reaching the encoder, for that conversion.
330    source_range: ColorRange,
331    out_index: usize,
332    /// `-frames:v` limit for this output, and how many we have sent.
333    max_video_frames: Option<u64>,
334    video_frames_sent: u64,
335    /// pts of video frames handed to the encoder, FIFO. Video encoders that
336    /// don't timestamp their packets (VP9, H.264) get these stamped on in
337    /// [`TranscodeOp::drain`] — without it every Matroska block lands at t=0.
338    pending_pts: std::collections::VecDeque<Option<i64>>,
339}
340
341/// Composite the `-filter_complex` overlay onto a video frame, if one is set.
342/// Audio frames and the no-overlay case pass through untouched.
343fn apply_overlay(overlay: &Option<(VideoFrame, u32, u32)>, frame: Frame) -> Result<Frame> {
344    match (overlay, frame) {
345        (Some((over, x, y)), Frame::Video(v)) => {
346            Ok(Frame::Video(rff_filter::overlay(v, over, *x, *y)?))
347        }
348        (_, frame) => Ok(frame),
349    }
350}
351
352/// Apply a video filter chain to a frame. Filters are video-only; audio passes
353/// through untouched, and an empty chain is a no-op.
354fn apply_filters(filters: &mut FilterChain, frame: Frame) -> Result<Frame> {
355    if filters.is_empty() {
356        return Ok(frame);
357    }
358    match frame {
359        Frame::Video(v) => Ok(Frame::Video(filters.apply(v)?)),
360        other => Ok(other),
361    }
362}
363
364/// Pick the accepted rate closest to `target`.
365fn nearest_rate(rates: &[u32], target: u32) -> u32 {
366    rates
367        .iter()
368        .copied()
369        .min_by_key(|r| (*r as i64 - target as i64).abs())
370        .unwrap_or(target)
371}
372
373/// Read interleaved `s16`/`f32` plane 0 of an audio frame into `f32` samples.
374fn audio_to_f32(af: &AudioFrame) -> Result<Vec<f32>> {
375    match af.format {
376        SampleFormat::F32 => Ok(af.planes[0]
377            .chunks_exact(4)
378            .map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]]))
379            .collect()),
380        SampleFormat::S16 => Ok(af.planes[0]
381            .chunks_exact(2)
382            .map(|b| i16::from_le_bytes([b[0], b[1]]) as f32 / 32768.0)
383            .collect()),
384        other => Err(Error::unsupported(format!(
385            "resample: sample format `{}` (need interleaved s16/f32)",
386            other.name()
387        ))),
388    }
389}
390
391/// Wrap interleaved `f32` samples as an `f32` [`AudioFrame`].
392fn f32_frame(samples: Vec<f32>, rate: u32, channels: u16, pts: Option<i64>) -> Frame {
393    let bytes: Vec<u8> = samples.iter().flat_map(|s| s.to_le_bytes()).collect();
394    Frame::Audio(AudioFrame {
395        sample_rate: rate,
396        channels,
397        format: SampleFormat::F32,
398        samples: samples.len() / channels.max(1) as usize,
399        planes: vec![bytes],
400        pts,
401    })
402}
403
404/// Convert an audio frame to `target` sample format, if one is pinned and the
405/// frame is not already in it. The audio analogue of `conform_video`: raw PCM
406/// output declares its format in the container header, so the frames handed to
407/// the muxer must actually BE that format, not merely be labelled it.
408fn conform_sample_format(target: Option<SampleFormat>, frame: Frame) -> Result<Frame> {
409    let Some(target) = target else {
410        return Ok(frame);
411    };
412    let Frame::Audio(af) = frame else {
413        return Ok(frame);
414    };
415    if af.format == target {
416        return Ok(Frame::Audio(af));
417    }
418    let samples = audio_to_f32(&af)?;
419    let planes = match target {
420        SampleFormat::F32 => vec![samples.iter().flat_map(|s| s.to_le_bytes()).collect()],
421        SampleFormat::S16 => vec![samples
422            .iter()
423            .flat_map(|s| {
424                // Round-to-nearest and clamp, matching the usual f32->s16 rule:
425                // truncation would bias every sample toward zero.
426                let v = (s * 32768.0).round().clamp(-32768.0, 32767.0) as i16;
427                v.to_le_bytes()
428            })
429            .collect()],
430        other => {
431            return Err(Error::unsupported(format!(
432                "sample format conversion to `{}` (only interleaved s16/f32)",
433                other.name()
434            )))
435        }
436    };
437    Ok(Frame::Audio(AudioFrame {
438        sample_rate: af.sample_rate,
439        channels: af.channels,
440        format: target,
441        samples: af.samples,
442        planes,
443        pts: af.pts,
444    }))
445}
446
447/// Resample an audio frame to `target_rate` if needed (the resampler is built
448/// lazily from the first frame's rate/channels). `target_rate == 0` is a no-op.
449fn conform_audio(
450    resampler: &mut Option<Resampler>,
451    target_rate: u32,
452    frame: Frame,
453) -> Result<Frame> {
454    if target_rate == 0 {
455        return Ok(frame);
456    }
457    let Frame::Audio(af) = frame else {
458        return Ok(frame);
459    };
460    if af.sample_rate == target_rate {
461        return Ok(Frame::Audio(af));
462    }
463    let rs =
464        resampler.get_or_insert_with(|| Resampler::new(af.sample_rate, target_rate, af.channels));
465    let out = rs.process(&audio_to_f32(&af)?);
466    let pts = af
467        .pts
468        .map(|p| p * target_rate as i64 / af.sample_rate.max(1) as i64);
469    Ok(f32_frame(out, target_rate, af.channels, pts))
470}
471
472/// Convert a video frame into a pixel format the encoder accepts, if it isn't
473/// already in one. The converting chain is built lazily from the first frame
474/// that needs it, mirroring the audio resampler above.
475///
476/// This is what lets a decoder emit its cheapest native layout: the JPEG decoder
477/// hands out planar Y'CbCr, and a PNG encoder downstream still gets the RGB it
478/// requires because this inserts the conversion.
479fn conform_video(
480    converter: &mut Option<FilterChain>,
481    accepted: &Option<Vec<PixelFormat>>,
482    source_range: ColorRange,
483    frame: Frame,
484) -> Result<Frame> {
485    let Some(accepted) = accepted else {
486        return Ok(frame);
487    };
488    let Frame::Video(vf) = frame else {
489        return Ok(frame);
490    };
491    if accepted.is_empty() || accepted.contains(&vf.format) {
492        return Ok(Frame::Video(vf));
493    }
494    if converter.is_none() {
495        // First choice the encoder listed that we can actually convert to.
496        let target = accepted
497            .iter()
498            .find(|f| matches!(f, PixelFormat::Rgb24 | PixelFormat::Yuv420p))
499            .ok_or_else(|| {
500                Error::unsupported(format!(
501                    "no conversion from `{}` to any format this encoder accepts",
502                    vf.format.name()
503                ))
504            })?;
505        // Pin the range: `spec.limited` describes the Y'CbCr side either way,
506        // and guessing it wrong is a silent several-dB error.
507        let range = match source_range {
508            ColorRange::Limited => "limited",
509            _ => "full",
510        };
511        *converter = Some(FilterChain::parse(&format!(
512            "format={}:bt601:{}",
513            target.name(),
514            range
515        ))?);
516    }
517    let chain = converter.as_mut().expect("just built");
518    Ok(Frame::Video(chain.apply(vf)?))
519}
520
521/// Mix an audio frame to `target` channels (0 = keep). Stereo↔mono only: a
522/// downmix averages pairs, an upmix duplicates the channel.
523fn conform_channels(target: u16, frame: Frame) -> Result<Frame> {
524    if target == 0 {
525        return Ok(frame);
526    }
527    let Frame::Audio(af) = frame else {
528        return Ok(frame);
529    };
530    if af.channels == target {
531        return Ok(Frame::Audio(af));
532    }
533    let samples = audio_to_f32(&af)?;
534    let mixed: Vec<f32> = match (af.channels, target) {
535        (2, 1) => samples.chunks_exact(2).map(|p| 0.5 * (p[0] + p[1])).collect(),
536        (1, 2) => samples.iter().flat_map(|s| [*s, *s]).collect(),
537        (from, to) => {
538            return Err(Error::unsupported(format!(
539                "-ac: only 1<->2 channel mixing supported (input {from} ch, requested {to})"
540            )))
541        }
542    };
543    Ok(f32_frame(mixed, af.sample_rate, target, af.pts))
544}
545
546impl TranscodeOp {
547    /// A frame's start time in seconds (input timeline). Audio without pts
548    /// falls back to a running sample clock; video without pts cannot be
549    /// placed and errors — only called when trim/fps actually need the time.
550    fn frame_time(&self, frame: &Frame) -> Result<f64> {
551        let tb = self.in_time_base;
552        let tick = if tb.num > 0 && tb.den > 0 {
553            tb.num as f64 / tb.den as f64
554        } else {
555            0.001
556        };
557        match frame.pts() {
558            Some(p) => Ok(p as f64 * tick),
559            None => match frame {
560                Frame::Audio(af) => {
561                    Ok(self.audio_clock_samples as f64 / af.sample_rate.max(1) as f64)
562                }
563                Frame::Video(_) => Err(Error::unsupported(
564                    "-ss/-t/-to/-r need timestamps and this video stream has none",
565                )),
566            },
567        }
568    }
569
570    /// Decoded-frame entry point: trim window, timestamp shift, CFR stage,
571    /// then the per-frame pipeline in [`TranscodeOp::emit`].
572    fn handle_frame(
573        &mut self,
574        frame: Frame,
575        muxer: &mut dyn Muxer,
576        report: &mut TranscodeReport,
577    ) -> Result<()> {
578        report.frames_decoded += 1;
579        let needs_time = self.trim.is_some() || (self.is_video && self.fps.is_some());
580        let t = if needs_time {
581            Some(self.frame_time(&frame)?)
582        } else {
583            None
584        };
585        // The sample clock tracks ALL input audio, kept or dropped.
586        if let Frame::Audio(af) = &frame {
587            self.audio_clock_samples += af.samples as u64;
588        }
589        let mut frame = frame;
590        if let (Some(trim), Some(t)) = (self.trim, t) {
591            match frame {
592                // Audio frames can be arbitrarily long (a WAV decodes as one
593                // frame), so the window slices samples instead of keep/drop.
594                Frame::Audio(af) => {
595                    match rff_filter::trim_audio_frame(af, Some(t), Some(trim.start), trim.end)? {
596                        Some(af) => frame = Frame::Audio(af),
597                        None => return Ok(()),
598                    }
599                }
600                Frame::Video(_) => {
601                    if !trim.keeps(t) {
602                        return Ok(());
603                    }
604                }
605            }
606        }
607        // Shift timestamps so the output timeline starts at zero.
608        if self.shift_ticks != 0 {
609            match &mut frame {
610                Frame::Video(v) => v.pts = v.pts.map(|p| p - self.shift_ticks),
611                Frame::Audio(a) => a.pts = a.pts.map(|p| p - self.shift_ticks),
612            }
613        }
614        match frame {
615            Frame::Video(v) if self.fps.is_some() => {
616                let t_out = t.expect("fps requires time") - self.trim.map_or(0.0, |w| w.start);
617                let mut fps = self.fps.take().expect("checked");
618                let due = fps.push(v, t_out);
619                self.fps = Some(fps);
620                for f in due {
621                    self.emit(Frame::Video(f), muxer, report)?;
622                }
623                Ok(())
624            }
625            other => self.emit(other, muxer, report),
626        }
627    }
628
629    /// Post-fps pipeline: `-frames:v` gate, filters, overlay, audio conforms,
630    /// encode, drain.
631    fn emit(
632        &mut self,
633        frame: Frame,
634        muxer: &mut dyn Muxer,
635        report: &mut TranscodeReport,
636    ) -> Result<()> {
637        if let Some(limit) = self.max_video_frames {
638            if matches!(frame, Frame::Video(_)) {
639                if self.video_frames_sent >= limit {
640                    return Ok(());
641                }
642                self.video_frames_sent += 1;
643            }
644        }
645        let frame = apply_filters(&mut self.filters, frame)?;
646        let frame = apply_overlay(&self.overlay, frame)?;
647        let frame = conform_channels(self.target_channels, frame)?;
648        let frame = match frame {
649            Frame::Audio(af) if !self.audio_chain.is_empty() => {
650                let tb = self.in_time_base;
651                let t = af
652                    .pts
653                    .map(|p| p as f64 * tb.num as f64 / tb.den.max(1) as f64);
654                match self.audio_chain.apply(af, t)? {
655                    Some(af) => Frame::Audio(af),
656                    None => return Ok(()), // atrim consumed the whole frame
657                }
658            }
659            other => other,
660        };
661        let frame = conform_audio(&mut self.resampler, self.target_rate, frame)?;
662        let frame = conform_sample_format(self.target_sample_format, frame)?;
663        let frame = conform_video(
664            &mut self.pixel_converter,
665            &self.accepted_formats,
666            self.source_range,
667            frame,
668        )?;
669        if self.is_video {
670            self.pending_pts.push_back(frame.pts());
671        }
672        self.encoder.send_frame(&frame)?;
673        self.drain(muxer, report)
674    }
675
676    /// Pull every ready packet out of the encoder and mux it. Video encoders
677    /// that don't timestamp their packets (VP9, H.264 emit `pts: None`) get the
678    /// corresponding source frame's pts stamped on, FIFO — without this every
679    /// Matroska block would land at t=0.
680    fn drain(&mut self, muxer: &mut dyn Muxer, report: &mut TranscodeReport) -> Result<()> {
681        loop {
682            match self.encoder.receive_packet() {
683                Ok(mut packet) => {
684                    if self.is_video {
685                        let queued = self.pending_pts.pop_front().flatten();
686                        if packet.pts.is_none() {
687                            packet.pts = queued;
688                        }
689                    }
690                    packet.stream_index = self.out_index;
691                    muxer.write_packet(&packet)?;
692                    report.packets_written += 1;
693                }
694                Err(Error::Again) | Err(Error::Eof) => break,
695                Err(e) => return Err(e),
696            }
697        }
698        Ok(())
699    }
700
701    /// End of input: flush the decoder, the CFR stage's held frame, the
702    /// resampler's FIR tail, then the encoder.
703    fn finish(&mut self, muxer: &mut dyn Muxer, report: &mut TranscodeReport) -> Result<()> {
704        self.decoder.flush();
705        loop {
706            match self.decoder.receive_frame() {
707                Ok(frame) => self.handle_frame(frame, muxer, report)?,
708                Err(Error::Again) | Err(Error::Eof) => break,
709                Err(e) => return Err(e),
710            }
711        }
712        if let Some(mut fps) = self.fps.take() {
713            let held = fps.finish();
714            self.fps = Some(fps);
715            if let Some(f) = held {
716                self.emit(Frame::Video(f), muxer, report)?;
717            }
718        }
719        if let Some(rs) = &mut self.resampler {
720            let tail = rs.finish();
721            if !tail.is_empty() {
722                let frame = f32_frame(tail, rs.out_rate(), rs.channels(), None);
723                let frame = conform_sample_format(self.target_sample_format, frame)?;
724                self.encoder.send_frame(&frame)?;
725                self.drain(muxer, report)?;
726            }
727        }
728        self.encoder.flush();
729        self.drain(muxer, report)
730    }
731}
732
733/// Resolve and run a transcode job against `engine`.
734///
735/// Resolution (formats, decoders, encoders) happens up front so failures are
736/// reported before any output file is touched; then the demux→decode→encode→mux
737/// loop runs to completion.
738pub fn run(engine: &Engine, spec: &TranscodeSpec) -> Result<TranscodeReport> {
739    if spec.inputs.is_empty() {
740        return Err(Error::Option("no input files specified".into()));
741    }
742    if spec.outputs.is_empty() {
743        return Err(Error::Option("no output file specified".into()));
744    }
745    if spec.outputs.len() > 1 {
746        return Err(Error::unsupported("multiple outputs are not supported yet"));
747    }
748    let output = &spec.outputs[0];
749
750    // -filter_complex overlay: the last input is composited over input #0's
751    // video. Resolve it up front so stream selection can exclude that input.
752    let overlay_xy = output
753        .filter_complex
754        .as_deref()
755        .map(FilterComplex::parse)
756        .transpose()?
757        .and_then(|fc| fc.overlay);
758    let overlay_input = match overlay_xy {
759        Some(_) if spec.inputs.len() >= 2 => Some(spec.inputs.len() - 1),
760        Some(_) => {
761            return Err(Error::Option(
762                "filter_complex overlay needs a second input (the overlay image/video)".into(),
763            ))
764        }
765        None => None,
766    };
767
768    // --- open every input demuxer and read its streams ---
769    let mut demuxers: Vec<Box<dyn rff_format::Demuxer>> = Vec::new();
770    let mut input_streams: Vec<Vec<Stream>> = Vec::new();
771    for input in &spec.inputs {
772        let (in_format, reader) = open_input(engine, input)?;
773        let mut demuxer = engine.formats.open_demuxer(&in_format, reader)?;
774        input_streams.push(demuxer.read_header()?);
775        demuxers.push(demuxer);
776    }
777
778    // --- select which (input, stream) pairs go to the output, in order ---
779    let mut selection = select_streams(&input_streams, output)?;
780    // The overlay input is consumed by the filter, not muxed as its own stream.
781    if let Some(oin) = overlay_input {
782        selection.retain(|(inp, _)| *inp != oin);
783    }
784    if selection.is_empty() {
785        return Err(Error::unsupported("no streams selected for the output"));
786    }
787
788    // Single-codec containers can't stream-copy a foreign codec, so an unset
789    // `-c:v` must mean TRANSCODE to the container's one codec, not copy (which
790    // the muxer would reject). y4m additionally can't carry anything compressed.
791    // Mirrors ffmpeg picking the muxer's default codec (`out.webp` → webp).
792    // When the source already IS that codec, `None` stays: stream-copy wins.
793    let mut output_owned;
794    let output = if output.video_codec.is_none() {
795        let format_name = resolve_output_format(engine, output).ok();
796        // y4m keeps its unconditional default: it can't carry anything
797        // compressed, so even rawvideo input goes through the decode path.
798        let required = match format_name.as_deref() {
799            Some("yuv4mpegpipe") => Some((CodecId::RawVideo, true)),
800            Some("webp") => Some((CodecId::Webp, false)),
801            Some("png") => Some((CodecId::Png, false)),
802            Some("jpeg") => Some((CodecId::Jpeg, false)),
803            Some("gif") => Some((CodecId::Gif, false)),
804            Some("avif") => Some((CodecId::Avif, false)),
805            Some("jpegxl") => Some((CodecId::Jxl, false)),
806            _ => None,
807        };
808        let apply = match required {
809            Some((_, true)) => true,
810            // Image containers: only when the source codec differs — a matching
811            // source keeps `None` and takes the stream-copy fast path.
812            Some((codec, false)) => !selection.iter().any(|&(inp, local)| {
813                let s = &input_streams[inp][local];
814                s.media_type == MediaType::Video && s.codec_id == codec
815            }),
816            None => false,
817        };
818        if apply {
819            output_owned = output.clone();
820            output_owned.video_codec = Some(StreamCodec {
821                codec: required.expect("apply implies required").0,
822                options: Dictionary::default(),
823                sample_format: None, // video: no audio sample format to pin
824            });
825            &output_owned
826        } else {
827            output
828        }
829    } else {
830        output
831    };
832
833    // Per-input op tables (Skip for unselected streams) + ordered output streams.
834    let mut per_input_ops: Vec<Vec<StreamOp>> = input_streams
835        .iter()
836        .map(|s| (0..s.len()).map(|_| StreamOp::Skip).collect())
837        .collect();
838    let mut out_streams: Vec<Stream> = Vec::new();
839    for (out_index, &(inp, local)) in selection.iter().enumerate() {
840        let (op, os) = build_op(engine, &input_streams[inp][local], output, out_index)?;
841        per_input_ops[inp][local] = op;
842        out_streams.push(os);
843    }
844
845    // --- filter_complex overlay: pre-decode the overlay frame and hand it to
846    // input #0's video transcode op (which composites it onto every frame) ---
847    if let (Some((x, y)), Some(oin)) = (overlay_xy, overlay_input) {
848        let over = decode_overlay_frame(engine, &mut *demuxers[oin], &input_streams[oin])?;
849        let vidx = input_streams[0]
850            .iter()
851            .position(|s| s.media_type == MediaType::Video)
852            .ok_or_else(|| Error::Option("filter_complex overlay: input #0 has no video".into()))?;
853        match &mut per_input_ops[0][vidx] {
854            StreamOp::Transcode(op) => op.overlay = Some((over, x, y)),
855            _ => {
856                return Err(Error::unsupported(
857                    "filter_complex overlay needs input #0's video re-encoded — pass -c:v",
858                ))
859            }
860        }
861    }
862
863    // --- pick the muxer from the registry. Path-based formats (HLS/DASH: a
864    // playlist plus many segment files) are built with the output path; every
865    // other format writes one byte sink through rff-io (file, pipe, udp).
866    let out_format = resolve_output_format(engine, output)?;
867    let muxer_path_factory = engine
868        .formats
869        .by_name(&out_format)
870        .filter(|fmt| fmt.can_mux())
871        .ok_or_else(|| Error::MuxerNotFound(out_format.clone()))?
872        .muxer_path;
873
874    // --- refuse to clobber an existing file unless -y was given (pipes and
875    // sockets have nothing to clobber) ---
876    let out_path_str = output.path.to_str().unwrap_or_default();
877    let is_sink_stream = rff_io::is_pipe(out_path_str)
878        || rff_io::is_udp(out_path_str)
879        || rff_io::is_rtmp(out_path_str);
880    if !is_sink_stream && !output.overwrite && output.path.exists() {
881        return Err(Error::Option(format!(
882            "{} already exists (pass -y to overwrite)",
883            output.path.display()
884        )));
885    }
886    let mut muxer: Box<dyn Muxer> = match muxer_path_factory {
887        Some(factory) => factory(&output.path, &output.format_options)?,
888        None => engine
889            .formats
890            .open_muxer(&out_format, rff_io::create(out_path_str)?)?,
891    };
892    if !output.metadata.is_empty() {
893        muxer.set_metadata(&output.metadata);
894    }
895    muxer.write_header(&out_streams)?;
896
897    let mut report = TranscodeReport::default();
898
899    // --- drive each input through its plan into the shared muxer ---
900    for (demuxer, ops) in demuxers.iter_mut().zip(per_input_ops.iter_mut()) {
901        loop {
902            match demuxer.read_packet() {
903                Ok(packet) => process_packet(packet, ops, &mut *muxer, &mut report)?,
904                Err(Error::Eof) => break,
905                Err(e) => return Err(e),
906            }
907        }
908        flush_streams(ops, &mut *muxer, &mut report)?;
909    }
910    muxer.write_trailer()?;
911
912    Ok(report)
913}
914
915/// Resolve which `(input_index, stream_index)` pairs go to the output, in
916/// output order. With no `-map`, defaults to every video + audio stream across
917/// all inputs (input order, then stream order).
918fn select_streams(inputs: &[Vec<Stream>], output: &OutputSpec) -> Result<Vec<(usize, usize)>> {
919    let mut selection = Vec::new();
920    if output.maps.is_empty() {
921        for (ii, streams) in inputs.iter().enumerate() {
922            for (i, s) in streams.iter().enumerate() {
923                if matches!(
924                    s.media_type,
925                    MediaType::Video | MediaType::Audio | MediaType::Subtitle
926                ) {
927                    selection.push((ii, i));
928                }
929            }
930        }
931    } else {
932        for map in &output.maps {
933            let streams = inputs
934                .get(map.input)
935                .ok_or_else(|| Error::Option(format!("-map: no input #{}", map.input)))?;
936            for (i, s) in streams.iter().enumerate() {
937                let hit = match &map.selector {
938                    MapSelector::All => true,
939                    MapSelector::Kind(k) => s.media_type == *k,
940                    MapSelector::Index(idx) => i == *idx,
941                };
942                if hit {
943                    selection.push((map.input, i));
944                }
945            }
946        }
947    }
948    Ok(selection)
949}
950
951/// Build the [`StreamOp`] (transcode or copy) and matching output [`Stream`] for
952/// one selected input stream at output position `out_index`.
953fn build_op(
954    engine: &Engine,
955    stream: &Stream,
956    output: &OutputSpec,
957    out_index: usize,
958) -> Result<(StreamOp, Stream)> {
959    let requested = match stream.media_type {
960        MediaType::Video => output.video_codec.as_ref(),
961        MediaType::Audio => output.audio_codec.as_ref(),
962        _ => None,
963    };
964    match requested {
965        // Transcode: decode the input codec, re-encode to the requested one.
966        Some(target) => {
967            let mut decoder = engine.codecs.find_decoder(stream.codec_id)?;
968            decoder.configure(&codec_params(stream))?;
969            let mut encoder = engine.codecs.find_encoder(target.codec)?;
970            encoder.configure(&target.options)?; // rate control: -crf / -preset / -b
971                                                 // Video filter graph (`-vf`); applies to video streams only.
972            let mut filters = if stream.media_type == MediaType::Video {
973                FilterChain::parse(output.video_filters.as_deref().unwrap_or(""))?
974            } else {
975                FilterChain::default()
976            };
977            // Colour range has to be reconciled before any filter runs. JPEG (and
978            // the other image codecs) are defined on full-range samples, so
979            // handing them limited-range video encodes the wrong signal — it
980            // measured 28.35 dB where it should have been 36.15 dB, a deficit
981            // that looked exactly like poor encoder quality. Convert once, up
982            // front, then tell the rest of the chain what it is now looking at.
983            let mut source_range = ColorRange::Unspecified;
984            if stream.media_type == MediaType::Video {
985                source_range = effective_color_range(stream);
986                if is_full_range_codec(target.codec) && source_range == ColorRange::Limited {
987                    filters.prepend_range_conversion(true);
988                    source_range = ColorRange::Full;
989                }
990                filters.set_source_color_range(source_range);
991            }
992            let (out_w, out_h) = filters.output_dims(stream.width, stream.height);
993            let is_video = stream.media_type == MediaType::Video;
994            let is_audio = stream.media_type == MediaType::Audio;
995
996            // Audio filter chain (`-af`); audio streams only.
997            let audio_chain = if is_audio {
998                rff_filter::AudioFilterChain::parse(output.audio_filters.as_deref().unwrap_or(""))?
999            } else {
1000                rff_filter::AudioFilterChain::default()
1001            };
1002
1003            // `-ac`: validate up front — only mono/stereo mixing exists.
1004            let target_channels = match output.audio_channels {
1005                Some(c) if is_audio => {
1006                    if !(1..=2).contains(&c) {
1007                        return Err(Error::unsupported(format!(
1008                            "-ac {c}: only 1 or 2 channels supported"
1009                        )));
1010                    }
1011                    c
1012                }
1013                _ => 0,
1014            };
1015
1016            // Output sample rate: `-ar` / `aresample=` wins, else the input
1017            // rate; either way the encoder's accepted-rate list has the last
1018            // word (nearest accepted).
1019            let mut target_rate = 0;
1020            let mut out_rate = stream.sample_rate;
1021            if is_audio && stream.sample_rate > 0 {
1022                let desired = output
1023                    .audio_rate
1024                    .or(audio_chain.resample_target())
1025                    .unwrap_or(stream.sample_rate);
1026                let desired = match encoder.accepted_sample_rates() {
1027                    Some(rates) if !rates.contains(&desired) => nearest_rate(&rates, desired),
1028                    _ => desired,
1029                };
1030                if desired != stream.sample_rate {
1031                    target_rate = desired;
1032                }
1033                out_rate = desired;
1034            }
1035
1036            // Trim window and the tick shift that re-zeroes the output timeline.
1037            let trim = Trim::from_spec(output);
1038            let tb = stream.time_base;
1039            let shift_ticks = trim
1040                .map(|t| (t.start * tb.den as f64 / tb.num.max(1) as f64).round() as i64)
1041                .unwrap_or(0);
1042
1043            // `-r`: CFR stage; the output stream ticks in 1/fps units.
1044            let fps = match output.frame_rate {
1045                Some((num, den)) if is_video => Some(FpsConv::new(num, den)),
1046                _ => None,
1047            };
1048
1049            let mut accepted_formats = encoder.accepted_pixel_formats();
1050
1051            let mut os = Stream::new(out_index, target.codec);
1052            os.media_type = stream.media_type;
1053            os.time_base = stream.time_base;
1054            os.width = out_w;
1055            os.height = out_h;
1056            os.pixel_format = stream.pixel_format;
1057            // Raw output declares its pixel format in the container header,
1058            // which the muxer writes before a single frame has been seen. So the
1059            // format cannot be discovered later — pin it now and force frames to
1060            // match, or the header ends up describing data it does not have.
1061            // (A 4:4:4 JPEG decoded straight to planes under a `C420mpeg2`
1062            // header is silent corruption, not a mislabel.)
1063            if target.codec == CodecId::RawVideo && stream.media_type == MediaType::Video {
1064                let pinned = os.pixel_format.unwrap_or(PixelFormat::Yuv420p);
1065                os.pixel_format = Some(pinned);
1066                accepted_formats = Some(vec![pinned]);
1067            }
1068            // Colour range: a `format=` filter that converts to Y'CbCr decides
1069            // it; otherwise it rides through from the input. Decoders whose
1070            // native output is RGB (JPEG, PNG, GIF, WebP) are full-range by
1071            // definition, so an untagged input from one of those is Full, not
1072            // "unknown" — leaving it unspecified makes the muxer label the
1073            // samples limited-range and every reader then rescales them.
1074            // A colour conversion in the chain decides the output range;
1075            // otherwise it is whatever we reconciled the source to above.
1076            os.color_range = filters.output_color_range().unwrap_or(source_range);
1077            os.sample_rate = out_rate;
1078            os.channels = if target_channels > 0 {
1079                target_channels
1080            } else {
1081                stream.channels
1082            };
1083            // A CFR conversion re-times the stream: pts become slot indices.
1084            if let Some((num, den)) = output.frame_rate {
1085                if is_video {
1086                    os.time_base = rff_core::Rational::new(den as i32, num as i32);
1087                }
1088            }
1089            // A format pinned by the output codec NAME (`-c:a pcm_s16le`) wins:
1090            // it is an explicit request, and the muxer writes it into the
1091            // container header before any frame is seen. Otherwise compressed
1092            // decoders (AAC/Opus/Vorbis/FLAC/MP3) emit f32, so default to that
1093            // when the input doesn't declare a format.
1094            let target_sample_format = target.sample_format;
1095            os.sample_format = target_sample_format
1096                .or(stream.sample_format)
1097                .or(Some(SampleFormat::F32));
1098            // The muxer writes os.sample_format into the container header
1099            // BEFORE any frame is seen, so every frame must be conformed to
1100            // that committed format — decoders may legitimately emit either
1101            // layout (the FLAC decoder emits native s16 for 16-bit streams).
1102            let target_sample_format = target_sample_format.or(os.sample_format);
1103            // Audio encoders timestamp packets in per-channel samples, so the
1104            // output stream's time base is 1/sample_rate.
1105            if stream.media_type == MediaType::Audio && out_rate > 0 {
1106                os.time_base = rff_core::Rational::new(1, out_rate as i32);
1107            }
1108            Ok((
1109                StreamOp::Transcode(Box::new(TranscodeOp {
1110                    decoder,
1111                    encoder,
1112                    filters,
1113                    overlay: None,
1114                    in_time_base: stream.time_base,
1115                    is_video,
1116                    trim,
1117                    shift_ticks,
1118                    audio_clock_samples: 0,
1119                    fps,
1120                    audio_chain,
1121                    target_channels,
1122                    target_rate,
1123                    target_sample_format,
1124                    resampler: None,
1125                    accepted_formats,
1126                    pixel_converter: None,
1127                    source_range,
1128                    out_index,
1129                    max_video_frames: output.max_video_frames,
1130                    video_frames_sent: 0,
1131                    pending_pts: std::collections::VecDeque::new(),
1132                })),
1133                os,
1134            ))
1135        }
1136        // Stream copy: carry the same codec/packets through unchanged (the trim
1137        // window still applies, cutting video on keyframes).
1138        None => {
1139            let mut os = stream.clone();
1140            os.index = out_index;
1141            // `-c:s`: text subtitles share one packet contract, so choosing a
1142            // subtitle codec relabels the stream (SubRip ↔ WebVTT) in place.
1143            if stream.media_type == MediaType::Subtitle {
1144                if let Some(target) = output.subtitle_codec {
1145                    if target.media_type() != MediaType::Subtitle {
1146                        return Err(Error::unsupported(format!(
1147                            "-c:s: `{}` is not a subtitle codec",
1148                            target.name()
1149                        )));
1150                    }
1151                    os.codec_id = target;
1152                }
1153            }
1154            let trim = Trim::from_spec(output);
1155            let tb = stream.time_base;
1156            let to_ticks = |s: f64| (s * tb.den as f64 / tb.num.max(1) as f64).round() as i64;
1157            Ok((
1158                StreamOp::Copy(CopyOp {
1159                    out_index,
1160                    is_video: stream.media_type == MediaType::Video,
1161                    start_ticks: trim.map(|t| to_ticks(t.start)).unwrap_or(0),
1162                    end_ticks: trim.and_then(|t| t.end).map(to_ticks),
1163                    trimming: trim.is_some(),
1164                    started: false,
1165                }),
1166                os,
1167            ))
1168        }
1169    }
1170}
1171
1172/// Decode the first video frame of the overlay input and convert it to 4:2:0,
1173/// ready to composite onto the base's YUV frames.
1174fn decode_overlay_frame(
1175    engine: &Engine,
1176    demuxer: &mut dyn rff_format::Demuxer,
1177    streams: &[Stream],
1178) -> Result<VideoFrame> {
1179    let vidx = streams
1180        .iter()
1181        .position(|s| s.media_type == MediaType::Video)
1182        .ok_or_else(|| {
1183            Error::Option("filter_complex overlay: overlay input has no video".into())
1184        })?;
1185    let mut decoder = engine.codecs.find_decoder(streams[vidx].codec_id)?;
1186    decoder.configure(&codec_params(&streams[vidx]))?;
1187    let mut to_yuv = FilterChain::parse("format=yuv420p")?;
1188    let mut got_eof = false;
1189    loop {
1190        let frame = match demuxer.read_packet() {
1191            Ok(pkt) if pkt.stream_index as usize != vidx => continue,
1192            Ok(pkt) => {
1193                decoder.send_packet(&pkt)?;
1194                decoder.receive_frame()
1195            }
1196            Err(Error::Eof) if !got_eof => {
1197                got_eof = true;
1198                decoder.flush();
1199                decoder.receive_frame()
1200            }
1201            Err(e) => return Err(e),
1202        };
1203        match frame {
1204            Ok(Frame::Video(v)) => return to_yuv.apply(v),
1205            Ok(_) => continue,
1206            Err(Error::Again) if !got_eof => continue,
1207            Err(Error::Again) | Err(Error::Eof) => {
1208                return Err(Error::Option(
1209                    "filter_complex overlay: no decodable frame in the overlay input".into(),
1210                ))
1211            }
1212            Err(e) => return Err(e),
1213        }
1214    }
1215}
1216
1217/// Build the decoder configuration from a demuxed input stream.
1218fn codec_params(s: &Stream) -> CodecParams {
1219    CodecParams {
1220        codec_id: s.codec_id,
1221        width: s.width,
1222        height: s.height,
1223        pixel_format: s.pixel_format,
1224        sample_rate: s.sample_rate,
1225        channels: s.channels,
1226        sample_format: s.sample_format,
1227        extradata: s.extradata.clone(),
1228    }
1229}
1230
1231/// Route one demuxed packet through its stream's plan.
1232fn process_packet(
1233    packet: Packet,
1234    ops: &mut [StreamOp],
1235    muxer: &mut dyn Muxer,
1236    report: &mut TranscodeReport,
1237) -> Result<()> {
1238    let Some(op) = ops.get_mut(packet.stream_index) else {
1239        return Ok(()); // packet for a stream we didn't plan — drop it
1240    };
1241    match op {
1242        StreamOp::Skip => Ok(()),
1243        StreamOp::Copy(copy) => {
1244            if let Some(packet) = copy.process(packet) {
1245                muxer.write_packet(&packet)?;
1246                report.packets_written += 1;
1247            }
1248            Ok(())
1249        }
1250        StreamOp::Transcode(op) => {
1251            op.decoder.send_packet(&packet)?;
1252            loop {
1253                match op.decoder.receive_frame() {
1254                    Ok(frame) => op.handle_frame(frame, muxer, report)?,
1255                    Err(Error::Again) | Err(Error::Eof) => break,
1256                    Err(e) => return Err(e),
1257                }
1258            }
1259            Ok(())
1260        }
1261    }
1262}
1263
1264/// At end of input, flush each transcoded stream's decoder, filter state, and
1265/// encoder, writing out anything still buffered.
1266fn flush_streams(
1267    ops: &mut [StreamOp],
1268    muxer: &mut dyn Muxer,
1269    report: &mut TranscodeReport,
1270) -> Result<()> {
1271    for op in ops.iter_mut() {
1272        if let StreamOp::Transcode(op) = op {
1273            op.finish(muxer, report)?;
1274        }
1275    }
1276    Ok(())
1277}
1278
1279/// Open an input as a streaming reader and decide its container format —
1280/// local file or `http://` URL. Delegates to the shared [`crate::probe`] opener
1281/// so ffmpeg and ffprobe resolve inputs identically.
1282fn open_input(engine: &Engine, input: &InputSpec) -> Result<(String, Box<dyn Read + Send>)> {
1283    let path = input
1284        .path
1285        .to_str()
1286        .ok_or_else(|| Error::Option("input path is not valid UTF-8".into()))?;
1287    crate::probe::open_source(engine, path, input.format.as_deref())
1288}
1289
1290/// Decide which container to mux an output as: explicit `-f`, else by
1291/// extension. An `rtmp://` target implies FLV — that is the payload RTMP
1292/// carries (matching FFmpeg's default for rtmp outputs).
1293fn resolve_output_format(engine: &Engine, output: &OutputSpec) -> Result<String> {
1294    if let Some(forced) = &output.format {
1295        return Ok(forced.clone());
1296    }
1297    if output
1298        .path
1299        .to_str()
1300        .is_some_and(rff_io::is_rtmp)
1301    {
1302        return Ok("flv".to_string());
1303    }
1304    let ext = output
1305        .path
1306        .extension()
1307        .and_then(|e| e.to_str())
1308        .unwrap_or_default();
1309    engine
1310        .formats
1311        .by_extension(ext)
1312        .map(|f| f.name.to_string())
1313        .ok_or_else(|| {
1314            Error::MuxerNotFound(format!(
1315                "{} (no format for this extension — pass -f)",
1316                output.path.display()
1317            ))
1318        })
1319}
1320
1321#[cfg(test)]
1322mod sample_format_tests {
1323    use super::*;
1324
1325    fn f32_af(samples: &[f32]) -> Frame {
1326        f32_frame(samples.to_vec(), 44_100, 1, Some(0))
1327    }
1328
1329    /// `-c:a pcm_s16le` used to be silently ignored on decode: the codec NAME
1330    /// carries the format but `CodecId` does not, so raw output kept whatever
1331    /// the decoder emitted (f32), writing double the bytes under an s16 request.
1332    #[test]
1333    fn pinned_format_converts_f32_to_s16() {
1334        let src = [0.0f32, 0.5, -0.5, 1.0, -1.0];
1335        let out = conform_sample_format(Some(SampleFormat::S16), f32_af(&src)).unwrap();
1336        let Frame::Audio(af) = out else { unreachable!() };
1337        assert_eq!(af.format, SampleFormat::S16);
1338        assert_eq!(af.planes[0].len(), src.len() * 2, "s16 is 2 bytes/sample");
1339        let got: Vec<i16> = af.planes[0]
1340            .chunks_exact(2)
1341            .map(|b| i16::from_le_bytes([b[0], b[1]]))
1342            .collect();
1343        // Round-to-nearest, and full scale must CLAMP rather than wrap: 1.0 maps
1344        // to 32767, not to -32768 via overflow.
1345        assert_eq!(got, vec![0, 16384, -16384, 32767, -32768]);
1346    }
1347
1348    /// No pin, or a frame already in the target format, must pass through
1349    /// untouched — the conversion is not allowed to cost anything when idle.
1350    #[test]
1351    fn unpinned_or_matching_format_is_a_passthrough() {
1352        let src = [0.25f32, -0.25];
1353        let Frame::Audio(a) = conform_sample_format(None, f32_af(&src)).unwrap() else {
1354            unreachable!()
1355        };
1356        assert_eq!(a.format, SampleFormat::F32);
1357        let Frame::Audio(b) =
1358            conform_sample_format(Some(SampleFormat::F32), f32_af(&src)).unwrap()
1359        else {
1360            unreachable!()
1361        };
1362        assert_eq!(b.format, SampleFormat::F32);
1363        assert_eq!(b.planes[0], a.planes[0]);
1364    }
1365
1366    /// The codec name is the only place the format lives for raw PCM.
1367    #[test]
1368    fn codec_name_pins_the_sample_format() {
1369        assert_eq!(
1370            CodecId::sample_format_from_name("pcm_s16le"),
1371            Some(SampleFormat::S16)
1372        );
1373        assert_eq!(
1374            CodecId::sample_format_from_name("pcm_f32le"),
1375            Some(SampleFormat::F32)
1376        );
1377        // Bare `pcm` pins nothing, and compressed codecs never do.
1378        assert_eq!(CodecId::sample_format_from_name("pcm"), None);
1379        assert_eq!(CodecId::sample_format_from_name("mp3"), None);
1380    }
1381}