PES pipeline: InputStream on DiscStream + IsoStream, MkvOutputStream
- DiscStream: next_frame() reads sectors → decrypts → demuxes → returns PesFrame - IsoStream: same pattern, reads from ISO file - MkvOutputStream: accepts PesFrame, writes MKV via MkvMuxer - InputStream trait: next_frame(), info(), codec_private(), headers_ready() - OutputStream trait: write_frame(), finish() - FileSectorReader: SectorReader backed by a file - PesFrame: track + pts + keyframe + data Old Read/Write IOStream impls preserved for backward compatibility. Next: replace pipe() in CLI to use PES pipeline.
This commit is contained in:
@@ -402,6 +402,26 @@ impl crate::pes::InputStream for DiscStream {
|
|||||||
fn info(&self) -> &DiscTitle {
|
fn info(&self) -> &DiscTitle {
|
||||||
&self.title
|
&self.title
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn codec_private(&self, track: usize) -> Option<Vec<u8>> {
|
||||||
|
let pid = self.pid_to_track.iter()
|
||||||
|
.find(|(_, idx)| *idx == track)
|
||||||
|
.map(|(pid, _)| *pid)?;
|
||||||
|
self.parsers.iter()
|
||||||
|
.find(|(p, _)| *p == pid)
|
||||||
|
.and_then(|(_, parser)| parser.codec_private())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn headers_ready(&self) -> bool {
|
||||||
|
for (idx, s) in self.title.streams.iter().enumerate() {
|
||||||
|
if let crate::disc::Stream::Video(v) = s {
|
||||||
|
if !v.secondary && self.codec_private(idx).is_none() {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
true
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Read for DiscStream {
|
impl Read for DiscStream {
|
||||||
|
|||||||
@@ -77,6 +77,11 @@ pub struct IsoStream {
|
|||||||
// Write side
|
// Write side
|
||||||
iso_writer: Option<IsoWriter<io::BufWriter<File>>>,
|
iso_writer: Option<IsoWriter<io::BufWriter<File>>>,
|
||||||
write_started: bool,
|
write_started: bool,
|
||||||
|
// PES output (for InputStream impl)
|
||||||
|
demuxer: Option<super::ts::TsDemuxer>,
|
||||||
|
parsers: Vec<(u16, Box<dyn super::codec::CodecParser>)>,
|
||||||
|
pending_frames: std::collections::VecDeque<crate::pes::PesFrame>,
|
||||||
|
pid_to_track: Vec<(u16, usize)>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl IsoStream {
|
impl IsoStream {
|
||||||
@@ -116,6 +121,21 @@ impl IsoStream {
|
|||||||
.collect();
|
.collect();
|
||||||
let sectors_remaining = extents.first().map(|e| e.1).unwrap_or(0);
|
let sectors_remaining = extents.first().map(|e| e.1).unwrap_or(0);
|
||||||
|
|
||||||
|
// Set up PES demux from title stream PIDs
|
||||||
|
let mut pids = Vec::new();
|
||||||
|
let mut parsers: Vec<(u16, Box<dyn super::codec::CodecParser>)> = Vec::new();
|
||||||
|
let mut pid_to_track = Vec::new();
|
||||||
|
for (i, s) in disc_title.streams.iter().enumerate() {
|
||||||
|
let (pid, codec) = match s {
|
||||||
|
crate::disc::Stream::Video(v) => (v.pid, v.codec),
|
||||||
|
crate::disc::Stream::Audio(a) => (a.pid, a.codec),
|
||||||
|
crate::disc::Stream::Subtitle(s) => (s.pid, s.codec),
|
||||||
|
};
|
||||||
|
pids.push(pid);
|
||||||
|
pid_to_track.push((pid, i));
|
||||||
|
parsers.push((pid, super::codec::parser_for_codec(codec)));
|
||||||
|
}
|
||||||
|
|
||||||
Ok(IsoStream {
|
Ok(IsoStream {
|
||||||
disc_title,
|
disc_title,
|
||||||
disc: Some(disc),
|
disc: Some(disc),
|
||||||
@@ -130,6 +150,10 @@ impl IsoStream {
|
|||||||
decrypt_keys,
|
decrypt_keys,
|
||||||
iso_writer: None,
|
iso_writer: None,
|
||||||
write_started: false,
|
write_started: false,
|
||||||
|
demuxer: if pids.is_empty() { None } else { Some(super::ts::TsDemuxer::new(&pids)) },
|
||||||
|
parsers,
|
||||||
|
pending_frames: std::collections::VecDeque::new(),
|
||||||
|
pid_to_track,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -154,6 +178,10 @@ impl IsoStream {
|
|||||||
eof: false,
|
eof: false,
|
||||||
iso_writer: Some(iso_writer),
|
iso_writer: Some(iso_writer),
|
||||||
write_started: false,
|
write_started: false,
|
||||||
|
demuxer: None,
|
||||||
|
parsers: Vec::new(),
|
||||||
|
pending_frames: std::collections::VecDeque::new(),
|
||||||
|
pid_to_track: Vec::new(),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -228,6 +256,76 @@ impl IsoStream {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl crate::pes::InputStream for IsoStream {
|
||||||
|
fn next_frame(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
|
||||||
|
// Return buffered frame
|
||||||
|
if let Some(frame) = self.pending_frames.pop_front() {
|
||||||
|
return Ok(Some(frame));
|
||||||
|
}
|
||||||
|
|
||||||
|
if self.eof {
|
||||||
|
return Ok(None);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Read until we produce at least one frame
|
||||||
|
loop {
|
||||||
|
if !self.read_next_batch()? {
|
||||||
|
self.eof = true;
|
||||||
|
return Ok(None);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Data in batch_buf is already decrypted by fill_next_batch
|
||||||
|
if let Some(ref mut demuxer) = self.demuxer {
|
||||||
|
let packets = demuxer.feed(&self.batch_buf[..self.buf_len]);
|
||||||
|
for pes in &packets {
|
||||||
|
if let Some((pid_idx, _)) = self.pid_to_track.iter().enumerate()
|
||||||
|
.find(|(_, (pid, _))| *pid == pes.pid)
|
||||||
|
{
|
||||||
|
let track_idx = self.pid_to_track[pid_idx].1;
|
||||||
|
if let Some((_, parser)) = self.parsers.iter_mut()
|
||||||
|
.find(|(pid, _)| *pid == pes.pid)
|
||||||
|
{
|
||||||
|
for frame in parser.parse(pes) {
|
||||||
|
self.pending_frames.push_back(
|
||||||
|
crate::pes::PesFrame::from_codec_frame(track_idx, frame)
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Some(frame) = self.pending_frames.pop_front() {
|
||||||
|
return Ok(Some(frame));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn info(&self) -> &crate::disc::DiscTitle {
|
||||||
|
&self.disc_title
|
||||||
|
}
|
||||||
|
|
||||||
|
fn codec_private(&self, track: usize) -> Option<Vec<u8>> {
|
||||||
|
let pid = self.pid_to_track.iter()
|
||||||
|
.find(|(_, idx)| *idx == track)
|
||||||
|
.map(|(pid, _)| *pid)?;
|
||||||
|
self.parsers.iter()
|
||||||
|
.find(|(p, _)| *p == pid)
|
||||||
|
.and_then(|(_, parser)| parser.codec_private())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn headers_ready(&self) -> bool {
|
||||||
|
for (idx, s) in self.disc_title.streams.iter().enumerate() {
|
||||||
|
if let crate::disc::Stream::Video(v) = s {
|
||||||
|
if !v.secondary && self.codec_private(idx).is_none() {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
impl IOStream for IsoStream {
|
impl IOStream for IsoStream {
|
||||||
fn info(&self) -> &DiscTitle {
|
fn info(&self) -> &DiscTitle {
|
||||||
&self.disc_title
|
&self.disc_title
|
||||||
|
|||||||
@@ -0,0 +1,66 @@
|
|||||||
|
//! MKV output stream — accepts PES frames, writes Matroska container.
|
||||||
|
//!
|
||||||
|
//! Implements OutputStream. Takes PES frames directly — no TS demuxing needed.
|
||||||
|
//! Creates the MKV muxer once codec_private is provided for all tracks.
|
||||||
|
|
||||||
|
use super::mkv::{MkvMuxer, MkvTrack};
|
||||||
|
use super::WriteSeek;
|
||||||
|
use crate::disc::DiscTitle;
|
||||||
|
use crate::pes::{OutputStream, PesFrame};
|
||||||
|
use std::io;
|
||||||
|
|
||||||
|
pub struct MkvOutputStream {
|
||||||
|
muxer: Option<MkvMuxer<Box<dyn WriteSeek>>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl MkvOutputStream {
|
||||||
|
/// Create an MKV output stream.
|
||||||
|
/// `codec_privates` provides initialization data per track (from InputStream).
|
||||||
|
/// Tracks without codec_private get None.
|
||||||
|
pub fn create(
|
||||||
|
writer: Box<dyn WriteSeek>,
|
||||||
|
title: &DiscTitle,
|
||||||
|
codec_privates: &[Option<Vec<u8>>],
|
||||||
|
) -> io::Result<Self> {
|
||||||
|
let mut tracks = Vec::new();
|
||||||
|
for (idx, s) in title.streams.iter().enumerate() {
|
||||||
|
let mut track = match s {
|
||||||
|
crate::disc::Stream::Video(v) => MkvTrack::video(v),
|
||||||
|
crate::disc::Stream::Audio(a) => MkvTrack::audio(a),
|
||||||
|
crate::disc::Stream::Subtitle(s) => MkvTrack::subtitle(s),
|
||||||
|
};
|
||||||
|
if let Some(cp) = codec_privates.get(idx).and_then(|c| c.as_ref()) {
|
||||||
|
track.codec_private = Some(cp.clone());
|
||||||
|
}
|
||||||
|
tracks.push(track);
|
||||||
|
}
|
||||||
|
|
||||||
|
let muxer = MkvMuxer::new_with_chapters(
|
||||||
|
writer,
|
||||||
|
&tracks,
|
||||||
|
Some(&title.playlist),
|
||||||
|
title.duration_secs,
|
||||||
|
&title.chapters,
|
||||||
|
)?;
|
||||||
|
|
||||||
|
Ok(Self { muxer: Some(muxer) })
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl OutputStream for MkvOutputStream {
|
||||||
|
fn write_frame(&mut self, frame: &PesFrame) -> io::Result<()> {
|
||||||
|
if let Some(ref mut muxer) = self.muxer {
|
||||||
|
muxer.write_frame(frame.track, frame.pts, frame.keyframe, &frame.data)
|
||||||
|
} else {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn finish(&mut self) -> io::Result<()> {
|
||||||
|
if let Some(muxer) = self.muxer.take() {
|
||||||
|
muxer.finish()
|
||||||
|
} else {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -28,6 +28,7 @@ pub mod lookahead;
|
|||||||
mod m2ts;
|
mod m2ts;
|
||||||
pub mod meta;
|
pub mod meta;
|
||||||
pub mod mkv;
|
pub mod mkv;
|
||||||
|
pub mod mkvout;
|
||||||
mod mkvstream;
|
mod mkvstream;
|
||||||
pub mod network;
|
pub mod network;
|
||||||
pub mod null;
|
pub mod null;
|
||||||
|
|||||||
@@ -39,6 +39,13 @@ pub trait InputStream {
|
|||||||
|
|
||||||
/// Stream metadata (tracks, duration, etc).
|
/// Stream metadata (tracks, duration, etc).
|
||||||
fn info(&self) -> &crate::disc::DiscTitle;
|
fn info(&self) -> &crate::disc::DiscTitle;
|
||||||
|
|
||||||
|
/// Codec initialization data for a track (SPS/PPS for HEVC, etc).
|
||||||
|
/// Returns None until enough frames have been parsed.
|
||||||
|
fn codec_private(&self, track: usize) -> Option<Vec<u8>>;
|
||||||
|
|
||||||
|
/// True when codec_private is available for all video tracks.
|
||||||
|
fn headers_ready(&self) -> bool;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Output stream — consumes PES frames to any destination.
|
/// Output stream — consumes PES frames to any destination.
|
||||||
|
|||||||
Reference in New Issue
Block a user