nihav-core/src/demuxers/mod.rs

   1 //! Demuxer definitions.
   2 pub use crate::frame::*;
   3 pub use crate::io::byteio::*;
   4
   5 /// A list specifying general demuxing errors.
   6 #[derive(Debug,Clone,Copy,PartialEq)]
   7 #[allow(dead_code)]
   8 pub enum DemuxerError {
   9     /// Reader got to end of stream.
  10     EOF,
  11     /// Demuxer encountered empty container.
  12     NoSuchInput,
  13     /// Demuxer encountered invalid input data.
  14     InvalidData,
  15     /// Data reading error.
  16     IOError,
  17     /// Feature is not implemented.
  18     NotImplemented,
  19     /// Allocation failed.
  20     MemoryError,
  21     /// The operation should be repeated.
  22     TryAgain,
  23     /// Seeking failed.
  24     SeekError,
  25     /// Operation cannot succeed in principle (e.g. seeking in a format not supporting seeking).
  26     NotPossible,
  27 }
  28
  29 /// A specialised `Result` type for demuxing operations.
  30 pub type DemuxerResult<T> = Result<T, DemuxerError>;
  31
  32 /// A trait for demuxing operations.
  33 pub trait DemuxCore<'a> {
  34     /// Opens the input stream, reads required headers and prepares everything for packet demuxing.
  35     fn open(&mut self, strmgr: &mut StreamManager, seek_idx: &mut SeekIndex) -> DemuxerResult<()>;
  36     /// Demuxes a packet.
  37     fn get_frame(&mut self, strmgr: &mut StreamManager) -> DemuxerResult<NAPacket>;
  38     /// Seeks to the requested time.
  39     fn seek(&mut self, time: u64, seek_idx: &SeekIndex) -> DemuxerResult<()>;
  40 }
  41
  42 /// An auxiliary trait to make bytestream reader read packet data.
  43 pub trait NAPacketReader {
  44     /// Reads input and constructs a packet containing it.
  45     fn read_packet(&mut self, str: NAStreamRef, ts: NATimeInfo, keyframe: bool, size: usize) -> DemuxerResult<NAPacket>;
  46     /// Reads input into already existing packet.
  47     fn fill_packet(&mut self, pkt: &mut NAPacket) -> DemuxerResult<()>;
  48 }
  49
  50 impl<'a> NAPacketReader for ByteReader<'a> {
  51     fn read_packet(&mut self, str: NAStreamRef, ts: NATimeInfo, kf: bool, size: usize) -> DemuxerResult<NAPacket> {
  52         let mut buf: Vec<u8> = Vec::with_capacity(size);
  53         if buf.capacity() < size { return Err(DemuxerError::MemoryError); }
  54         buf.resize(size, 0);
  55         self.read_buf(buf.as_mut_slice())?;
  56         let pkt = NAPacket::new(str, ts, kf, buf);
  57         Ok(pkt)
  58     }
  59     fn fill_packet(&mut self, pkt: &mut NAPacket) -> DemuxerResult<()> {
  60         let mut refbuf = pkt.get_buffer();
  61         let buf = refbuf.as_mut().unwrap();
  62         self.read_buf(buf.as_mut_slice())?;
  63         Ok(())
  64     }
  65 }
  66
  67 /// An auxiliary structure for operations with individual streams inside the container.
  68 #[derive(Default)]
  69 pub struct StreamManager {
  70     streams: Vec<NAStreamRef>,
  71     ignored: Vec<bool>,
  72     no_ign:  bool,
  73 }
  74
  75 impl StreamManager {
  76     /// Constructs a new instance of `StreamManager`.
  77     pub fn new() -> Self {
  78         StreamManager {
  79             streams: Vec::new(),
  80             ignored: Vec::new(),
  81             no_ign:  true,
  82         }
  83     }
  84     /// Returns stream iterator.
  85     pub fn iter(&self) -> StreamIter { StreamIter::new(&self.streams) }
  86
  87     /// Adds a new stream.
  88     pub fn add_stream(&mut self, stream: NAStream) -> Option<usize> {
  89         let stream_num = self.streams.len();
  90         let mut str = stream.clone();
  91         str.set_num(stream_num);
  92         self.streams.push(str.into_ref());
  93         self.ignored.push(false);
  94         Some(stream_num)
  95     }
  96     /// Returns stream with the requested index.
  97     pub fn get_stream(&self, idx: usize) -> Option<NAStreamRef> {
  98         if idx < self.streams.len() {
  99             Some(self.streams[idx].clone())
 100         } else {
 101             None
 102         }
 103     }
 104     /// Returns stream with the requested stream ID.
 105     pub fn get_stream_by_id(&self, id: u32) -> Option<NAStreamRef> {
 106         for i in 0..self.streams.len() {
 107             if self.streams[i].get_id() == id {
 108                 return Some(self.streams[i].clone());
 109             }
 110         }
 111         None
 112     }
 113     /// Returns the number of known streams.
 114     pub fn get_num_streams(&self) -> usize { self.streams.len() }
 115     /// Reports whether the stream is marked as ignored.
 116     pub fn is_ignored(&self, idx: usize) -> bool {
 117         if self.no_ign {
 118             true
 119         } else if idx < self.ignored.len() {
 120             self.ignored[idx]
 121         } else {
 122             false
 123         }
 124     }
 125     /// Reports whether the stream with certain ID is marked as ignored.
 126     pub fn is_ignored_id(&self, id: u32) -> bool {
 127         for i in 0..self.streams.len() {
 128             if self.streams[i].get_id() == id {
 129                 return self.ignored[i];
 130             }
 131         }
 132         false
 133     }
 134     /// Marks requested stream as ignored.
 135     pub fn set_ignored(&mut self, idx: usize) {
 136         if idx < self.ignored.len() {
 137             self.ignored[idx] = true;
 138             self.no_ign = false;
 139         }
 140     }
 141     /// Clears the ignored mark for the requested stream.
 142     pub fn set_unignored(&mut self, idx: usize) {
 143         if idx < self.ignored.len() {
 144             self.ignored[idx] = false;
 145         }
 146     }
 147 }
 148
 149 /// Stream iterator.
 150 pub struct StreamIter<'a> {
 151     streams:    &'a [NAStreamRef],
 152     pos:        usize,
 153 }
 154
 155 impl<'a> StreamIter<'a> {
 156     /// Constructs a new instance of `StreamIter`.
 157     pub fn new(streams: &'a [NAStreamRef]) -> Self {
 158         StreamIter { streams, pos: 0 }
 159     }
 160 }
 161
 162 impl<'a> Iterator for StreamIter<'a> {
 163     type Item = NAStreamRef;
 164
 165     fn next(&mut self) -> Option<Self::Item> {
 166         if self.pos >= self.streams.len() { return None; }
 167         let ret = self.streams[self.pos].clone();
 168         self.pos += 1;
 169         Some(ret)
 170     }
 171 }
 172
 173 /// Seeking modes.
 174 #[derive(Clone,Copy,PartialEq)]
 175 pub enum SeekIndexMode {
 176     /// No seeking index present.
 177     None,
 178     /// Seeking index is present.
 179     Present,
 180     /// Seeking index should be constructed by the demuxer if possible.
 181     Automatic,
 182 }
 183
 184 impl Default for SeekIndexMode {
 185     fn default() -> Self { SeekIndexMode::None }
 186 }
 187
 188 /// A structure holding seeking information.
 189 #[derive(Clone,Copy,Default)]
 190 pub struct SeekEntry {
 191     /// Time in milliseconds.
 192     pub time:   u64,
 193     /// PTS
 194     pub pts:    u64,
 195     /// Position in file.
 196     pub pos:    u64,
 197 }
 198
 199 /// Seeking information for individual streams.
 200 #[derive(Clone)]
 201 pub struct StreamSeekInfo {
 202     /// Stream ID.
 203     pub id:         u32,
 204     /// Index is present.
 205     pub filled:     bool,
 206     /// Packet seeking information.
 207     pub entries:    Vec<SeekEntry>,
 208 }
 209
 210 impl StreamSeekInfo {
 211     /// Constructs a new `StreamSeekInfo` instance.
 212     pub fn new(id: u32) -> Self {
 213         Self {
 214             id,
 215             filled:     false,
 216             entries:    Vec::new(),
 217         }
 218     }
 219     /// Adds new seeking point.
 220     pub fn add_entry(&mut self, entry: SeekEntry) {
 221         self.entries.push(entry);
 222     }
 223     /// Searches for an appropriate seek position before requested time.
 224     pub fn find_pos(&self, time: u64) -> Option<SeekEntry> {
 225         if !self.entries.is_empty() {
 226 // todo something faster like binary search
 227             let mut cand = None;
 228             for entry in self.entries.iter() {
 229                 if entry.time <= time {
 230                     cand = Some(*entry);
 231                 } else {
 232                     break;
 233                 }
 234             }
 235             cand
 236         } else {
 237             None
 238         }
 239     }
 240 }
 241
 242 /// Structure for holding seeking point search results.
 243 #[derive(Clone,Copy,Default)]
 244 pub struct SeekIndexResult {
 245     /// Packet PTS.
 246     pub pts:        u64,
 247     /// Position in file.
 248     pub pos:        u64,
 249     /// Stream ID.
 250     pub str_id:     u32,
 251 }
 252
 253 /// Seek information for the whole container.
 254 #[derive(Default)]
 255 pub struct SeekIndex {
 256     /// Seek information for individual streams.
 257     pub seek_info:  Vec<StreamSeekInfo>,
 258     /// Seeking index mode.
 259     pub mode:       SeekIndexMode,
 260     /// Ignore index flag.
 261     pub skip_index: bool,
 262 }
 263
 264 impl SeekIndex {
 265     /// Constructs a new `SeekIndex` instance.
 266     pub fn new() -> Self { Self::default() }
 267     pub fn add_stream(&mut self, id: u32) -> usize {
 268         let ret = self.stream_id_to_index(id);
 269         if ret.is_none() {
 270             self.seek_info.push(StreamSeekInfo::new(id));
 271             self.seek_info.len() - 1
 272         } else {
 273             ret.unwrap()
 274         }
 275     }
 276     /// Adds a new stream to the index.
 277     pub fn stream_id_to_index(&self, id: u32) -> Option<usize> {
 278         for (idx, str) in self.seek_info.iter().enumerate() {
 279             if str.id == id {
 280                 return Some(idx);
 281             }
 282         }
 283         None
 284     }
 285     /// Returns stream reference for provided stream ID.
 286     pub fn get_stream_index(&mut self, id: u32) -> Option<&mut StreamSeekInfo> {
 287         for str in self.seek_info.iter_mut() {
 288             if str.id == id {
 289                 return Some(str);
 290             }
 291         }
 292         None
 293     }
 294     /// Adds seeking information to the index.
 295     pub fn add_entry(&mut self, id: u32, entry: SeekEntry) {
 296         let mut idx = self.stream_id_to_index(id);
 297         if idx.is_none() {
 298             idx = Some(self.add_stream(id));
 299         }
 300         self.seek_info[idx.unwrap()].add_entry(entry);
 301         self.seek_info[idx.unwrap()].filled = true;
 302     }
 303     /// Searches for a seek position before requested time.
 304     pub fn find_pos(&self, time: u64) -> Option<SeekIndexResult> {
 305         let mut cand = None;
 306         for str in self.seek_info.iter() {
 307             if !str.filled { continue; }
 308             let res = str.find_pos(time);
 309             if res.is_none() { continue; }
 310             let res = res.unwrap();
 311             if cand.is_none() {
 312                 cand = Some(SeekIndexResult { pts: res.pts, pos: res.pos, str_id: str.id });
 313             } else if let Some(entry) = cand {
 314                 if res.pos < entry.pos {
 315                     cand = Some(SeekIndexResult { pts: res.pts, pos: res.pos, str_id: str.id });
 316                 }
 317             }
 318         }
 319         cand
 320     }
 321 }
 322
 323 /// Demuxer structure with auxiliary data.
 324 pub struct Demuxer<'a> {
 325     dmx:        Box<dyn DemuxCore<'a> + 'a>,
 326     streams:    StreamManager,
 327     seek_idx:   SeekIndex,
 328 }
 329
 330 impl<'a> Demuxer<'a> {
 331     /// Constructs a new `Demuxer` instance.
 332     fn new(dmx: Box<dyn DemuxCore<'a> + 'a>, str: StreamManager, seek_idx: SeekIndex) -> Self {
 333         Demuxer {
 334             dmx,
 335             streams:    str,
 336             seek_idx,
 337         }
 338     }
 339     /// Returns a stream reference by its number.
 340     pub fn get_stream(&self, idx: usize) -> Option<NAStreamRef> {
 341         self.streams.get_stream(idx)
 342     }
 343     /// Returns a stream reference by its ID.
 344     pub fn get_stream_by_id(&self, id: u32) -> Option<NAStreamRef> {
 345         self.streams.get_stream_by_id(id)
 346     }
 347     /// Reports the total number of streams.
 348     pub fn get_num_streams(&self) -> usize {
 349         self.streams.get_num_streams()
 350     }
 351     /// Returns a reference to the internal stream manager.
 352     pub fn get_stream_manager(&self) -> &StreamManager {
 353         &self.streams
 354     }
 355     /// Returns an iterator over streams.
 356     pub fn get_streams(&self) -> StreamIter {
 357         self.streams.iter()
 358     }
 359     /// Returns 'ignored' marker for requested stream.
 360     pub fn is_ignored_stream(&self, idx: usize) -> bool {
 361         self.streams.is_ignored(idx)
 362     }
 363     /// Sets 'ignored' marker for requested stream.
 364     pub fn set_ignored_stream(&mut self, idx: usize) {
 365         self.streams.set_ignored(idx)
 366     }
 367     /// Clears 'ignored' marker for requested stream.
 368     pub fn set_unignored_stream(&mut self, idx: usize) {
 369         self.streams.set_unignored(idx)
 370     }
 371
 372     /// Demuxes a new packet from the container.
 373     pub fn get_frame(&mut self) -> DemuxerResult<NAPacket> {
 374         loop {
 375             let res = self.dmx.get_frame(&mut self.streams);
 376             if self.streams.no_ign || res.is_err() { return res; }
 377             let res = res.unwrap();
 378             let idx = res.get_stream().get_num();
 379             if !self.is_ignored_stream(idx) {
 380                 return Ok(res);
 381             }
 382         }
 383     }
 384     /// Seeks to the requested time (in milliseconds) if possible.
 385     pub fn seek(&mut self, time: u64) -> DemuxerResult<()> {
 386         if self.seek_idx.skip_index {
 387             return Err(DemuxerError::NotPossible);
 388         }
 389         self.dmx.seek(time, &self.seek_idx)
 390     }
 391     /// Returns internal seek index.
 392     pub fn get_seek_index(&self) -> &SeekIndex {
 393         &self.seek_idx
 394     }
 395 }
 396
 397 impl From<ByteIOError> for DemuxerError {
 398     fn from(_: ByteIOError) -> Self { DemuxerError::IOError }
 399 }
 400
 401 /// The trait for creating demuxers.
 402 pub trait DemuxerCreator {
 403     /// Creates new demuxer instance that will use `ByteReader` source as an input.
 404     fn new_demuxer<'a>(&self, br: &'a mut ByteReader<'a>) -> Box<dyn DemuxCore<'a> + 'a>;
 405     /// Returns the name of current demuxer creator (equal to the container name it can demux).
 406     fn get_name(&self) -> &'static str;
 407 }
 408
 409 /// Creates demuxer for a provided bytestream.
 410 pub fn create_demuxer<'a>(dmxcr: &DemuxerCreator, br: &'a mut ByteReader<'a>) -> DemuxerResult<Demuxer<'a>> {
 411     let mut dmx = dmxcr.new_demuxer(br);
 412     let mut str = StreamManager::new();
 413     let mut seek_idx = SeekIndex::new();
 414     dmx.open(&mut str, &mut seek_idx)?;
 415     Ok(Demuxer::new(dmx, str, seek_idx))
 416 }
 417
 418 /// List of registered demuxers.
 419 #[derive(Default)]
 420 pub struct RegisteredDemuxers {
 421     dmxs:   Vec<&'static DemuxerCreator>,
 422 }
 423
 424 impl RegisteredDemuxers {
 425     /// Constructs a new `RegisteredDemuxers` instance.
 426     pub fn new() -> Self {
 427         Self { dmxs: Vec::new() }
 428     }
 429     /// Registers a new demuxer.
 430     pub fn add_demuxer(&mut self, dmx: &'static DemuxerCreator) {
 431         self.dmxs.push(dmx);
 432     }
 433     /// Searches for a demuxer that supports requested container format.
 434     pub fn find_demuxer(&self, name: &str) -> Option<&DemuxerCreator> {
 435         for &dmx in self.dmxs.iter() {
 436             if dmx.get_name() == name {
 437                 return Some(dmx);
 438             }
 439         }
 440         None
 441     }
 442 }