use anyhow::{Context, Result, bail};
use codec::frame::{ColorSpace, PixelFormat, StreamInfo};
use crate::ac3_sync::{
self, Eac3SyncInfo, SyncInfo, ac3_bit_rate_kbps, channel_count, eac3_sample_rate_hz,
eac3_samples_per_frame,
};
use crate::demux::{AudioTrack, DemuxResult};
use crate::mux::{dac3_body_from_sync, dec3_body_from_sync};
use crate::streaming::{DemuxHeader, Sample, StreamingDemuxer};
const TS_PACKET: usize = 188;
const TS_SYNC: u8 = 0x47;
const STREAM_TYPE_MPEG2_VIDEO: u8 = 0x02;
const STREAM_TYPE_H264: u8 = 0x1B;
const STREAM_TYPE_HEVC: u8 = 0x24;
const STREAM_TYPE_PES_PRIVATE: u8 = 0x06;
const STREAM_TYPE_AAC_ADTS: u8 = 0x0F;
const STREAM_TYPE_AC3: u8 = 0x81;
const STREAM_TYPE_EAC3: u8 = 0x87;
const DESC_TAG_REGISTRATION: u8 = 0x05;
const REG_AC3: u32 = 0x41432D33; const REG_EAC3: u32 = 0x45414333;
pub(crate) fn demux_ts(data: &[u8]) -> Result<DemuxResult> {
let (packets, packet_stride, prefix_len) = detect_packet_layout(data)?;
if packets == 0 {
bail!("TS: file contains no TS packets");
}
let mut pmt_pid: Option<u16> = None;
let mut chosen_video: Option<VideoStreamInfo> = None;
let mut chosen_audio: Option<AudioStreamInfo> = None;
for i in 0..packets {
let start = i * packet_stride + prefix_len;
let pkt = &data[start..start + TS_PACKET];
if pkt[0] != TS_SYNC {
continue;
}
let pid = (((pkt[1] & 0x1F) as u16) << 8) | pkt[2] as u16;
if pmt_pid.is_none() && pid == 0 {
if let Some(payload) = ts_psi_payload(pkt)
&& let Some(p) = parse_pat_first_pmt_pid(payload)
{
pmt_pid = Some(p);
}
continue;
}
if let (Some(pmt), None) = (pmt_pid, chosen_video)
&& pid == pmt
&& let Some(payload) = ts_psi_payload(pkt)
&& let Some((video_streams, audio_streams)) = parse_pmt_streams(payload)
{
chosen_video = video_streams.into_iter().next();
chosen_audio = audio_streams.into_iter().next();
if chosen_video.is_some() {
break;
}
}
}
let video = chosen_video.context("TS: no video elementary stream found in PMT")?;
let video_pid = video.pid;
let codec = match video.stream_type {
STREAM_TYPE_MPEG2_VIDEO => "mpeg2",
STREAM_TYPE_H264 => "h264",
STREAM_TYPE_HEVC => "h265",
other => bail!("TS: unsupported stream_type 0x{:02X}", other),
}
.to_string();
let mut samples: Vec<Vec<u8>> = Vec::new();
let mut pending: Vec<u8> = Vec::new();
let mut have_first_start = false;
let mut first_pts: Option<u64> = None;
let mut last_pts: Option<u64> = None;
let mut ptses: Vec<u64> = Vec::new();
let flush = |pending: &mut Vec<u8>, samples: &mut Vec<Vec<u8>>| {
if !pending.is_empty() {
samples.push(std::mem::take(pending));
}
};
for i in 0..packets {
let start = i * packet_stride + prefix_len;
let pkt = &data[start..start + TS_PACKET];
if pkt[0] != TS_SYNC {
continue;
}
let pid = (((pkt[1] & 0x1F) as u16) << 8) | pkt[2] as u16;
if pid != video_pid {
continue;
}
let pusi = pkt[1] & 0x40 != 0;
let scramble = (pkt[3] >> 6) & 0x03;
if scramble != 0 {
continue;
} let adaptation = (pkt[3] >> 4) & 0x03;
let has_payload = adaptation & 0x01 != 0;
let has_adaptation = adaptation & 0x02 != 0;
if !has_payload {
continue;
}
let mut offset = 4usize;
if has_adaptation {
if offset >= TS_PACKET {
continue;
}
let adap_len = pkt[offset] as usize;
offset += 1 + adap_len;
if offset > TS_PACKET {
continue;
}
}
if offset >= TS_PACKET {
continue;
}
let payload = &pkt[offset..];
if pusi {
if have_first_start {
flush(&mut pending, &mut samples);
}
have_first_start = true;
let Some((es_start, pts)) = parse_pes_header(payload) else {
have_first_start = false;
pending.clear();
continue;
};
if let Some(p) = pts {
if first_pts.is_none() {
first_pts = Some(p);
}
last_pts = Some(p);
ptses.push(p);
}
if es_start < payload.len() {
pending.extend_from_slice(&payload[es_start..]);
}
} else if have_first_start {
pending.extend_from_slice(payload);
}
}
flush(&mut pending, &mut samples);
if samples.is_empty() {
bail!("TS: reassembled zero video samples from PID {}", video_pid);
}
let duration = match (first_pts, last_pts) {
(Some(a), Some(b)) if b >= a => (b - a) as f64 / 90_000.0,
_ => 0.0,
};
let frame_rate = estimate_frame_rate_from_ptses(&ptses)
.or_else(|| {
if duration > 0.0 && samples.len() > 1 {
Some((samples.len() - 1) as f64 / duration)
} else {
None
}
})
.unwrap_or(30.0);
let (width, height) = codec::pixel_format::detect_dims(&codec, &samples).unwrap_or((0, 0));
if width == 0 || height == 0 {
tracing::warn!(
codec = codec.as_str(),
"TS demux: could not recover width/height from first sample — \
downstream encoder may reject the 0×0 config"
);
}
let info = StreamInfo {
codec: codec.clone(),
width,
height,
frame_rate,
duration,
pixel_format: PixelFormat::Yuv420p,
color_space: ColorSpace::Bt709,
total_frames: samples.len() as u64,
bitrate: 0,
color_metadata: Default::default(),
};
let detected_pf = codec::pixel_format::detect(&codec, &samples);
let info = StreamInfo {
pixel_format: detected_pf,
..info
};
let audio = chosen_audio.and_then(|info| {
match extract_ts_audio(data, packets, packet_stride, prefix_len, info) {
Ok(track) => track,
Err(e) => {
tracing::warn!(
audio_pid = info.pid,
audio_kind = ?info.kind,
error = %e,
"TS audio extraction failed; emitting video-only"
);
None
}
}
});
Ok(DemuxResult {
codec,
info,
samples,
audio,
})
}
fn detect_packet_layout(data: &[u8]) -> Result<(usize, usize, usize)> {
if data.len() < TS_PACKET {
bail!("TS: file too small");
}
if data[0] == TS_SYNC && data.len() >= 2 * TS_PACKET && data[TS_PACKET] == TS_SYNC {
return Ok((data.len() / TS_PACKET, TS_PACKET, 0));
}
if data.len() >= 192 + 4 && data[4] == TS_SYNC && data[196] == TS_SYNC {
return Ok((data.len() / 192, 192, 4));
}
bail!("TS: could not locate 0x47 sync pattern at 188- or 192-byte intervals")
}
fn ts_psi_payload(pkt: &[u8]) -> Option<&[u8]> {
let pusi = pkt[1] & 0x40 != 0;
let adaptation = (pkt[3] >> 4) & 0x03;
let has_payload = adaptation & 0x01 != 0;
let has_adaptation = adaptation & 0x02 != 0;
if !has_payload {
return None;
}
let mut offset = 4usize;
if has_adaptation {
if offset >= TS_PACKET {
return None;
}
let adap_len = pkt[offset] as usize;
offset += 1 + adap_len;
if offset > TS_PACKET {
return None;
}
}
if pusi {
if offset >= TS_PACKET {
return None;
}
let pointer = pkt[offset] as usize;
offset += 1 + pointer;
if offset >= TS_PACKET {
return None;
}
}
Some(&pkt[offset..])
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct PatProgram {
pub program_number: u16,
pub pmt_pid: u16,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AudioCodecKind {
AacAdts,
Ac3,
Eac3,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct VideoStreamInfo {
pub pid: u16,
pub stream_type: u8,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct AudioStreamInfo {
pub pid: u16,
pub stream_type: u8,
pub kind: AudioCodecKind,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProgramInfo {
pub program_number: u16,
pub pmt_pid: u16,
pub video_streams: Vec<VideoStreamInfo>,
pub audio_streams: Vec<AudioStreamInfo>,
}
fn parse_pat_all_programs(section: &[u8]) -> Vec<PatProgram> {
let mut out = Vec::new();
if section.len() < 12 {
return out;
}
if section[0] != 0x00 {
return out;
}
let section_length = (((section[1] & 0x0F) as usize) << 8) | section[2] as usize;
let total = 3 + section_length;
if total > section.len() {
return out;
}
let loop_start = 8;
let loop_end = total - 4;
let mut i = loop_start;
while i + 4 <= loop_end {
let program = u16::from_be_bytes([section[i], section[i + 1]]);
let pid = (((section[i + 2] & 0x1F) as u16) << 8) | section[i + 3] as u16;
if program != 0 {
out.push(PatProgram {
program_number: program,
pmt_pid: pid,
});
}
i += 4;
}
out
}
fn parse_pat_first_pmt_pid(section: &[u8]) -> Option<u16> {
parse_pat_all_programs(section).first().map(|p| p.pmt_pid)
}
fn parse_pmt_streams(section: &[u8]) -> Option<(Vec<VideoStreamInfo>, Vec<AudioStreamInfo>)> {
if section.len() < 12 {
return None;
}
if section[0] != 0x02 {
return None;
}
let section_length = (((section[1] & 0x0F) as usize) << 8) | section[2] as usize;
let total = 3 + section_length;
if total > section.len() {
return None;
}
if section.len() < 12 {
return None;
}
let pil = (((section[10] & 0x0F) as usize) << 8) | section[11] as usize;
let mut i = 12 + pil;
let loop_end = total - 4; let mut video: Vec<VideoStreamInfo> = Vec::new();
let mut audio: Vec<AudioStreamInfo> = Vec::new();
while i + 5 <= loop_end {
let stype = section[i];
let pid = (((section[i + 1] & 0x1F) as u16) << 8) | section[i + 2] as u16;
let esi_len = (((section[i + 3] & 0x0F) as usize) << 8) | section[i + 4] as usize;
let desc_start = i + 5;
let desc_end = (desc_start + esi_len).min(loop_end);
let descriptors = if desc_start <= desc_end {
§ion[desc_start..desc_end]
} else {
&[][..]
};
match stype {
STREAM_TYPE_MPEG2_VIDEO | STREAM_TYPE_H264 | STREAM_TYPE_HEVC => {
video.push(VideoStreamInfo {
pid,
stream_type: stype,
});
}
STREAM_TYPE_AAC_ADTS => {
audio.push(AudioStreamInfo {
pid,
stream_type: stype,
kind: AudioCodecKind::AacAdts,
});
}
STREAM_TYPE_AC3 => {
audio.push(AudioStreamInfo {
pid,
stream_type: stype,
kind: AudioCodecKind::Ac3,
});
}
STREAM_TYPE_EAC3 => {
audio.push(AudioStreamInfo {
pid,
stream_type: stype,
kind: AudioCodecKind::Eac3,
});
}
STREAM_TYPE_PES_PRIVATE => {
if let Some(reg) = find_registration(descriptors) {
match reg {
REG_AC3 => audio.push(AudioStreamInfo {
pid,
stream_type: stype,
kind: AudioCodecKind::Ac3,
}),
REG_EAC3 => audio.push(AudioStreamInfo {
pid,
stream_type: stype,
kind: AudioCodecKind::Eac3,
}),
_ => {}
}
}
}
_ => {}
}
i += 5 + esi_len;
}
Some((video, audio))
}
fn find_registration(descriptors: &[u8]) -> Option<u32> {
let mut i = 0usize;
while i + 2 <= descriptors.len() {
let tag = descriptors[i];
let len = descriptors[i + 1] as usize;
let body_start = i + 2;
let body_end = body_start + len;
if body_end > descriptors.len() {
break;
}
if tag == DESC_TAG_REGISTRATION && len >= 4 {
let id = u32::from_be_bytes([
descriptors[body_start],
descriptors[body_start + 1],
descriptors[body_start + 2],
descriptors[body_start + 3],
]);
return Some(id);
}
i = body_end;
}
None
}
fn parse_pes_header(payload: &[u8]) -> Option<(usize, Option<u64>)> {
if payload.len() < 9 {
return None;
}
if payload[0] != 0 || payload[1] != 0 || payload[2] != 1 {
return None;
}
let stream_id = payload[3];
if !(0xE0..=0xEF).contains(&stream_id) {
return None;
}
let flags = payload[7];
let pts_dts_flags = (flags >> 6) & 0x03;
let header_data_len = payload[8] as usize;
let es_start = 9 + header_data_len;
if es_start > payload.len() {
return None;
}
let pts = if pts_dts_flags == 0b10 || pts_dts_flags == 0b11 {
if payload.len() < 14 {
return None;
}
let p0 = ((payload[9] >> 1) & 0x07) as u64;
let p1 = (((payload[10] as u64) << 7) | ((payload[11] as u64) >> 1)) & 0x7FFF;
let p2 = (((payload[12] as u64) << 7) | ((payload[13] as u64) >> 1)) & 0x7FFF;
Some((p0 << 30) | (p1 << 15) | p2)
} else {
None
};
Some((es_start, pts))
}
pub(crate) struct VideoStreamScan {
pub first_au: Option<Vec<u8>>,
pub ptses: Vec<u64>,
}
fn scan_first_video_au(
data: &[u8],
packets: usize,
packet_stride: usize,
prefix_len: usize,
video_pid: u16,
max_pts_samples: usize,
) -> VideoStreamScan {
let mut accumulator: Vec<u8> = Vec::new();
let mut first_au: Option<Vec<u8>> = None;
let mut ptses: Vec<u64> = Vec::new();
let mut au_started = false;
let mut au_done = false;
for i in 0..packets {
let start = i * packet_stride + prefix_len;
let pkt = &data[start..start + TS_PACKET];
if pkt[0] != TS_SYNC {
continue;
}
let pid = (((pkt[1] & 0x1F) as u16) << 8) | pkt[2] as u16;
if pid != video_pid {
continue;
}
let pusi = pkt[1] & 0x40 != 0;
let scramble = (pkt[3] >> 6) & 0x03;
if scramble != 0 {
continue;
} let adaptation = (pkt[3] >> 4) & 0x03;
let has_payload = adaptation & 0x01 != 0;
let has_adaptation = adaptation & 0x02 != 0;
if !has_payload {
continue;
}
let mut offset = 4usize;
if has_adaptation {
if offset >= TS_PACKET {
continue;
}
let adap_len = pkt[offset] as usize;
offset += 1 + adap_len;
if offset > TS_PACKET {
continue;
}
}
if offset >= TS_PACKET {
continue;
}
let payload = &pkt[offset..];
if pusi {
if au_started && !au_done {
first_au = Some(std::mem::take(&mut accumulator));
au_done = true;
}
if let Some((es_start, pts)) = parse_pes_header(payload) {
if let Some(p) = pts
&& ptses.len() < max_pts_samples
{
ptses.push(p);
}
if !au_done {
if es_start < payload.len() {
accumulator.extend_from_slice(&payload[es_start..]);
}
au_started = true;
}
}
} else if au_started && !au_done {
accumulator.extend_from_slice(payload);
}
if au_done && ptses.len() >= max_pts_samples {
break;
}
}
if first_au.is_none() && au_started && !accumulator.is_empty() {
first_au = Some(accumulator);
}
VideoStreamScan { first_au, ptses }
}
fn estimate_frame_rate_from_ptses(ptses: &[u64]) -> Option<f64> {
if ptses.len() < 2 {
return None;
}
let mut sorted: Vec<u64> = ptses.to_vec();
sorted.sort_unstable();
let mut deltas: Vec<u64> = sorted.windows(2).map(|w| w[1] - w[0]).collect();
deltas.retain(|&d| d > 0);
if deltas.is_empty() {
return None;
}
deltas.sort_unstable();
let median = deltas[deltas.len() / 2];
if median == 0 {
return None;
}
let fps = 90000.0 / median as f64;
if !fps.is_finite() || !(1.0..=240.0).contains(&fps) {
return None;
}
Some(fps)
}
const AAC_SAMPLE_RATES: [u32; 13] = [
96000, 88200, 64000, 48000, 44100, 32000, 24000, 22050, 16000, 12000, 11025, 8000, 7350,
];
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct AdtsHeader {
profile: u8,
sampling_frequency_index: u8,
channel_configuration: u8,
frame_length: usize,
header_len: usize,
}
fn parse_adts_header(buf: &[u8]) -> Option<AdtsHeader> {
if buf.len() < 7 {
return None;
}
if buf[0] != 0xFF || (buf[1] & 0xF0) != 0xF0 {
return None;
}
let protection_absent = (buf[1] & 0x01) != 0;
let header_len = if protection_absent { 7 } else { 9 };
if buf.len() < header_len {
return None;
}
let profile = (buf[2] >> 6) & 0x03;
let sampling_frequency_index = (buf[2] >> 2) & 0x0F;
let channel_configuration = ((buf[2] & 0x01) << 2) | ((buf[3] >> 6) & 0x03);
let frame_length =
(((buf[3] & 0x03) as usize) << 11) | ((buf[4] as usize) << 3) | ((buf[5] >> 5) as usize);
if frame_length < header_len {
return None;
}
Some(AdtsHeader {
profile,
sampling_frequency_index,
channel_configuration,
frame_length,
header_len,
})
}
fn decode_sample_rate_index(idx: u8) -> Option<u32> {
AAC_SAMPLE_RATES.get(idx as usize).copied()
}
fn synthesize_asc(adts: &AdtsHeader) -> [u8; 2] {
let aot = adts.profile + 1; let sr_idx = adts.sampling_frequency_index;
let ch_cfg = adts.channel_configuration;
let mut bits: u16 = 0;
bits |= ((aot as u16) & 0x1F) << 11;
bits |= ((sr_idx as u16) & 0x0F) << 7;
bits |= ((ch_cfg as u16) & 0x0F) << 3;
bits.to_be_bytes()
}
fn extract_ts_aac_audio(
data: &[u8],
packets: usize,
packet_stride: usize,
prefix_len: usize,
audio_pid: u16,
) -> Result<Option<AudioTrack>> {
let es = reassemble_audio_pes(data, packets, packet_stride, prefix_len, audio_pid);
if es.is_empty() {
return Ok(None);
}
let mut cursor = match find_adts_sync(&es, 0) {
Some(idx) => idx,
None => return Ok(None),
};
let first = parse_adts_header(&es[cursor..]).context("TS: first ADTS frame failed to parse")?;
let sample_rate = decode_sample_rate_index(first.sampling_frequency_index)
.context("TS: AAC sampling_frequency_index out of range")?;
let channels = first.channel_configuration as u16;
if channels == 0 {
bail!("TS: AAC channel_configuration=0 (PCE-defined); not supported");
}
let asc = synthesize_asc(&first).to_vec();
let mut samples: Vec<Vec<u8>> = Vec::new();
let mut durations: Vec<u32> = Vec::new();
while cursor < es.len() {
let Some(found) = find_adts_sync(&es, cursor) else {
break;
};
cursor = found;
let Some(hdr) = parse_adts_header(&es[cursor..]) else {
break;
};
if hdr.sampling_frequency_index != first.sampling_frequency_index
|| hdr.channel_configuration != first.channel_configuration
{
tracing::warn!(
"TS: AAC ADTS stream switched sr_idx/ch_cfg mid-stream; truncating audio at frame {}",
samples.len()
);
break;
}
let end = cursor + hdr.frame_length;
if end > es.len() {
break;
}
let payload_start = cursor + hdr.header_len;
if payload_start > end {
break;
}
samples.push(es[payload_start..end].to_vec());
durations.push(1024);
cursor = end;
}
if samples.is_empty() {
return Ok(None);
}
Ok(Some(AudioTrack {
codec: "aac".into(),
samples,
sample_rate,
channels,
asc,
codec_private: Vec::new(),
timescale: sample_rate,
durations,
}))
}
fn find_ac3_sync(es: &[u8], from: usize) -> Option<usize> {
let mut i = from;
while i + 1 < es.len() {
if es[i] == 0x0B && es[i + 1] == 0x77 {
return Some(i);
}
i += 1;
}
None
}
fn ac3_frame_size(brc: u8, fscod: u8, frmsizecod_low_bit: u8) -> Option<usize> {
let kbps = ac3_bit_rate_kbps(brc) as usize;
if kbps == 0 {
return None;
}
let sr = ac3_sync::ac3_sample_rate_hz(fscod) as usize;
if sr == 0 {
return None;
}
let base = (kbps * 1000 * 1536) / (sr * 8);
let extra = if fscod == 1 && frmsizecod_low_bit != 0 {
2
} else {
0
};
Some(base + extra)
}
fn eac3_frame_size(frmsiz: u16) -> usize {
((frmsiz as usize) + 1) * 2
}
fn extract_ts_ac3_audio(
data: &[u8],
packets: usize,
packet_stride: usize,
prefix_len: usize,
audio_pid: u16,
) -> Result<Option<AudioTrack>> {
let es = reassemble_audio_pes(data, packets, packet_stride, prefix_len, audio_pid);
if es.is_empty() {
return Ok(None);
}
let mut cursor = match find_ac3_sync(&es, 0) {
Some(idx) => idx,
None => return Ok(None),
};
let first = match ac3_sync::parse_sync_info(&es[cursor..])
.context("TS: first AC-3 frame failed to parse sync header")?
{
SyncInfo::Ac3(s) => s,
SyncInfo::Eac3(_) => bail!("TS: AC-3 PMT entry but bitstream is E-AC-3 (bsid=16)"),
};
let sample_rate = ac3_sync::ac3_sample_rate_hz(first.fscod);
if sample_rate == 0 {
bail!("TS: AC-3 fscod={} reserved", first.fscod);
}
let channels = channel_count(first.acmod, first.lfeon);
let dac3 = dac3_body_from_sync(&first).to_vec();
let mut samples: Vec<Vec<u8>> = Vec::new();
let mut durations: Vec<u32> = Vec::new();
while cursor < es.len() {
let Some(found) = find_ac3_sync(&es, cursor) else {
break;
};
cursor = found;
if cursor + 5 > es.len() {
break;
}
let frmsizecod = es[cursor + 4] & 0x3F;
let bit_rate_code = frmsizecod >> 1;
let low_bit = frmsizecod & 0x01;
let fscod = (es[cursor + 4] >> 6) & 0x03;
let Some(size) = ac3_frame_size(bit_rate_code, fscod, low_bit) else {
break;
};
let end = cursor + size;
if end > es.len() {
break;
}
samples.push(es[cursor..end].to_vec());
durations.push(1536);
cursor = end;
}
if samples.is_empty() {
return Ok(None);
}
Ok(Some(AudioTrack {
codec: "ac3".into(),
samples,
sample_rate,
channels,
asc: Vec::new(),
codec_private: dac3,
timescale: sample_rate,
durations,
}))
}
fn extract_ts_eac3_audio(
data: &[u8],
packets: usize,
packet_stride: usize,
prefix_len: usize,
audio_pid: u16,
) -> Result<Option<AudioTrack>> {
let es = reassemble_audio_pes(data, packets, packet_stride, prefix_len, audio_pid);
if es.is_empty() {
return Ok(None);
}
let mut cursor = match find_ac3_sync(&es, 0) {
Some(idx) => idx,
None => return Ok(None),
};
let first: Eac3SyncInfo = match ac3_sync::parse_sync_info(&es[cursor..])
.context("TS: first E-AC-3 frame failed to parse sync header")?
{
SyncInfo::Eac3(s) => s,
SyncInfo::Ac3(_) => bail!("TS: E-AC-3 PMT entry but bitstream is AC-3 (bsid<=10)"),
};
let sample_rate = eac3_sample_rate_hz(first.fscod, first.fscod2);
if sample_rate == 0 {
bail!(
"TS: E-AC-3 reserved sample rate (fscod={}, fscod2={})",
first.fscod,
first.fscod2
);
}
let channels = channel_count(first.acmod, first.lfeon);
let spf = eac3_samples_per_frame(first.numblkscod) as u64;
let frame_bytes = ((first.frmsiz as u64) + 1) * 2;
let bitrate_kbps = if spf > 0 && sample_rate > 0 {
(frame_bytes * 8 * sample_rate as u64) / spf / 1000
} else {
0
};
let data_rate = bitrate_kbps.div_ceil(2) as u16;
let dec3 = dec3_body_from_sync(&first, data_rate).to_vec();
let mut samples: Vec<Vec<u8>> = Vec::new();
let mut durations: Vec<u32> = Vec::new();
while cursor < es.len() {
let Some(found) = find_ac3_sync(&es, cursor) else {
break;
};
cursor = found;
if cursor + 5 > es.len() {
break;
}
let raw = u16::from_be_bytes([es[cursor + 2], es[cursor + 3]]);
let frmsiz = raw & 0x07FF;
let size = eac3_frame_size(frmsiz);
let end = cursor + size;
if end > es.len() {
break;
}
samples.push(es[cursor..end].to_vec());
durations.push(spf as u32);
cursor = end;
}
if samples.is_empty() {
return Ok(None);
}
Ok(Some(AudioTrack {
codec: "eac3".into(),
samples,
sample_rate,
channels,
asc: Vec::new(),
codec_private: dec3,
timescale: sample_rate,
durations,
}))
}
fn reassemble_audio_pes(
data: &[u8],
packets: usize,
packet_stride: usize,
prefix_len: usize,
audio_pid: u16,
) -> Vec<u8> {
let mut es: Vec<u8> = Vec::new();
let mut have_first_start = false;
for i in 0..packets {
let start = i * packet_stride + prefix_len;
let pkt = &data[start..start + TS_PACKET];
if pkt[0] != TS_SYNC {
continue;
}
let pid = (((pkt[1] & 0x1F) as u16) << 8) | pkt[2] as u16;
if pid != audio_pid {
continue;
}
let pusi = pkt[1] & 0x40 != 0;
let scramble = (pkt[3] >> 6) & 0x03;
if scramble != 0 {
continue;
}
let adaptation = (pkt[3] >> 4) & 0x03;
let has_payload = adaptation & 0x01 != 0;
let has_adaptation = adaptation & 0x02 != 0;
if !has_payload {
continue;
}
let mut offset = 4usize;
if has_adaptation {
if offset >= TS_PACKET {
continue;
}
let adap_len = pkt[offset] as usize;
offset += 1 + adap_len;
if offset > TS_PACKET {
continue;
}
}
if offset >= TS_PACKET {
continue;
}
let payload = &pkt[offset..];
if pusi {
let Some((es_start, _pts)) = parse_pes_header_audio(payload) else {
have_first_start = false;
continue;
};
have_first_start = true;
if es_start < payload.len() {
es.extend_from_slice(&payload[es_start..]);
}
} else if have_first_start {
es.extend_from_slice(payload);
}
}
es
}
fn extract_ts_audio(
data: &[u8],
packets: usize,
packet_stride: usize,
prefix_len: usize,
info: AudioStreamInfo,
) -> Result<Option<AudioTrack>> {
match info.kind {
AudioCodecKind::AacAdts => {
extract_ts_aac_audio(data, packets, packet_stride, prefix_len, info.pid)
}
AudioCodecKind::Ac3 => {
extract_ts_ac3_audio(data, packets, packet_stride, prefix_len, info.pid)
}
AudioCodecKind::Eac3 => {
extract_ts_eac3_audio(data, packets, packet_stride, prefix_len, info.pid)
}
}
}
fn find_adts_sync(es: &[u8], from: usize) -> Option<usize> {
let mut i = from;
while i + 1 < es.len() {
if es[i] == 0xFF && (es[i + 1] & 0xF0) == 0xF0 {
return Some(i);
}
i += 1;
}
None
}
fn parse_pes_header_audio(payload: &[u8]) -> Option<(usize, Option<u64>)> {
if payload.len() < 9 {
return None;
}
if payload[0] != 0 || payload[1] != 0 || payload[2] != 1 {
return None;
}
let stream_id = payload[3];
if !(0xC0..=0xDF).contains(&stream_id) {
return None;
}
let flags = payload[7];
let pts_dts_flags = (flags >> 6) & 0x03;
let header_data_len = payload[8] as usize;
let es_start = 9 + header_data_len;
if es_start > payload.len() {
return None;
}
let pts = if pts_dts_flags == 0b10 || pts_dts_flags == 0b11 {
if payload.len() < 14 {
return None;
}
let p0 = ((payload[9] >> 1) & 0x07) as u64;
let p1 = (((payload[10] as u64) << 7) | ((payload[11] as u64) >> 1)) & 0x7FFF;
let p2 = (((payload[12] as u64) << 7) | ((payload[13] as u64) >> 1)) & 0x7FFF;
Some((p0 << 30) | (p1 << 15) | p2)
} else {
None
};
Some((es_start, pts))
}
pub struct TsStreamingDemuxer {
data: Vec<u8>,
header: DemuxHeader,
audio: Option<AudioTrack>,
packets: usize,
packet_stride: usize,
prefix_len: usize,
programs: Vec<ProgramInfo>,
active_program_idx: usize,
video_pid: u16,
next_pkt: usize,
pending: Vec<u8>,
pending_pts: Option<u64>,
have_first_start: bool,
eof: bool,
pixel_format_detected: bool,
encrypted_drop: bool,
}
pub(crate) fn demux_ts_streaming_init(data: &[u8]) -> Result<TsStreamingDemuxer> {
let owned = data.to_vec();
let (packets, packet_stride, prefix_len) = detect_packet_layout(&owned)?;
if packets == 0 {
bail!("TS: file contains no TS packets");
}
let mut pat_programs: Vec<PatProgram> = Vec::new();
for i in 0..packets {
let start = i * packet_stride + prefix_len;
let pkt = &owned[start..start + TS_PACKET];
if pkt[0] != TS_SYNC {
continue;
}
let pid = (((pkt[1] & 0x1F) as u16) << 8) | pkt[2] as u16;
if pid == 0
&& let Some(payload) = ts_psi_payload(pkt)
{
let progs = parse_pat_all_programs(payload);
if !progs.is_empty() {
pat_programs = progs;
break;
}
}
}
if pat_programs.is_empty() {
bail!("TS: no PAT entries found");
}
let mut programs: Vec<ProgramInfo> = pat_programs
.iter()
.map(|p| ProgramInfo {
program_number: p.program_number,
pmt_pid: p.pmt_pid,
video_streams: Vec::new(),
audio_streams: Vec::new(),
})
.collect();
let mut need: std::collections::HashSet<u16> = pat_programs.iter().map(|p| p.pmt_pid).collect();
for i in 0..packets {
if need.is_empty() {
break;
}
let start = i * packet_stride + prefix_len;
let pkt = &owned[start..start + TS_PACKET];
if pkt[0] != TS_SYNC {
continue;
}
let pid = (((pkt[1] & 0x1F) as u16) << 8) | pkt[2] as u16;
if !need.contains(&pid) {
continue;
}
if let Some(payload) = ts_psi_payload(pkt)
&& let Some((video_streams, audio_streams)) = parse_pmt_streams(payload)
{
if let Some(prog) = programs.iter_mut().find(|p| p.pmt_pid == pid) {
prog.video_streams = video_streams;
prog.audio_streams = audio_streams;
}
need.remove(&pid);
}
}
let active_program_idx = programs
.iter()
.position(|p| !p.video_streams.is_empty())
.context("TS: no program advertises a recognised video elementary stream")?;
let active = &programs[active_program_idx];
let video = active.video_streams[0];
let audio = active.audio_streams.first().copied();
let codec = match video.stream_type {
STREAM_TYPE_MPEG2_VIDEO => "mpeg2",
STREAM_TYPE_H264 => "h264",
STREAM_TYPE_HEVC => "h265",
other => bail!("TS: unsupported stream_type 0x{:02X}", other),
}
.to_string();
let scan = scan_first_video_au(&owned, packets, packet_stride, prefix_len, video.pid, 64);
let (width, height) = match &scan.first_au {
Some(au) => codec::pixel_format::detect_dims(&codec, std::slice::from_ref(au))
.unwrap_or_else(|| {
tracing::warn!(
codec = codec.as_str(),
video_pid = video.pid,
"TS streaming demux: first AU SPS parse failed; width/height=0×0"
);
(0, 0)
}),
None => {
tracing::warn!(
codec = codec.as_str(),
video_pid = video.pid,
"TS streaming demux: could not locate first video AU during init; width/height=0×0"
);
(0, 0)
}
};
let frame_rate = estimate_frame_rate_from_ptses(&scan.ptses).unwrap_or_else(|| {
tracing::warn!(
codec = codec.as_str(),
video_pid = video.pid,
pts_samples = scan.ptses.len(),
"TS streaming demux: could not derive frame_rate from PTS window; defaulting to 30.0"
);
30.0
});
let info = StreamInfo {
codec: codec.clone(),
width,
height,
frame_rate,
duration: 0.0,
pixel_format: PixelFormat::Yuv420p,
color_space: ColorSpace::Bt709,
total_frames: 0,
bitrate: 0,
color_metadata: Default::default(),
};
let audio_track = audio.and_then(|info| {
match extract_ts_audio(&owned, packets, packet_stride, prefix_len, info) {
Ok(track) => track,
Err(e) => {
tracing::warn!(
audio_pid = info.pid,
audio_kind = ?info.kind,
error = %e,
"TS audio extraction failed; emitting video-only"
);
None
}
}
});
Ok(TsStreamingDemuxer {
data: owned,
header: DemuxHeader { codec, info },
audio: audio_track,
packets,
packet_stride,
prefix_len,
programs,
active_program_idx,
video_pid: video.pid,
next_pkt: 0,
pending: Vec::new(),
pending_pts: None,
have_first_start: false,
eof: false,
pixel_format_detected: false,
encrypted_drop: false,
})
}
impl TsStreamingDemuxer {
pub fn programs(&self) -> &[ProgramInfo] {
&self.programs
}
pub fn active_program_index(&self) -> usize {
self.active_program_idx
}
pub fn select_program(&mut self, program_number: u16) -> Result<()> {
let new_idx = self
.programs
.iter()
.position(|p| p.program_number == program_number)
.with_context(|| format!("TS: program_number {} not found in PAT", program_number))?;
if self.programs[new_idx].video_streams.is_empty() {
bail!(
"TS: program {} has no recognised video stream",
program_number
);
}
let video = self.programs[new_idx].video_streams[0];
let audio = self.programs[new_idx].audio_streams.first().copied();
let codec = match video.stream_type {
STREAM_TYPE_MPEG2_VIDEO => "mpeg2",
STREAM_TYPE_H264 => "h264",
STREAM_TYPE_HEVC => "h265",
other => bail!(
"TS: program {} video stream_type 0x{:02X} unsupported",
program_number,
other
),
}
.to_string();
self.active_program_idx = new_idx;
self.video_pid = video.pid;
self.header.codec = codec.clone();
self.header.info.codec = codec.clone();
self.header.info.pixel_format = PixelFormat::Yuv420p;
self.pixel_format_detected = false;
let scan = scan_first_video_au(
&self.data,
self.packets,
self.packet_stride,
self.prefix_len,
video.pid,
64,
);
let (w, h) = match &scan.first_au {
Some(au) => {
codec::pixel_format::detect_dims(&codec, std::slice::from_ref(au)).unwrap_or((0, 0))
}
None => (0, 0),
};
self.header.info.width = w;
self.header.info.height = h;
self.header.info.frame_rate = estimate_frame_rate_from_ptses(&scan.ptses).unwrap_or(30.0);
self.next_pkt = 0;
self.pending.clear();
self.pending_pts = None;
self.have_first_start = false;
self.eof = false;
self.encrypted_drop = false;
self.audio = audio.and_then(|info| {
match extract_ts_audio(
&self.data,
self.packets,
self.packet_stride,
self.prefix_len,
info,
) {
Ok(track) => track,
Err(e) => {
tracing::warn!(
audio_pid = info.pid,
audio_kind = ?info.kind,
error = %e,
"TS audio extraction failed on program switch; emitting video-only"
);
None
}
}
});
Ok(())
}
fn yield_sample(&mut self, data: Vec<u8>, pts: Option<u64>) -> Sample {
if !self.pixel_format_detected {
let detected =
codec::pixel_format::detect(&self.header.codec, std::slice::from_ref(&data));
self.header.info.pixel_format = detected;
self.pixel_format_detected = true;
}
Sample {
data,
pts_ticks: pts.map(|p| p as i64).unwrap_or(0),
duration_ticks: 0,
}
}
}
impl StreamingDemuxer for TsStreamingDemuxer {
fn header(&self) -> &DemuxHeader {
&self.header
}
fn next_video_sample(&mut self) -> Result<Option<Sample>> {
if self.eof || self.encrypted_drop {
return Ok(None);
}
loop {
if self.next_pkt >= self.packets {
self.eof = true;
if !self.pending.is_empty() {
let data = std::mem::take(&mut self.pending);
let pts = self.pending_pts.take();
return Ok(Some(self.yield_sample(data, pts)));
}
return Ok(None);
}
let i = self.next_pkt;
self.next_pkt += 1;
let start = i * self.packet_stride + self.prefix_len;
let pkt = &self.data[start..start + TS_PACKET];
if pkt[0] != TS_SYNC {
continue;
}
let pid = (((pkt[1] & 0x1F) as u16) << 8) | pkt[2] as u16;
if pid != self.video_pid {
continue;
}
let pusi = pkt[1] & 0x40 != 0;
let scramble = (pkt[3] >> 6) & 0x03;
if scramble != 0 {
tracing::warn!(
video_pid = self.video_pid,
transport_scrambling_control = scramble,
error_kind = "encrypted_ts",
"encrypted TS stream; we don't carry CA tables — drop video output"
);
self.encrypted_drop = true;
self.pending.clear();
self.pending_pts = None;
self.have_first_start = false;
self.eof = true;
return Ok(None);
}
let adaptation = (pkt[3] >> 4) & 0x03;
let has_payload = adaptation & 0x01 != 0;
let has_adaptation = adaptation & 0x02 != 0;
if !has_payload {
continue;
}
let mut offset = 4usize;
if has_adaptation {
if offset >= TS_PACKET {
continue;
}
let adap_len = pkt[offset] as usize;
offset += 1 + adap_len;
if offset > TS_PACKET {
continue;
}
}
if offset >= TS_PACKET {
continue;
}
let payload = &pkt[offset..];
if pusi {
let had_pending = self.have_first_start;
let prev_data = if had_pending {
std::mem::take(&mut self.pending)
} else {
Vec::new()
};
let prev_pts = self.pending_pts.take();
self.have_first_start = true;
let Some((es_start, pts)) = parse_pes_header(payload) else {
self.have_first_start = false;
self.pending.clear();
if !prev_data.is_empty() {
return Ok(Some(self.yield_sample(prev_data, prev_pts)));
}
continue;
};
self.pending_pts = pts;
if es_start < payload.len() {
self.pending.extend_from_slice(&payload[es_start..]);
}
if !prev_data.is_empty() {
return Ok(Some(self.yield_sample(prev_data, prev_pts)));
}
} else if self.have_first_start {
self.pending.extend_from_slice(payload);
}
}
}
fn audio(&self) -> Option<&AudioTrack> {
self.audio.as_ref()
}
}
#[cfg(test)]
mod tests {
use super::*;
fn ts_pkt(pid: u16, pusi: bool, adaptation: u8, payload: &[u8]) -> [u8; TS_PACKET] {
let mut p = [0xFFu8; TS_PACKET];
p[0] = TS_SYNC;
p[1] = if pusi { 0x40 } else { 0x00 } | ((pid >> 8) & 0x1F) as u8;
p[2] = (pid & 0xFF) as u8;
p[3] = (adaptation & 0x03) << 4;
let mut off = 4;
let pay_len = payload.len().min(TS_PACKET - off);
p[off..off + pay_len].copy_from_slice(&payload[..pay_len]);
off += pay_len;
let _ = off;
p
}
#[test]
fn estimate_frame_rate_from_uniform_ptses_returns_exact_fps() {
let ptses: Vec<u64> = (0..64).map(|i| i as u64 * 3750).collect();
let fps = estimate_frame_rate_from_ptses(&ptses).expect("24 fps");
assert!((fps - 24.0).abs() < 1e-9, "{} != 24.0", fps);
}
#[test]
fn estimate_frame_rate_from_reordered_ptses_sorts_before_delta() {
let mut ptses: Vec<u64> = (0..32).map(|i| i as u64 * 3750).collect();
ptses.swap(5, 6);
ptses.swap(10, 11);
let fps = estimate_frame_rate_from_ptses(&ptses).expect("24 fps after swap");
assert!((fps - 24.0).abs() < 1e-9, "{} != 24.0", fps);
}
#[test]
fn estimate_frame_rate_from_single_outlier_delta_uses_median() {
let mut ptses: Vec<u64> = (0..24).map(|i| i as u64 * 3750).collect();
ptses.push(24 * 3750 + 37500); let fps = estimate_frame_rate_from_ptses(&ptses).expect("24 fps despite outlier");
assert!((fps - 24.0).abs() < 1e-9);
}
#[test]
fn estimate_frame_rate_returns_none_when_all_ptses_equal() {
let ptses = vec![0u64; 10];
assert!(estimate_frame_rate_from_ptses(&ptses).is_none());
}
#[test]
fn estimate_frame_rate_returns_none_when_fewer_than_two() {
assert!(estimate_frame_rate_from_ptses(&[]).is_none());
assert!(estimate_frame_rate_from_ptses(&[1234]).is_none());
}
#[test]
fn estimate_frame_rate_rejects_out_of_range_values() {
let ptses = vec![0u64, 1];
assert!(estimate_frame_rate_from_ptses(&ptses).is_none());
}
#[test]
fn estimate_frame_rate_handles_29_97_ntsc() {
let ptses: Vec<u64> = (0..32).map(|i| i as u64 * 3003).collect();
let fps = estimate_frame_rate_from_ptses(&ptses).expect("29.97");
assert!((fps - 30.0).abs() < 0.05, "got {}", fps); }
#[test]
fn detects_plain_ts_layout() {
let mut buf = Vec::with_capacity(3 * TS_PACKET);
for _ in 0..3 {
let pkt = ts_pkt(0x1FFF, false, 0b01, &[]);
buf.extend_from_slice(&pkt);
}
let (count, stride, prefix) = detect_packet_layout(&buf).unwrap();
assert_eq!((count, stride, prefix), (3, 188, 0));
}
#[test]
fn parses_minimal_pat_pmt_and_reassembles_one_sample() {
let mut pat = vec![0u8; 0];
pat.push(0x00); let section_length: usize = 5 + 4 + 4; pat.push(0xB0 | ((section_length >> 8) & 0x0F) as u8);
pat.push((section_length & 0xFF) as u8);
pat.extend_from_slice(&[0x00, 0x01, 0xC1, 0x00, 0x00]); pat.extend_from_slice(&[0x00, 0x01]); pat.extend_from_slice(&[0xE1, 0x00]); pat.extend_from_slice(&[0, 0, 0, 0]);
let mut pat_payload = vec![0u8];
pat_payload.extend_from_slice(&pat);
let pat_pkt = ts_pkt(0x0000, true, 0b01, &pat_payload);
let mut pmt = vec![0u8; 0];
pmt.push(0x02);
let pmt_sec_len: usize = 9 + 5 + 4; pmt.push(0xB0 | ((pmt_sec_len >> 8) & 0x0F) as u8);
pmt.push((pmt_sec_len & 0xFF) as u8);
pmt.extend_from_slice(&[0x00, 0x01, 0xC1, 0x00, 0x00]); pmt.extend_from_slice(&[0xE2, 0x00]); pmt.extend_from_slice(&[0xF0, 0x00]); pmt.extend_from_slice(&[STREAM_TYPE_MPEG2_VIDEO, 0xE2, 0x00, 0xF0, 0x00]); pmt.extend_from_slice(&[0, 0, 0, 0]); let mut pmt_payload = vec![0u8];
pmt_payload.extend_from_slice(&pmt);
let pmt_pkt = ts_pkt(0x0100, true, 0b01, &pmt_payload);
let make_pes = |byte: u8| {
let mut pes = vec![0u8, 0u8, 1u8]; pes.push(0xE0); pes.extend_from_slice(&[0u8, 0u8]); pes.push(0x80);
pes.push(0x80); pes.push(5); pes.extend_from_slice(&[0x21, 0x00, 0x01, 0x00, 0x01]); pes.extend_from_slice(&[byte; 16]);
pes
};
let pes_pkt_a = ts_pkt(0x0200, true, 0b01, &make_pes(0xAA));
let pes_pkt_b = ts_pkt(0x0200, true, 0b01, &make_pes(0xBB));
let mut buf = Vec::new();
buf.extend_from_slice(&pat_pkt);
buf.extend_from_slice(&pmt_pkt);
buf.extend_from_slice(&pes_pkt_a);
buf.extend_from_slice(&pes_pkt_b);
buf.extend_from_slice(&ts_pkt(0x1FFF, false, 0b01, &[]));
let d = demux_ts(&buf).expect("demux");
assert_eq!(d.codec, "mpeg2");
assert_eq!(d.samples.len(), 2);
assert_eq!(&d.samples[0][..16], &[0xAA; 16]);
assert_eq!(&d.samples[1][..16], &[0xBB; 16]);
}
#[test]
fn rejects_file_with_no_sync() {
let garbage = vec![0u8; TS_PACKET * 3];
assert!(demux_ts(&garbage).is_err());
}
fn build_adts_header_7(profile: u8, sr_idx: u8, ch_cfg: u8, frame_length: usize) -> [u8; 7] {
let mut h = [0u8; 7];
h[0] = 0xFF;
h[1] = 0xF0 | 0x01; h[2] = ((profile & 0x03) << 6) | ((sr_idx & 0x0F) << 2) | ((ch_cfg >> 2) & 0x01);
h[3] = ((ch_cfg & 0x03) << 6) | (((frame_length >> 11) & 0x03) as u8);
h[4] = ((frame_length >> 3) & 0xFF) as u8;
h[5] = (((frame_length & 0x07) << 5) | 0x1F) as u8;
h[6] = 0xFC;
h
}
fn build_adts_header_9(profile: u8, sr_idx: u8, ch_cfg: u8, frame_length: usize) -> [u8; 9] {
let mut h = [0u8; 9];
h[0] = 0xFF;
h[1] = 0xF0; h[2] = ((profile & 0x03) << 6) | ((sr_idx & 0x0F) << 2) | ((ch_cfg >> 2) & 0x01);
h[3] = ((ch_cfg & 0x03) << 6) | (((frame_length >> 11) & 0x03) as u8);
h[4] = ((frame_length >> 3) & 0xFF) as u8;
h[5] = (((frame_length & 0x07) << 5) | 0x1F) as u8;
h[6] = 0xFC;
h
}
#[test]
fn adts_parser_decodes_canonical_lc_stereo_7byte_header() {
let h = build_adts_header_7(1, 3, 2, 107);
let parsed = parse_adts_header(&h).expect("must parse 7-byte ADTS header");
assert_eq!(parsed.profile, 1, "ADTS profile=1 LC");
assert_eq!(parsed.sampling_frequency_index, 3, "sr_idx=3 → 48kHz");
assert_eq!(parsed.channel_configuration, 2, "ch_cfg=2 stereo");
assert_eq!(parsed.frame_length, 107);
assert_eq!(parsed.header_len, 7, "protection_absent=1 → 7-byte header");
assert_eq!(
decode_sample_rate_index(parsed.sampling_frequency_index),
Some(48000)
);
}
#[test]
fn adts_parser_decodes_9byte_header_with_crc() {
let h = build_adts_header_9(1, 4, 2, 109);
let parsed = parse_adts_header(&h).expect("must parse 9-byte ADTS header");
assert_eq!(parsed.profile, 1);
assert_eq!(parsed.sampling_frequency_index, 4, "sr_idx=4 → 44.1kHz");
assert_eq!(parsed.channel_configuration, 2);
assert_eq!(parsed.frame_length, 109);
assert_eq!(
parsed.header_len, 9,
"protection_absent=0 → 9-byte header (incl CRC)"
);
assert_eq!(
decode_sample_rate_index(parsed.sampling_frequency_index),
Some(44100)
);
}
#[test]
fn adts_parser_decodes_aac_profile_bits_full_range() {
for profile in 0u8..=3 {
let h = build_adts_header_7(profile, 3, 2, 32);
let parsed =
parse_adts_header(&h).unwrap_or_else(|| panic!("must parse profile={profile}"));
assert_eq!(parsed.profile, profile);
}
}
#[test]
fn adts_parser_rejects_missing_sync() {
let mut h = build_adts_header_7(1, 3, 2, 32);
h[0] = 0x00;
assert!(parse_adts_header(&h).is_none());
}
#[test]
fn adts_parser_rejects_short_buffer() {
let h = build_adts_header_7(1, 3, 2, 32);
assert!(
parse_adts_header(&h[..6]).is_none(),
"<7 bytes can't carry a complete ADTS header"
);
}
#[test]
fn synthesize_asc_lc_stereo_48k_emits_0x1190() {
let adts = AdtsHeader {
profile: 1,
sampling_frequency_index: 3,
channel_configuration: 2,
frame_length: 0,
header_len: 7,
};
let asc = synthesize_asc(&adts);
assert_eq!(asc, [0x11, 0x90], "LC/48k/stereo → ASC 0x11 0x90");
}
#[test]
fn synthesize_asc_lc_mono_44k() {
let adts = AdtsHeader {
profile: 1,
sampling_frequency_index: 4,
channel_configuration: 1,
frame_length: 0,
header_len: 7,
};
assert_eq!(synthesize_asc(&adts), [0x12, 0x08]);
}
#[test]
fn synthesize_asc_main_aot_at_44k_5p1_rejected_at_channel_layer() {
let adts = AdtsHeader {
profile: 0,
sampling_frequency_index: 4,
channel_configuration: 6,
frame_length: 0,
header_len: 7,
};
assert_eq!(synthesize_asc(&adts), [0x0A, 0x30]);
}
#[test]
fn adts_strip_7byte_header_yields_payload_only() {
let mut frame = Vec::with_capacity(107);
frame.extend_from_slice(&build_adts_header_7(1, 3, 2, 107));
frame.extend_from_slice(&[0x42u8; 100]);
let header = parse_adts_header(&frame).unwrap();
assert_eq!(header.frame_length, 107);
let payload = &frame[header.header_len..header.frame_length];
assert_eq!(payload.len(), 100);
assert!(payload.iter().all(|b| *b == 0x42));
}
#[test]
fn adts_sample_rate_table_covers_documented_indices() {
assert_eq!(decode_sample_rate_index(0), Some(96000));
assert_eq!(decode_sample_rate_index(3), Some(48000));
assert_eq!(decode_sample_rate_index(4), Some(44100));
assert_eq!(decode_sample_rate_index(12), Some(7350));
assert!(decode_sample_rate_index(13).is_none(), "13 is reserved");
assert!(
decode_sample_rate_index(15).is_none(),
"15 (escape) not supported"
);
}
#[test]
fn demux_ts_yields_audio_track_when_pmt_advertises_aac() {
let mut pat = vec![0x00];
let pat_section_len: usize = 5 + 4 + 4;
pat.push(0xB0 | ((pat_section_len >> 8) & 0x0F) as u8);
pat.push((pat_section_len & 0xFF) as u8);
pat.extend_from_slice(&[0x00, 0x01, 0xC1, 0x00, 0x00]);
pat.extend_from_slice(&[0x00, 0x01, 0xE1, 0x00, 0u8, 0u8, 0u8, 0u8]);
let mut pat_payload = vec![0u8];
pat_payload.extend_from_slice(&pat);
let pat_pkt = ts_pkt(0x0000, true, 0b01, &pat_payload);
let mut pmt = vec![0x02];
let pmt_section_len: usize = 9 + 5 + 5 + 4; pmt.push(0xB0 | ((pmt_section_len >> 8) & 0x0F) as u8);
pmt.push((pmt_section_len & 0xFF) as u8);
pmt.extend_from_slice(&[0x00, 0x01, 0xC1, 0x00, 0x00]);
pmt.extend_from_slice(&[0xE2, 0x00]); pmt.extend_from_slice(&[0xF0, 0x00]); pmt.extend_from_slice(&[STREAM_TYPE_MPEG2_VIDEO, 0xE2, 0x00, 0xF0, 0x00]);
pmt.extend_from_slice(&[STREAM_TYPE_AAC_ADTS, 0xE3, 0x00, 0xF0, 0x00]);
pmt.extend_from_slice(&[0u8; 4]); let mut pmt_payload = vec![0u8];
pmt_payload.extend_from_slice(&pmt);
let pmt_pkt = ts_pkt(0x0100, true, 0b01, &pmt_payload);
let video_pes = {
let mut pes = vec![
0u8, 0u8, 1u8, 0xE0, 0u8, 0u8, 0x80, 0x80, 5, 0x21, 0x00, 0x01, 0x00, 0x01,
];
pes.extend_from_slice(&[0xAAu8; 16]);
pes
};
let video_pkt = ts_pkt(0x0200, true, 0b01, &video_pes);
let mut adts_stream = Vec::new();
for fill in [0xCCu8, 0xDDu8] {
adts_stream.extend_from_slice(&build_adts_header_7(1, 3, 2, 39));
adts_stream.extend_from_slice(&[fill; 32]);
}
let audio_pes = {
let mut pes = vec![
0u8, 0u8, 1u8, 0xC0, 0u8, 0u8, 0x80, 0x80, 5, 0x21, 0x00, 0x01, 0x00, 0x01,
];
pes.extend_from_slice(&adts_stream);
pes
};
let audio_pkt = ts_pkt(0x0300, true, 0b01, &audio_pes);
let mut buf = Vec::new();
buf.extend_from_slice(&pat_pkt);
buf.extend_from_slice(&pmt_pkt);
buf.extend_from_slice(&video_pkt);
buf.extend_from_slice(&audio_pkt);
buf.extend_from_slice(&ts_pkt(0x1FFF, false, 0b01, &[]));
let d = demux_ts(&buf).expect("demux must succeed");
assert_eq!(d.codec, "mpeg2");
let audio = d.audio.expect("AAC audio track must be surfaced");
assert_eq!(audio.codec, "aac");
assert_eq!(audio.channels, 2, "ch_cfg=2 stereo");
assert_eq!(audio.sample_rate, 48000, "sr_idx=3 → 48k");
assert_eq!(audio.timescale, 48000, "AAC timescale = sample_rate");
assert_eq!(
audio.asc,
vec![0x11, 0x90],
"synthesized ASC for LC/48k/stereo"
);
assert_eq!(audio.samples.len(), 2, "two ADTS frames → two samples");
assert_eq!(
audio.samples[0].len(),
32,
"32-byte payload after 7-byte header strip"
);
assert!(audio.samples[0].iter().all(|b| *b == 0xCC));
assert!(audio.samples[1].iter().all(|b| *b == 0xDD));
assert_eq!(
audio.durations,
vec![1024, 1024],
"AAC-LC frame duration = 1024 ticks @ sample-rate timescale"
);
}
#[test]
fn demux_ts_emits_audio_none_when_no_aac_stream_in_pmt() {
let mut buf = Vec::new();
let mut pat = vec![0x00];
let pat_section_len: usize = 5 + 4 + 4;
pat.push(0xB0 | ((pat_section_len >> 8) & 0x0F) as u8);
pat.push((pat_section_len & 0xFF) as u8);
pat.extend_from_slice(&[0x00, 0x01, 0xC1, 0x00, 0x00]);
pat.extend_from_slice(&[0x00, 0x01, 0xE1, 0x00, 0u8, 0u8, 0u8, 0u8]);
let mut pat_payload = vec![0u8];
pat_payload.extend_from_slice(&pat);
buf.extend_from_slice(&ts_pkt(0x0000, true, 0b01, &pat_payload));
let mut pmt = vec![0x02];
let pmt_section_len: usize = 9 + 5 + 4;
pmt.push(0xB0 | ((pmt_section_len >> 8) & 0x0F) as u8);
pmt.push((pmt_section_len & 0xFF) as u8);
pmt.extend_from_slice(&[0x00, 0x01, 0xC1, 0x00, 0x00]);
pmt.extend_from_slice(&[0xE2, 0x00, 0xF0, 0x00]);
pmt.extend_from_slice(&[STREAM_TYPE_MPEG2_VIDEO, 0xE2, 0x00, 0xF0, 0x00]);
pmt.extend_from_slice(&[0u8; 4]);
let mut pmt_payload = vec![0u8];
pmt_payload.extend_from_slice(&pmt);
buf.extend_from_slice(&ts_pkt(0x0100, true, 0b01, &pmt_payload));
let video_pes = {
let mut pes = vec![
0u8, 0u8, 1u8, 0xE0, 0u8, 0u8, 0x80, 0x80, 5, 0x21, 0x00, 0x01, 0x00, 0x01,
];
pes.extend_from_slice(&[0xAAu8; 16]);
pes
};
buf.extend_from_slice(&ts_pkt(0x0200, true, 0b01, &video_pes));
buf.extend_from_slice(&ts_pkt(0x1FFF, false, 0b01, &[]));
let d = demux_ts(&buf).expect("demux");
assert!(
d.audio.is_none(),
"PMT without AAC-ADTS stream → no audio track surfaced"
);
}
fn synth_ac3_frame_stereo_48k_128k() -> Vec<u8> {
let mut bw = BitWriter::new();
bw.put(16, 0x0B77); bw.put(16, 0); bw.put(2, 0); bw.put(6, 8 << 1); bw.put(5, 8); bw.put(3, 0); bw.put(3, 2); bw.put(2, 0);
bw.put(1, 0); while bw.bytes.len() < 384 {
bw.put(8, 0);
}
bw.flush()
}
fn synth_eac3_frame_stereo_48k_192bytes() -> Vec<u8> {
let mut bw = BitWriter::new();
bw.put(16, 0x0B77);
bw.put(2, 0); bw.put(3, 0); bw.put(11, 0x5F); bw.put(2, 0); bw.put(2, 3); bw.put(3, 2); bw.put(1, 0); bw.put(5, 16); bw.put(5, 0); bw.put(1, 0); while bw.bytes.len() < 192 {
bw.put(8, 0);
}
bw.flush()
}
struct BitWriter {
bytes: Vec<u8>,
bit_pos: usize,
}
impl BitWriter {
fn new() -> Self {
Self {
bytes: Vec::new(),
bit_pos: 0,
}
}
fn put(&mut self, n: usize, v: u32) {
for i in (0..n).rev() {
let bit = ((v >> i) & 0x01) as u8;
if self.bit_pos % 8 == 0 {
self.bytes.push(0);
}
let byte_idx = self.bit_pos / 8;
let bit_idx = 7 - (self.bit_pos % 8);
self.bytes[byte_idx] |= bit << bit_idx;
self.bit_pos += 1;
}
}
fn flush(self) -> Vec<u8> {
self.bytes
}
}
fn ts_pkt_continuation(pid: u16, payload: &[u8]) -> [u8; TS_PACKET] {
let mut p = [0xFFu8; TS_PACKET];
p[0] = TS_SYNC;
p[1] = ((pid >> 8) & 0x1F) as u8; p[2] = (pid & 0xFF) as u8;
p[3] = 0b01 << 4; let pay_len = payload.len().min(TS_PACKET - 4);
p[4..4 + pay_len].copy_from_slice(&payload[..pay_len]);
p
}
fn build_ts_with_audio(
audio_stream_type: u8,
audio_descriptors: &[u8],
audio_pid: u16,
audio_es: &[u8],
) -> Vec<u8> {
let mut pat = vec![0x00];
let pat_section_len: usize = 5 + 4 + 4;
pat.push(0xB0 | ((pat_section_len >> 8) & 0x0F) as u8);
pat.push((pat_section_len & 0xFF) as u8);
pat.extend_from_slice(&[0x00, 0x01, 0xC1, 0x00, 0x00]);
pat.extend_from_slice(&[0x00, 0x01, 0xE1, 0x00, 0u8, 0u8, 0u8, 0u8]);
let mut pat_payload = vec![0u8];
pat_payload.extend_from_slice(&pat);
let pat_pkt = ts_pkt(0x0000, true, 0b01, &pat_payload);
let mut pmt = vec![0x02];
let pmt_stream_entries = 5 + 5 + audio_descriptors.len(); let pmt_section_len: usize = 9 + pmt_stream_entries + 4;
pmt.push(0xB0 | ((pmt_section_len >> 8) & 0x0F) as u8);
pmt.push((pmt_section_len & 0xFF) as u8);
pmt.extend_from_slice(&[0x00, 0x01, 0xC1, 0x00, 0x00]);
pmt.extend_from_slice(&[0xE2, 0x00]); pmt.extend_from_slice(&[0xF0, 0x00]); pmt.extend_from_slice(&[STREAM_TYPE_MPEG2_VIDEO, 0xE2, 0x00, 0xF0, 0x00]);
pmt.push(audio_stream_type);
pmt.push(0xE0 | ((audio_pid >> 8) & 0x1F) as u8);
pmt.push((audio_pid & 0xFF) as u8);
let esi_len = audio_descriptors.len() as u16;
pmt.push(0xF0 | ((esi_len >> 8) & 0x0F) as u8);
pmt.push((esi_len & 0xFF) as u8);
pmt.extend_from_slice(audio_descriptors);
pmt.extend_from_slice(&[0u8; 4]); let mut pmt_payload = vec![0u8];
pmt_payload.extend_from_slice(&pmt);
let pmt_pkt = ts_pkt(0x0100, true, 0b01, &pmt_payload);
let video_pes = {
let mut pes = vec![
0u8, 0u8, 1u8, 0xE0, 0u8, 0u8, 0x80, 0x80, 5, 0x21, 0x00, 0x01, 0x00, 0x01,
];
pes.extend_from_slice(&[0xAAu8; 16]);
pes
};
let video_pkt = ts_pkt(0x0200, true, 0b01, &video_pes);
let mut audio_pes = vec![
0u8, 0u8, 1u8, 0xC0, 0u8, 0u8, 0x80, 0x80, 5, 0x21, 0x00, 0x01, 0x00, 0x01,
];
audio_pes.extend_from_slice(audio_es);
let first_chunk_max = TS_PACKET - 4; let mut audio_pkts: Vec<[u8; TS_PACKET]> = Vec::new();
let first_len = audio_pes.len().min(first_chunk_max);
audio_pkts.push(ts_pkt(audio_pid, true, 0b01, &audio_pes[..first_len]));
let mut cursor = first_len;
while cursor < audio_pes.len() {
let end = (cursor + first_chunk_max).min(audio_pes.len());
audio_pkts.push(ts_pkt_continuation(audio_pid, &audio_pes[cursor..end]));
cursor = end;
}
let mut buf = Vec::new();
buf.extend_from_slice(&pat_pkt);
buf.extend_from_slice(&pmt_pkt);
buf.extend_from_slice(&video_pkt);
for pkt in &audio_pkts {
buf.extend_from_slice(pkt);
}
buf.extend_from_slice(&ts_pkt(0x1FFF, false, 0b01, &[]));
buf
}
#[test]
fn pmt_walker_classifies_aac_ac3_eac3_stream_types() {
let mut pmt = vec![0x02];
let stream_entries = 5 + 5 + 5 + 5; let pmt_section_len: usize = 9 + stream_entries + 4;
pmt.push(0xB0 | ((pmt_section_len >> 8) & 0x0F) as u8);
pmt.push((pmt_section_len & 0xFF) as u8);
pmt.extend_from_slice(&[0x00, 0x01, 0xC1, 0x00, 0x00]);
pmt.extend_from_slice(&[0xE2, 0x00, 0xF0, 0x00]); pmt.extend_from_slice(&[STREAM_TYPE_MPEG2_VIDEO, 0xE2, 0x00, 0xF0, 0x00]);
pmt.extend_from_slice(&[STREAM_TYPE_AAC_ADTS, 0xE3, 0x00, 0xF0, 0x00]);
pmt.extend_from_slice(&[STREAM_TYPE_AC3, 0xE4, 0x00, 0xF0, 0x00]);
pmt.extend_from_slice(&[STREAM_TYPE_EAC3, 0xE5, 0x00, 0xF0, 0x00]);
pmt.extend_from_slice(&[0u8; 4]);
let (video, audio) = parse_pmt_streams(&pmt).expect("parse");
assert_eq!(video.len(), 1);
assert_eq!(video[0].pid, 0x200);
assert_eq!(audio.len(), 3);
assert_eq!(
(audio[0].pid, audio[0].kind),
(0x300, AudioCodecKind::AacAdts)
);
assert_eq!((audio[1].pid, audio[1].kind), (0x400, AudioCodecKind::Ac3));
assert_eq!((audio[2].pid, audio[2].kind), (0x500, AudioCodecKind::Eac3));
}
#[test]
fn pmt_walker_recognises_dvb_ac3_via_registration_descriptor() {
let mut pmt = vec![0x02];
let descriptors: [u8; 6] = [DESC_TAG_REGISTRATION, 4, b'A', b'C', b'-', b'3'];
let stream_entries = 5 + 5 + descriptors.len();
let pmt_section_len: usize = 9 + stream_entries + 4;
pmt.push(0xB0 | ((pmt_section_len >> 8) & 0x0F) as u8);
pmt.push((pmt_section_len & 0xFF) as u8);
pmt.extend_from_slice(&[0x00, 0x01, 0xC1, 0x00, 0x00]);
pmt.extend_from_slice(&[0xE2, 0x00, 0xF0, 0x00]);
pmt.extend_from_slice(&[STREAM_TYPE_MPEG2_VIDEO, 0xE2, 0x00, 0xF0, 0x00]);
pmt.push(STREAM_TYPE_PES_PRIVATE);
pmt.extend_from_slice(&[0xE3, 0x00]);
let esi_len = descriptors.len() as u16;
pmt.push(0xF0 | ((esi_len >> 8) & 0x0F) as u8);
pmt.push((esi_len & 0xFF) as u8);
pmt.extend_from_slice(&descriptors);
pmt.extend_from_slice(&[0u8; 4]);
let (_, audio) = parse_pmt_streams(&pmt).expect("parse");
assert_eq!(audio.len(), 1);
assert_eq!(audio[0].kind, AudioCodecKind::Ac3);
assert_eq!(audio[0].stream_type, STREAM_TYPE_PES_PRIVATE);
}
#[test]
fn pmt_walker_recognises_dvb_eac3_via_registration_descriptor() {
let mut pmt = vec![0x02];
let descriptors: [u8; 6] = [DESC_TAG_REGISTRATION, 4, b'E', b'A', b'C', b'3'];
let stream_entries = 5 + 5 + descriptors.len();
let pmt_section_len: usize = 9 + stream_entries + 4;
pmt.push(0xB0 | ((pmt_section_len >> 8) & 0x0F) as u8);
pmt.push((pmt_section_len & 0xFF) as u8);
pmt.extend_from_slice(&[0x00, 0x01, 0xC1, 0x00, 0x00]);
pmt.extend_from_slice(&[0xE2, 0x00, 0xF0, 0x00]);
pmt.extend_from_slice(&[STREAM_TYPE_MPEG2_VIDEO, 0xE2, 0x00, 0xF0, 0x00]);
pmt.push(STREAM_TYPE_PES_PRIVATE);
pmt.extend_from_slice(&[0xE3, 0x00]);
let esi_len = descriptors.len() as u16;
pmt.push(0xF0 | ((esi_len >> 8) & 0x0F) as u8);
pmt.push((esi_len & 0xFF) as u8);
pmt.extend_from_slice(&descriptors);
pmt.extend_from_slice(&[0u8; 4]);
let (_, audio) = parse_pmt_streams(&pmt).expect("parse");
assert_eq!(audio.len(), 1);
assert_eq!(audio[0].kind, AudioCodecKind::Eac3);
}
#[test]
fn extract_ac3_frames_from_synthetic_ts_yields_passthrough_track() {
let frame = synth_ac3_frame_stereo_48k_128k();
let mut es = frame.clone();
es.extend_from_slice(&frame);
let buf = build_ts_with_audio(STREAM_TYPE_AC3, &[], 0x300, &es);
let d = demux_ts(&buf).expect("demux");
let audio = d.audio.expect("AC-3 audio surfaced");
assert_eq!(audio.codec, "ac3");
assert_eq!(audio.channels, 2);
assert_eq!(audio.sample_rate, 48_000);
assert_eq!(audio.timescale, 48_000);
assert_eq!(audio.codec_private.len(), 3);
assert!(
audio.samples.len() >= 1,
"at least one AC-3 frame extracted"
);
assert_eq!(
&audio.samples[0][..2],
&[0x0B, 0x77],
"AC-3 frame begins with 0x0B77 sync word verbatim"
);
assert!(
audio.durations.iter().all(|&d| d == 1536),
"AC-3 frames are 1536 samples each"
);
}
#[test]
fn extract_eac3_frames_from_synthetic_ts_yields_passthrough_track() {
let frame = synth_eac3_frame_stereo_48k_192bytes();
let mut es = frame.clone();
es.extend_from_slice(&frame);
let buf = build_ts_with_audio(STREAM_TYPE_EAC3, &[], 0x300, &es);
let d = demux_ts(&buf).expect("demux");
let audio = d.audio.expect("E-AC-3 audio surfaced");
assert_eq!(audio.codec, "eac3");
assert_eq!(audio.channels, 2);
assert_eq!(audio.sample_rate, 48_000);
assert_eq!(audio.codec_private.len(), 5);
assert!(!audio.samples.is_empty());
assert_eq!(
&audio.samples[0][..2],
&[0x0B, 0x77],
"E-AC-3 frame begins with 0x0B77 sync word verbatim"
);
assert!(audio.durations.iter().all(|&d| d == 1536));
}
#[test]
fn extract_ac3_via_pes_private_with_dvb_registration() {
let frame = synth_ac3_frame_stereo_48k_128k();
let descriptors: [u8; 6] = [DESC_TAG_REGISTRATION, 4, b'A', b'C', b'-', b'3'];
let buf = build_ts_with_audio(STREAM_TYPE_PES_PRIVATE, &descriptors, 0x300, &frame);
let d = demux_ts(&buf).expect("demux");
let audio = d.audio.expect("AC-3 audio via DVB registration surfaced");
assert_eq!(audio.codec, "ac3");
assert_eq!(&audio.samples[0][..2], &[0x0B, 0x77]);
}
#[test]
fn dac3_body_synthesized_from_first_ts_frame_matches_sync_header() {
let frame = synth_ac3_frame_stereo_48k_128k();
let buf = build_ts_with_audio(STREAM_TYPE_AC3, &[], 0x300, &frame);
let d = demux_ts(&buf).expect("demux");
let audio = d.audio.expect("AC-3 audio");
let parsed = match crate::ac3_sync::parse_sync_info(&frame).unwrap() {
crate::ac3_sync::SyncInfo::Ac3(s) => s,
_ => panic!("expected AC-3"),
};
let expected = crate::mux::dac3_body_from_sync(&parsed);
assert_eq!(
audio.codec_private,
expected.to_vec(),
"TS-extracted dac3 must match the canonical helper"
);
}
fn build_two_program_ts() -> Vec<u8> {
let mut pat = vec![0x00];
let pat_section_len: usize = 5 + 4 + 4 + 4; pat.push(0xB0 | ((pat_section_len >> 8) & 0x0F) as u8);
pat.push((pat_section_len & 0xFF) as u8);
pat.extend_from_slice(&[0x00, 0x01, 0xC1, 0x00, 0x00]);
pat.extend_from_slice(&[0x00, 0x01, 0xE1, 0x00]); pat.extend_from_slice(&[0x00, 0x02, 0xE1, 0x01]); pat.extend_from_slice(&[0u8; 4]);
let mut pat_payload = vec![0u8];
pat_payload.extend_from_slice(&pat);
let pat_pkt = ts_pkt(0x0000, true, 0b01, &pat_payload);
let mut pmt1 = vec![0x02];
let pmt1_section_len: usize = 9 + 5 + 4;
pmt1.push(0xB0 | ((pmt1_section_len >> 8) & 0x0F) as u8);
pmt1.push((pmt1_section_len & 0xFF) as u8);
pmt1.extend_from_slice(&[0x00, 0x01, 0xC1, 0x00, 0x00]); pmt1.extend_from_slice(&[0xE2, 0x00, 0xF0, 0x00]);
pmt1.extend_from_slice(&[STREAM_TYPE_MPEG2_VIDEO, 0xE2, 0x00, 0xF0, 0x00]);
pmt1.extend_from_slice(&[0u8; 4]);
let mut pmt1_payload = vec![0u8];
pmt1_payload.extend_from_slice(&pmt1);
let pmt1_pkt = ts_pkt(0x0100, true, 0b01, &pmt1_payload);
let mut pmt2 = vec![0x02];
let pmt2_section_len: usize = 9 + 5 + 4;
pmt2.push(0xB0 | ((pmt2_section_len >> 8) & 0x0F) as u8);
pmt2.push((pmt2_section_len & 0xFF) as u8);
pmt2.extend_from_slice(&[0x00, 0x02, 0xC1, 0x00, 0x00]); pmt2.extend_from_slice(&[0xE3, 0x00, 0xF0, 0x00]);
pmt2.extend_from_slice(&[STREAM_TYPE_H264, 0xE3, 0x00, 0xF0, 0x00]);
pmt2.extend_from_slice(&[0u8; 4]);
let mut pmt2_payload = vec![0u8];
pmt2_payload.extend_from_slice(&pmt2);
let pmt2_pkt = ts_pkt(0x0101, true, 0b01, &pmt2_payload);
let make_pes = |fill: u8| {
let mut pes = vec![
0u8, 0u8, 1u8, 0xE0, 0u8, 0u8, 0x80, 0x80, 5, 0x21, 0x00, 0x01, 0x00, 0x01,
];
pes.extend_from_slice(&[fill; 16]);
pes
};
let p1_pes = ts_pkt(0x0200, true, 0b01, &make_pes(0xAA));
let p2_pes = ts_pkt(0x0300, true, 0b01, &make_pes(0xBB));
let mut buf = Vec::new();
buf.extend_from_slice(&pat_pkt);
buf.extend_from_slice(&pmt1_pkt);
buf.extend_from_slice(&pmt2_pkt);
buf.extend_from_slice(&p1_pes);
buf.extend_from_slice(&p2_pes);
buf.extend_from_slice(&ts_pkt(0x0200, true, 0b01, &make_pes(0xAA)));
buf.extend_from_slice(&ts_pkt(0x0300, true, 0b01, &make_pes(0xBB)));
buf.extend_from_slice(&ts_pkt(0x1FFF, false, 0b01, &[]));
buf
}
#[test]
fn streaming_demuxer_lists_all_pat_programs() {
let buf = build_two_program_ts();
let dem = demux_ts_streaming_init(&buf).expect("init");
let progs = dem.programs();
assert_eq!(progs.len(), 2, "PAT advertised 2 programs");
let nums: Vec<u16> = progs.iter().map(|p| p.program_number).collect();
assert_eq!(nums, vec![1, 2]);
assert_eq!(progs[0].pmt_pid, 0x100);
assert_eq!(progs[1].pmt_pid, 0x101);
assert_eq!(progs[0].video_streams[0].pid, 0x200);
assert_eq!(
progs[0].video_streams[0].stream_type,
STREAM_TYPE_MPEG2_VIDEO
);
assert_eq!(progs[1].video_streams[0].pid, 0x300);
assert_eq!(progs[1].video_streams[0].stream_type, STREAM_TYPE_H264);
}
#[test]
fn streaming_demuxer_default_picks_first_program() {
let buf = build_two_program_ts();
let mut dem = demux_ts_streaming_init(&buf).expect("init");
assert_eq!(dem.active_program_index(), 0);
assert_eq!(dem.header().codec, "mpeg2", "program 1 is MPEG-2");
let s = dem.next_video_sample().expect("sample").expect("some");
assert!(
s.data.iter().any(|&b| b == 0xAA),
"program 1 sample should carry 0xAA"
);
assert!(
!s.data.iter().any(|&b| b == 0xBB),
"program 1 sample must not carry program 2's 0xBB"
);
}
#[test]
fn streaming_demuxer_select_program_switches_active_streams() {
let buf = build_two_program_ts();
let mut dem = demux_ts_streaming_init(&buf).expect("init");
dem.select_program(2).expect("switch to program 2");
assert_eq!(dem.active_program_index(), 1);
assert_eq!(dem.header().codec, "h264", "program 2 is H.264");
let s = dem.next_video_sample().expect("sample").expect("some");
assert!(
s.data.iter().any(|&b| b == 0xBB),
"program 2 sample should carry 0xBB"
);
assert!(
!s.data.iter().any(|&b| b == 0xAA),
"program 2 sample must not carry program 1's 0xAA"
);
}
#[test]
fn streaming_demuxer_select_program_rejects_unknown_number() {
let buf = build_two_program_ts();
let mut dem = demux_ts_streaming_init(&buf).expect("init");
assert!(
dem.select_program(99).is_err(),
"unknown program_number must error rather than silently no-op"
);
}
fn build_encrypted_ts() -> Vec<u8> {
let mut pat = vec![0x00];
let pat_section_len: usize = 5 + 4 + 4;
pat.push(0xB0 | ((pat_section_len >> 8) & 0x0F) as u8);
pat.push((pat_section_len & 0xFF) as u8);
pat.extend_from_slice(&[0x00, 0x01, 0xC1, 0x00, 0x00]);
pat.extend_from_slice(&[0x00, 0x01, 0xE1, 0x00, 0u8, 0u8, 0u8, 0u8]);
let mut pat_payload = vec![0u8];
pat_payload.extend_from_slice(&pat);
let pat_pkt = ts_pkt(0x0000, true, 0b01, &pat_payload);
let mut pmt = vec![0x02];
let pmt_section_len: usize = 9 + 5 + 4;
pmt.push(0xB0 | ((pmt_section_len >> 8) & 0x0F) as u8);
pmt.push((pmt_section_len & 0xFF) as u8);
pmt.extend_from_slice(&[0x00, 0x01, 0xC1, 0x00, 0x00]);
pmt.extend_from_slice(&[0xE2, 0x00, 0xF0, 0x00]);
pmt.extend_from_slice(&[STREAM_TYPE_MPEG2_VIDEO, 0xE2, 0x00, 0xF0, 0x00]);
pmt.extend_from_slice(&[0u8; 4]);
let mut pmt_payload = vec![0u8];
pmt_payload.extend_from_slice(&pmt);
let pmt_pkt = ts_pkt(0x0100, true, 0b01, &pmt_payload);
let video_pes = {
let mut pes = vec![
0u8, 0u8, 1u8, 0xE0, 0u8, 0u8, 0x80, 0x80, 5, 0x21, 0x00, 0x01, 0x00, 0x01,
];
pes.extend_from_slice(&[0xAAu8; 16]);
pes
};
let mut video_pkt = ts_pkt(0x0200, true, 0b01, &video_pes);
video_pkt[3] = (video_pkt[3] & 0x3F) | (0x01 << 6);
let mut buf = Vec::new();
buf.extend_from_slice(&pat_pkt);
buf.extend_from_slice(&pmt_pkt);
buf.extend_from_slice(&video_pkt);
buf.extend_from_slice(&ts_pkt(0x1FFF, false, 0b01, &[]));
buf
}
#[test]
fn streaming_demuxer_drops_video_when_active_pid_is_scrambled() {
let buf = build_encrypted_ts();
let mut dem = demux_ts_streaming_init(&buf).expect("init");
let s = dem.next_video_sample().expect("call must not error");
assert!(
s.is_none(),
"encrypted TS → next_video_sample returns None on first call"
);
let s2 = dem.next_video_sample().expect("call must not error");
assert!(
s2.is_none(),
"encrypted TS → guard remains latched on subsequent calls"
);
}
}