def write_validation_mp4(
output_path: str,
frames: list[np.ndarray],
*,
fps: int,
audio: torch.Tensor | np.ndarray | None = None,
audio_sample_rate: int | None = None,
) -> None:
"""Write validation frames and optional synchronized audio atomically.
The video input must contain RGB uint8 frames. Audio can use
``[samples]``, ``[samples, channels]``, or ``[channels, samples]`` layout.
When both streams are present, the encoder trims them to their shortest
complete shared duration. The destination changes only after PyAV encodes
and verifies every requested stream.
"""
if not isinstance(fps, int) or fps <= 0:
raise ValueError(f"Validation video FPS must be a positive integer; got {fps!r}.")
if (audio is None) != (audio_sample_rate is None):
raise ValueError("Validation audio and its sample rate must be provided together.")
if audio_sample_rate is not None and audio_sample_rate <= 0:
raise ValueError("Validation audio sample rate must be positive; "
f"got {audio_sample_rate}.")
frame_array = _normalize_validation_frames(frames)
waveform = _normalize_audio_waveform(audio) if audio is not None else None
if waveform is not None and audio_sample_rate is not None:
frame_array, waveform = _trim_to_shared_duration(
frame_array,
fps,
waveform,
audio_sample_rate,
)
destination = os.path.abspath(output_path)
destination_dir = os.path.dirname(destination)
if not os.path.isdir(destination_dir):
raise FileNotFoundError(f"Validation media destination directory does not exist: {destination_dir}")
# A temporary file in the destination directory keeps replacement atomic
# and leaves any previous artifact intact when encoding or verification fails.
file_descriptor, temporary_path = tempfile.mkstemp(
prefix=f".{os.path.basename(destination)}.",
suffix=".mp4",
dir=destination_dir,
)
os.close(file_descriptor)
try:
_encode_mp4(
temporary_path,
frame_array,
fps,
waveform,
audio_sample_rate,
)
_verify_encoded_streams(
temporary_path,
expected_audio_channels=(int(waveform.shape[1]) if waveform is not None else None),
expected_audio_sample_rate=audio_sample_rate,
)
# Publish only an MP4 whose requested stream contract was verified.
os.replace(temporary_path, destination)
finally:
with contextlib.suppress(FileNotFoundError):
os.remove(temporary_path)