diff --git a/crates/align-decode/src/ff.rs b/crates/align-decode/src/ff.rs index 5fb6455..1a6797e 100644 --- a/crates/align-decode/src/ff.rs +++ b/crates/align-decode/src/ff.rs @@ -77,8 +77,10 @@ pub fn ffmpeg_bin() -> Option { resolve_bin("ALIGN_FFMPEG", "ffmpeg") } -/// Header-only inspect. No sample data is read (audio-only invariant). -pub fn inspect(path: &Path) -> Result { +/// Format and stream headers as ffprobe JSON. Each ffprobe start analyses +/// the opening packets (about 0.3 s for 4:2:2 10-bit camera video), so +/// inspection asks once and parses this for audio and video alike. +fn probe_json(path: &Path) -> Result, DecodeError> { let bin = ffprobe_bin().ok_or_else(|| DecodeError::FfmpegMissing("ffprobe not found".into()))?; let out = command(bin) @@ -99,7 +101,12 @@ pub fn inspect(path: &Path) -> Result { path.display() ))); } - parse_ffprobe_json(&out.stdout) + Ok(out.stdout) +} + +/// Header-only inspect. No sample data is read (audio-only invariant). +pub fn inspect(path: &Path) -> Result { + parse_ffprobe_json(&probe_json(path)?) .ok_or_else(|| DecodeError::Incomplete(format!("ffprobe JSON for {}", path.display()))) } @@ -165,16 +172,22 @@ pub fn parse_ffprobe_json(json: &[u8]) -> Option { duration_seconds, audio_streams, has_video, - video: None, // filled by inspect_video() below, not the pure parser + video: None, // filled by inspect_full() from the same JSON }) } /// Header inspect + video timing walk. Pure-header fast path when the /// container has no video track. pub fn inspect_full(path: &Path) -> Result { - let mut report = inspect(path)?; + // Taken before the header is read: the cached timing walk includes the + // header's frame rate. + let revision = media_revision(path); + let json = probe_json(path)?; + let mut report = parse_ffprobe_json(&json) + .ok_or_else(|| DecodeError::Incomplete(format!("ffprobe JSON for {}", path.display())))?; if report.has_video { - report.video = inspect_video(path); + report.video = + parse_video_stream(&json).and_then(|header| video_probe(path, &header, revision)); } Ok(report) } @@ -186,7 +199,8 @@ struct FfVideoStream { timecode: Option, } -/// First video stream's header facts + optional `timecode` tag. +/// First video stream's header facts + optional `timecode` tag. MXF keeps +/// the timecode in the format tags, which need `-show_format`. fn parse_video_stream(json: &[u8]) -> Option { let v: serde_json::Value = serde_json::from_slice(json).ok()?; let stream = v.get("streams")?.as_array()?.iter().find(|s| { @@ -271,6 +285,43 @@ impl PacketTiming { /// the shared [`align_core::classify`]. Early-exits once variability is /// proven; constant-rate classification requires the full walk. pub fn video_timing(path: &Path) -> Result { + cached_video_timing(path, media_revision(path), |bin| { + let header = command(bin) + .args([ + "-v", + "error", + "-print_format", + "json", + "-show_streams", + "-select_streams", + "v:0", + ]) + .arg(path) + .output() + .map_err(DecodeError::Io)?; + Ok(parse_video_stream(&header.stdout) + .as_ref() + .and_then(header_frame_duration)) + }) +} + +/// Nominal rate from the header, snapped through the broadcast table. +fn header_frame_duration(header: &FfVideoStream) -> Option { + align_core::canonical_frame_duration(header.avg_fps?, None) +} + +fn media_revision(path: &Path) -> Option { + align_core::cache::media_revision(path).ok() +} + +/// The walk is cached per file revision; `frame_duration` runs only when +/// it is not. `revision` is the file's revision from before its header was +/// read, so a header and a walk of different versions are never cached. +fn cached_video_timing( + path: &Path, + revision: Option, + frame_duration: impl FnOnce(&Path) -> Result, DecodeError>, +) -> Result { let bin = ffprobe_bin().ok_or_else(|| DecodeError::FfmpegMissing("ffprobe not found".into()))?; let decoder = bin.canonicalize().ok().and_then(|bin| { @@ -280,18 +331,16 @@ pub fn video_timing(path: &Path) -> Result Result, ) -> Result { - use align_core::canonical_frame_duration; use std::io::BufRead; let mut child = command(bin) @@ -325,26 +374,6 @@ fn video_timing_uncached( .map_err(DecodeError::Io)?; let stdout = child.stdout.take().ok_or(DecodeError::UnsupportedOutput)?; - // Nominal rate + dimensions from the header call. - let header_json = command(bin) - .args([ - "-v", - "error", - "-print_format", - "json", - "-show_streams", - "-select_streams", - "v:0", - ]) - .arg(path) - .output() - .map_err(DecodeError::Io)?; - let header = parse_video_stream(&header_json.stdout); - let frame_duration = header - .as_ref() - .and_then(|h| h.avg_fps) - .and_then(|fps| canonical_frame_duration(fps, None)); - let mut timing = PacketTiming::default(); let read_result = (|| { for line in std::io::BufReader::new(stdout).lines() { @@ -382,32 +411,18 @@ fn video_timing_uncached( } /// Full video probe: header facts + timing walk + timecode tag. -/// `None` when the container has no usable video stream. -pub fn inspect_video(path: &Path) -> Option { - let header_json = command(ffprobe_bin()?) - .args([ - "-v", - "error", - "-print_format", - "json", - "-show_streams", - "-select_streams", - "v:0", - ]) - .arg(path) - .output() - .ok()?; - if !header_json.status.success() { - return None; - } - let header = parse_video_stream(&header_json.stdout)?; - let timing = video_timing(path).ok()?; +fn video_probe( + path: &Path, + header: &FfVideoStream, + revision: Option, +) -> Option { + let timing = cached_video_timing(path, revision, |_| Ok(header_frame_duration(header))).ok()?; Some(VideoProbe { width: header.width, height: header.height, frame_duration: timing.frame_duration, mode: timing.mode, - source_timecode: timecode_from_header(&header), + source_timecode: timecode_from_header(header), }) } @@ -416,7 +431,7 @@ pub fn inspect_video(path: &Path) -> Option { /// `1/30`), then the label validates as NDF/DF. `None` when the rate is /// unknown or the label is forbidden — never guessed. Pure (no binary). fn timecode_from_header(header: &FfVideoStream) -> Option { - let duration = align_core::canonical_frame_duration(header.avg_fps?, None)?; + let duration = header_frame_duration(header)?; align_core::SourceTimecode::from_label(header.timecode.as_deref()?, duration) } pub struct AudioPipe { @@ -877,4 +892,43 @@ mod tests { assert!((peak - 1.0).abs() < 0.01, "peak={peak}"); let _ = std::fs::remove_dir_all(&dir); } + + /// MXF stores its start timecode with the material package, which + /// ffprobe reports in the format tags rather than on the video stream. + #[test] + fn mxf_timecode_comes_from_the_format_tags() { + let (Some(ffmpeg), Some(_)) = (ffmpeg_bin(), ffprobe_bin()) else { + return; + }; + let lists = |kind: &str| { + command(&ffmpeg) + .args(["-hide_banner", kind]) + .output() + .map(|out| String::from_utf8_lossy(&out.stdout).into_owned()) + .unwrap_or_default() + }; + if !lists("-muxers").contains(" mxf ") || !lists("-encoders").contains(" mpeg2video ") { + eprintln!("SKIP: FFmpeg without MXF or MPEG-2 support"); + return; + } + let dir = tempfile::tempdir().unwrap(); + let clip = dir.path().join("A001C001.MXF"); + let ok = command(&ffmpeg) + .args(["-v", "error", "-y", "-f", "lavfi", "-i"]) + .arg("testsrc2=size=320x240:rate=25") + .args(["-f", "lavfi", "-i", "sine=r=48000", "-t", "1"]) + .args(["-c:v", "mpeg2video", "-c:a", "pcm_s16le"]) + .args(["-timecode", "10:20:30:12"]) + .arg(&clip) + .status() + .is_ok_and(|s| s.success()); + assert!(ok, "FFmpeg could not write the MXF fixture"); + let video = inspect_full(&clip).unwrap().video.expect("video probe"); + let timecode = video.source_timecode.expect("container timecode"); + assert_eq!(timecode.text, "10:20:30:12"); + assert_eq!( + video.frame_duration, + Some(align_core::MediaTime::new(1, 25)) + ); + } } diff --git a/crates/align-decode/src/pipeline.rs b/crates/align-decode/src/pipeline.rs index 2af4ab5..b326cd7 100644 --- a/crates/align-decode/src/pipeline.rs +++ b/crates/align-decode/src/pipeline.rs @@ -6,8 +6,8 @@ //! (via [`BackendWindowProvider`]) → graph solve → [`align_core::SyncResult`]. //! //! Important implementation choices: -//! - fingerprinting and refinement use four bounded workers with indexed -//! deterministic results; +//! - inspection, fingerprinting and refinement use four bounded workers with +//! indexed deterministic results; //! - ClipIDs hash `path\0duration-micros`, using the same units across backends; //! - spanned-file, BWF/iXML/container/LTC timecode, FCP 7 XML and FCPXML //! evidence join the same waveform-first graph without overriding a @@ -27,6 +27,12 @@ use sha2::{Digest, Sha256}; use crate::backend::{BackendKind, MediaBackend, default_backend}; use crate::provider::BackendWindowProvider; +/// Files inspected at once. Inspection waits on ffprobe processes and header +/// reads, and the first time it sees an MXF or MTS file it reads the whole +/// file to classify frame timing. The bound matches the fingerprint stage's +/// four concurrent readers of external media. +const INSPECT_WORKERS: usize = 4; + // ------------------------------------------------------------ types #[derive(Clone, Debug, PartialEq)] @@ -316,8 +322,9 @@ impl Pipeline { } let mut clips = Vec::new(); let mut warnings = Vec::new(); - for url in &urls { - match self.inspect_one(url) { + let inspected = self.inspect_all(&urls, None, &std::sync::atomic::AtomicBool::new(false)); + for (url, outcome) in urls.iter().zip(inspected) { + match outcome { Ok(clip) => clips.push(clip), Err(e) => warnings.push(SyncWarning { url: url.clone(), @@ -348,8 +355,8 @@ impl Pipeline { /// Full pipeline: media files/dirs (+ optional one timeline) → /// [`align_core::SyncResult`]. /// - /// Fingerprint decode and fine refine run on up to 4 workers with - /// deterministic indexed results; `progress` + /// Inspection, fingerprint decode and fine refine run on up to 4 workers + /// with deterministic indexed results; `progress` /// is shared across those threads, `cancel` is polled per clip, /// candidate and job. pub fn synchronize( @@ -398,30 +405,15 @@ impl Pipeline { // ---- inspect let mut clips = Vec::new(); let mut warnings = Vec::new(); - for (i, url) in urls.iter().enumerate() { - if cancelled(cancel) { - return Err(PipelineError::Cancelled); - } - if let Some(p) = &progress { - p(PipelineProgress::basic( - Phase::Inspect, - i, - urls.len(), - Some(url.clone()), - )); - } - match self.inspect_one(url) { + let inspected = self.inspect_all(&urls, progress, cancel); + // Files already being inspected finish (or fail) after a + // cancellation; it must not turn into a missing-media error. + if cancelled(cancel) { + return Err(PipelineError::Cancelled); + } + for (url, outcome) in urls.iter().zip(inspected) { + match outcome { Ok(clip) => { - if let Some(p) = &progress { - let mut event = PipelineProgress::basic( - Phase::Inspect, - clips.len(), - urls.len(), - Some(url.clone()), - ); - event.discovered = Some(clip.clone()); - p(event); - } // Variable video timing does not affect audio sync, // but constant-rate interchange needs a CFR transcode. if let Some(video) = &clip.video { @@ -440,6 +432,7 @@ impl Pipeline { } clips.push(clip); } + Err(PipelineError::Cancelled) => return Err(PipelineError::Cancelled), Err(e) => warnings.push(SyncWarning { url: url.clone(), message: e.to_string(), @@ -743,6 +736,38 @@ impl Pipeline { }) } + /// Inspects files on bounded workers. Outcomes keep input order, so clips + /// and warnings never depend on which file finishes first; progress + /// reports each file as it starts and as it finishes. + fn inspect_all( + &self, + urls: &[PathBuf], + progress: Option<&(dyn Fn(PipelineProgress) + Send + Sync)>, + cancel: &std::sync::atomic::AtomicBool, + ) -> Vec> { + // Counting and publishing under one lock keeps the completed count + // monotonic for listeners such as the progress bar. + let done = std::sync::Mutex::new(0_usize); + let publish = |finished: bool, url: &PathBuf, discovered: Option| { + let Some(p) = progress else { return }; + let mut done = done.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + *done += usize::from(finished); + let mut event = + PipelineProgress::basic(Phase::Inspect, *done, urls.len(), Some(url.clone())); + event.discovered = discovered; + p(event); + }; + align_core::par_map(urls, INSPECT_WORKERS, |url| { + if cancelled(cancel) { + return Err(PipelineError::Cancelled); + } + publish(false, url, None); + let outcome = self.inspect_one(url); + publish(true, url, outcome.as_ref().ok().cloned()); + outcome + }) + } + fn inspect_one(&self, url: &Path) -> Result { let probe = self.backend.inspect(url).map_err(PipelineError::Decode)?; let micros = (probe.duration_seconds * 1_000_000.0).round() as i64; @@ -1525,6 +1550,128 @@ mod tests { } } + /// Inspection that fails every file after `delay`; nothing else runs. + struct FailingInspect { + delay: std::time::Duration, + } + + impl MediaBackend for FailingInspect { + fn kind(&self) -> BackendKind { + BackendKind::Portable + } + + fn inspect(&self, _: &Path) -> Result { + std::thread::sleep(self.delay); + Err(crate::DecodeError::Incomplete("unreadable".into())) + } + + fn decode_mono_8k( + &self, + _: &Path, + _: AudioAnalysisSource, + _: &mut dyn FnMut(&[f32]) -> Result<(), crate::DecodeError>, + ) -> Result<(), crate::DecodeError> { + unreachable!() + } + + fn decode_window_16k( + &self, + _: &Path, + _: f64, + _: f64, + _: AudioAnalysisSource, + ) -> Result<(f64, Vec), crate::DecodeError> { + unreachable!() + } + + fn decode_native( + &self, + _: &Path, + _: usize, + _: Option<(f64, Option)>, + _: &mut dyn FnMut(crate::backend::NativeBlock) -> Result<(), crate::DecodeError>, + ) -> Result { + unreachable!() + } + + fn decode_native_i32( + &self, + _: &Path, + _: usize, + _: Option<(f64, Option)>, + _: &mut dyn FnMut(crate::backend::NativeBlockI32) -> Result<(), crate::DecodeError>, + ) -> Result { + unreachable!() + } + } + + fn failing_inspection(tag: &str, files: usize, delay_ms: u64) -> (PathBuf, Pipeline) { + let dir = fixture_dir(tag); + for index in 0..files { + std::fs::write(dir.join(format!("{index:03}.wav")), b"").unwrap(); + } + let pipe = Pipeline { + backend: Box::new(FailingInspect { + delay: std::time::Duration::from_millis(delay_ms), + }), + cache: FingerprintCache::with_backend(Some(dir.join(".cache")), "failing-inspect"), + }; + (dir, pipe) + } + + /// Cancelling while every worker is busy used to end as "no media" once + /// those files failed. + #[test] + fn cancelling_during_inspection_is_reported_as_cancelled() { + let (dir, pipe) = failing_inspection("cancel-inspection", INSPECT_WORKERS, 400); + let cancel = std::sync::atomic::AtomicBool::new(false); + let result = std::thread::scope(|scope| { + scope.spawn(|| { + std::thread::sleep(std::time::Duration::from_millis(100)); + cancel.store(true, std::sync::atomic::Ordering::Relaxed); + }); + pipe.synchronize( + &[PipelineInput::Media(dir.clone())], + &[], + &PipelineOptions::default(), + None, + &cancel, + ) + }); + assert!( + matches!(result, Err(PipelineError::Cancelled)), + "{result:?}" + ); + let _ = std::fs::remove_dir_all(&dir); + } + + /// Workers finish in any order; the completed count the progress bar + /// shows must still only grow. + #[test] + fn inspection_progress_never_goes_backwards() { + let (dir, pipe) = failing_inspection("inspection-progress", 400, 0); + let counts = std::sync::Mutex::new(Vec::new()); + let progress = |event: PipelineProgress| { + if event.phase == Phase::Inspect { + counts.lock().unwrap().push(event.completed); + } + }; + let _ = pipe.synchronize( + &[PipelineInput::Media(dir.clone())], + &[], + &PipelineOptions::default(), + Some(&progress), + &std::sync::atomic::AtomicBool::new(false), + ); + let counts = counts.into_inner().unwrap(); + assert!(counts.len() > 400); + assert!( + counts.windows(2).all(|pair| pair[0] <= pair[1]), + "{counts:?}" + ); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn fixed_project_copy_repairs_every_sequence_and_preserves_source() { let dir = fixture_dir("fixed-project-copy");