use anyhow::{Context, Result};
use codec::frame::{ColorMetadata, VideoCodec};
use std::fs::{self, File};
use std::io::{BufWriter, Write};
use std::path::{Path, PathBuf};
use crate::AudioInfo;
use crate::mux::{build_avc1, build_avcc, build_hvc1, build_hvcc, extract_sequence_header};
use crate::nal_mux::{NalMuxCodec, NalSampleWriter};
mod fragment;
mod init;
#[cfg(test)]
mod tests;
pub use fragment::*;
pub use init::*;
pub mod brand {
pub const CMFC: &[u8; 4] = b"cmfc";
pub const CMFA: &[u8; 4] = b"cmfa";
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[allow(dead_code)]
pub enum CmafTrackKind {
Video,
Audio,
}
#[derive(Debug, Clone, Copy)]
pub struct SampleFlags {
pub is_sync: bool,
}
impl SampleFlags {
pub fn pack(self) -> u32 {
if self.is_sync {
0x02_00_00_00
} else {
0x01_01_00_00
}
}
pub fn keyframe() -> Self {
Self { is_sync: true }
}
pub fn delta_frame() -> Self {
Self { is_sync: false }
}
}
#[derive(Debug, Clone, Copy)]
pub struct CmafSample {
pub duration: u32,
pub size: u32,
pub flags: SampleFlags,
}
#[derive(Debug, Clone)]
pub struct SegmentInfo {
pub sequence_number: u32,
pub path: PathBuf,
pub byte_size: u64,
pub duration_ticks: u64,
}
#[derive(Debug, Clone)]
pub struct CmafTrackManifest {
pub init_path: PathBuf,
pub segments: Vec<SegmentInfo>,
pub timescale: u32,
}
impl CmafTrackManifest {
pub fn duration_seconds(&self) -> f64 {
let total_ticks: u64 = self.segments.iter().map(|s| s.duration_ticks).sum();
total_ticks as f64 / self.timescale as f64
}
}
struct PendingVideoSample {
payload: Vec<u8>,
duration: u32,
is_keyframe: bool,
}
struct PendingAudioSample {
payload: Vec<u8>,
duration: u32,
}
#[derive(Debug, Clone)]
pub struct CmafVideoMuxerOptions {
pub first_segment_index: u32,
pub first_segment_base_decode_time: u64,
pub write_init_segment: bool,
}
impl Default for CmafVideoMuxerOptions {
fn default() -> Self {
Self {
first_segment_index: 1,
first_segment_base_decode_time: 0,
write_init_segment: true,
}
}
}
pub struct CmafVideoMuxer {
output_dir: PathBuf,
width: u32,
height: u32,
timescale: u32,
color_metadata: ColorMetadata,
track_id: u32,
codec: VideoCodec,
config_obus: Option<Vec<u8>>,
nal_writer: Option<NalSampleWriter>,
init_path: PathBuf,
init_written: bool,
sequence_number: u32,
base_decode_time: u64,
pending: Vec<PendingVideoSample>,
segments: Vec<SegmentInfo>,
}
impl CmafVideoMuxer {
pub fn new(
output_dir: impl AsRef<Path>,
width: u32,
height: u32,
timescale: u32,
color_metadata: ColorMetadata,
) -> Result<Self> {
Self::new_with_options(
output_dir,
width,
height,
timescale,
color_metadata,
CmafVideoMuxerOptions::default(),
)
}
pub fn new_with_options(
output_dir: impl AsRef<Path>,
width: u32,
height: u32,
timescale: u32,
color_metadata: ColorMetadata,
options: CmafVideoMuxerOptions,
) -> Result<Self> {
Self::new_with_codec_options(
output_dir,
width,
height,
timescale,
color_metadata,
VideoCodec::Av1,
options,
)
}
pub fn new_with_codec_options(
output_dir: impl AsRef<Path>,
width: u32,
height: u32,
timescale: u32,
color_metadata: ColorMetadata,
codec: VideoCodec,
options: CmafVideoMuxerOptions,
) -> Result<Self> {
assert!(
options.first_segment_index >= 1,
"first_segment_index is 1-based; got {}",
options.first_segment_index,
);
let output_dir = output_dir.as_ref().to_path_buf();
fs::create_dir_all(&output_dir)
.with_context(|| format!("creating CMAF video output dir: {}", output_dir.display()))?;
let init_path = output_dir.join("init.mp4");
let nal_writer = match codec {
VideoCodec::Av1 => None,
VideoCodec::H264 => Some(NalSampleWriter::new_inline(NalMuxCodec::H264)),
VideoCodec::H265 => Some(NalSampleWriter::new_inline(NalMuxCodec::H265)),
};
Ok(Self {
output_dir,
width,
height,
timescale,
color_metadata,
track_id: 1,
codec,
config_obus: None,
nal_writer,
init_path,
init_written: !options.write_init_segment,
sequence_number: options.first_segment_index - 1,
base_decode_time: options.first_segment_base_decode_time,
pending: Vec::new(),
segments: Vec::new(),
})
}
pub fn add_packet(&mut self, payload: Vec<u8>, duration: u32, is_keyframe: bool) -> Result<()> {
match &mut self.nal_writer {
None => {
if self.config_obus.is_none() {
self.config_obus = Some(extract_sequence_header(&payload).context(
"extracting AV1 sequence header from first packet for av1C config record",
)?);
}
self.pending.push(PendingVideoSample {
payload,
duration,
is_keyframe,
});
}
Some(writer) => {
for au in writer.push_packet(&payload) {
self.pending.push(PendingVideoSample {
payload: au.data,
duration,
is_keyframe: au.is_keyframe,
});
}
}
}
Ok(())
}
pub fn first_pending_is_keyframe(&self) -> bool {
self.pending.first().is_some_and(|s| s.is_keyframe)
}
pub fn pending_duration_ticks(&self) -> u64 {
self.pending.iter().map(|s| s.duration as u64).sum()
}
pub fn segments(&self) -> &[SegmentInfo] {
&self.segments
}
pub fn clear_pending(&mut self) {
self.pending.clear();
}
pub fn flush_segment(&mut self) -> Result<Option<SegmentInfo>> {
if self.pending.is_empty() {
return Ok(None);
}
if !self.first_pending_is_keyframe() {
anyhow::bail!(
"CMAF segment must start with a sync sample; first pending sample is not a keyframe \
(segment_number={}, pending_count={})",
self.sequence_number + 1,
self.pending.len()
);
}
self.ensure_init_written()?;
self.sequence_number += 1;
let seq = self.sequence_number;
let samples_meta: Vec<CmafSample> = self
.pending
.iter()
.map(|s| CmafSample {
duration: s.duration,
size: s.payload.len() as u32,
flags: if s.is_keyframe {
SampleFlags::keyframe()
} else {
SampleFlags::delta_frame()
},
})
.collect();
let segment_duration: u64 = samples_meta.iter().map(|s| s.duration as u64).sum();
let mut moof = build_moof_video(seq, self.track_id, self.base_decode_time, &samples_meta);
moof.patch_default_no_gap();
let payload_total: u64 = self.pending.iter().map(|s| s.payload.len() as u64).sum();
let mdat_box_size: u64 = 8 + payload_total;
if mdat_box_size > u32::MAX as u64 {
anyhow::bail!(
"CMAF media segment payload {} bytes exceeds 32-bit mdat size limit",
payload_total
);
}
let path = self.output_dir.join(format!("seg-{:05}.m4s", seq));
let file = File::create(&path)
.with_context(|| format!("creating CMAF segment file: {}", path.display()))?;
let mut writer = BufWriter::new(file);
writer.write_all(&moof.bytes).context("writing moof")?;
writer
.write_all(&(mdat_box_size as u32).to_be_bytes())
.context("writing mdat size")?;
writer.write_all(b"mdat").context("writing mdat type")?;
for sample in &self.pending {
writer
.write_all(&sample.payload)
.context("writing mdat payload")?;
}
writer.flush().context("flushing CMAF segment writer")?;
let byte_size = moof.bytes.len() as u64 + mdat_box_size;
self.base_decode_time += segment_duration;
self.pending.clear();
let info = SegmentInfo {
sequence_number: seq,
path,
byte_size,
duration_ticks: segment_duration,
};
self.segments.push(info.clone());
Ok(Some(info))
}
pub fn finalize(mut self) -> Result<CmafTrackManifest> {
if !self.pending.is_empty() {
self.flush_segment()?;
}
self.ensure_init_written()?;
Ok(CmafTrackManifest {
init_path: self.init_path,
segments: self.segments,
timescale: self.timescale,
})
}
fn ensure_init_written(&mut self) -> Result<()> {
if self.init_written {
return Ok(());
}
let init = match self.codec {
VideoCodec::Av1 => {
let config = self.config_obus.as_ref().ok_or_else(|| {
anyhow::anyhow!(
"cannot write CMAF video init segment: no AV1 sequence header has been \
observed yet (call add_packet before flush_segment / finalize)"
)
})?;
build_init_segment_video(
self.width,
self.height,
self.timescale,
config,
&self.color_metadata,
)
}
VideoCodec::H264 => {
let w = self.nal_writer.as_ref().context("H.264 CMAF nal writer missing")?;
if !w.has_param_sets() {
anyhow::bail!("cannot write CMAF H.264 init segment: no SPS/PPS observed yet");
}
let avcc = build_avcc(&w.sps, &w.pps);
let entry = build_avc1(self.width, self.height, &avcc, &self.color_metadata, b"avc3");
build_init_segment_video_with_entry(
self.width,
self.height,
self.timescale,
&entry,
b"avc1",
)
}
VideoCodec::H265 => {
let w = self.nal_writer.as_ref().context("H.265 CMAF nal writer missing")?;
if !w.has_param_sets() {
anyhow::bail!(
"cannot write CMAF H.265 init segment: no VPS/SPS/PPS observed yet"
);
}
let hvcc = build_hvcc(&w.vps, &w.sps, &w.pps);
let entry = build_hvc1(self.width, self.height, &hvcc, &self.color_metadata, b"hev1");
build_init_segment_video_with_entry(
self.width,
self.height,
self.timescale,
&entry,
b"hvc1",
)
}
};
let mut file = File::create(&self.init_path).with_context(|| {
format!(
"creating CMAF video init segment: {}",
self.init_path.display()
)
})?;
file.write_all(&init)
.context("writing CMAF video init segment bytes")?;
file.flush().context("flushing CMAF video init segment")?;
self.init_written = true;
Ok(())
}
}
pub struct CmafAudioMuxer {
output_dir: PathBuf,
info: AudioInfo,
track_id: u32,
init_path: PathBuf,
init_written: bool,
sequence_number: u32,
base_decode_time: u64,
pending: Vec<PendingAudioSample>,
segments: Vec<SegmentInfo>,
}
impl CmafAudioMuxer {
pub fn new(output_dir: impl AsRef<Path>, info: AudioInfo) -> Result<Self> {
let output_dir = output_dir.as_ref().to_path_buf();
fs::create_dir_all(&output_dir)
.with_context(|| format!("creating CMAF audio output dir: {}", output_dir.display()))?;
let init_path = output_dir.join("init.mp4");
Ok(Self {
output_dir,
info,
track_id: 1,
init_path,
init_written: false,
sequence_number: 0,
base_decode_time: 0,
pending: Vec::new(),
segments: Vec::new(),
})
}
pub fn add_packet(&mut self, payload: Vec<u8>, duration: u32) -> Result<()> {
self.pending.push(PendingAudioSample { payload, duration });
Ok(())
}
pub fn pending_duration_ticks(&self) -> u64 {
self.pending.iter().map(|s| s.duration as u64).sum()
}
pub fn flush_segment(&mut self) -> Result<Option<SegmentInfo>> {
if self.pending.is_empty() {
return Ok(None);
}
self.ensure_init_written()?;
self.sequence_number += 1;
let seq = self.sequence_number;
let samples_meta: Vec<CmafSample> = self
.pending
.iter()
.map(|s| CmafSample {
duration: s.duration,
size: s.payload.len() as u32,
flags: SampleFlags::keyframe(),
})
.collect();
let segment_duration: u64 = samples_meta.iter().map(|s| s.duration as u64).sum();
let mut moof = build_moof_audio(seq, self.track_id, self.base_decode_time, &samples_meta);
moof.patch_default_no_gap();
let payload_total: u64 = self.pending.iter().map(|s| s.payload.len() as u64).sum();
let mdat_box_size: u64 = 8 + payload_total;
if mdat_box_size > u32::MAX as u64 {
anyhow::bail!(
"CMAF audio media segment payload {} bytes exceeds 32-bit mdat size limit",
payload_total
);
}
let path = self.output_dir.join(format!("seg-{:05}.m4s", seq));
let file = File::create(&path)
.with_context(|| format!("creating CMAF audio segment file: {}", path.display()))?;
let mut writer = BufWriter::new(file);
writer
.write_all(&moof.bytes)
.context("writing audio moof")?;
writer
.write_all(&(mdat_box_size as u32).to_be_bytes())
.context("writing audio mdat size")?;
writer
.write_all(b"mdat")
.context("writing audio mdat type")?;
for sample in &self.pending {
writer
.write_all(&sample.payload)
.context("writing audio mdat payload")?;
}
writer
.flush()
.context("flushing CMAF audio segment writer")?;
let byte_size = moof.bytes.len() as u64 + mdat_box_size;
self.base_decode_time += segment_duration;
self.pending.clear();
let info = SegmentInfo {
sequence_number: seq,
path,
byte_size,
duration_ticks: segment_duration,
};
self.segments.push(info.clone());
Ok(Some(info))
}
pub fn finalize(mut self) -> Result<CmafTrackManifest> {
if !self.pending.is_empty() {
self.flush_segment()?;
}
self.ensure_init_written()?;
let timescale = self.info.timescale;
Ok(CmafTrackManifest {
init_path: self.init_path,
segments: self.segments,
timescale,
})
}
fn ensure_init_written(&mut self) -> Result<()> {
if self.init_written {
return Ok(());
}
let init = build_init_segment_audio(&self.info);
let mut file = File::create(&self.init_path).with_context(|| {
format!(
"creating CMAF audio init segment: {}",
self.init_path.display()
)
})?;
file.write_all(&init)
.context("writing CMAF audio init segment bytes")?;
file.flush().context("flushing CMAF audio init segment")?;
self.init_written = true;
Ok(())
}
}