]> git.nihav.org Git - nihav.git/blob - nihav-core/src/demuxers/mod.rs
add interfaces for raw data stream handling
[nihav.git] / nihav-core / src / demuxers / mod.rs
1 //! Demuxer definitions.
2 pub use crate::frame::*;
3 pub use crate::io::byteio::*;
4 pub use crate::options::*;
5
6 /// A list specifying general demuxing errors.
7 #[derive(Debug,Clone,Copy,PartialEq)]
8 #[allow(dead_code)]
9 pub enum DemuxerError {
10 /// Reader got to end of stream.
11 EOF,
12 /// Demuxer encountered empty container.
13 NoSuchInput,
14 /// Demuxer encountered invalid input data.
15 InvalidData,
16 /// Data reading error.
17 IOError,
18 /// Feature is not implemented.
19 NotImplemented,
20 /// Allocation failed.
21 MemoryError,
22 /// The operation should be repeated.
23 TryAgain,
24 /// Seeking failed.
25 SeekError,
26 /// Operation cannot succeed in principle (e.g. seeking in a format not supporting seeking).
27 NotPossible,
28 }
29
30 /// A specialised `Result` type for demuxing operations.
31 pub type DemuxerResult<T> = Result<T, DemuxerError>;
32
33 /// A trait for demuxing operations.
34 pub trait DemuxCore<'a>: NAOptionHandler {
35 /// Opens the input stream, reads required headers and prepares everything for packet demuxing.
36 fn open(&mut self, strmgr: &mut StreamManager, seek_idx: &mut SeekIndex) -> DemuxerResult<()>;
37 /// Demuxes a packet.
38 fn get_frame(&mut self, strmgr: &mut StreamManager) -> DemuxerResult<NAPacket>;
39 /// Seeks to the requested time.
40 fn seek(&mut self, time: NATimePoint, seek_idx: &SeekIndex) -> DemuxerResult<()>;
41 /// Returns container duration in milliseconds (zero if not available).
42 fn get_duration(&self) -> u64;
43 }
44
45 /// An auxiliary trait to make bytestream reader read packet data.
46 pub trait NAPacketReader {
47 /// Reads input and constructs a packet containing it.
48 fn read_packet(&mut self, str: NAStreamRef, ts: NATimeInfo, keyframe: bool, size: usize) -> DemuxerResult<NAPacket>;
49 /// Reads input into already existing packet.
50 fn fill_packet(&mut self, pkt: &mut NAPacket) -> DemuxerResult<()>;
51 }
52
53 impl<'a> NAPacketReader for ByteReader<'a> {
54 fn read_packet(&mut self, str: NAStreamRef, ts: NATimeInfo, kf: bool, size: usize) -> DemuxerResult<NAPacket> {
55 let mut buf: Vec<u8> = Vec::with_capacity(size);
56 if buf.capacity() < size { return Err(DemuxerError::MemoryError); }
57 buf.resize(size, 0);
58 self.read_buf(buf.as_mut_slice())?;
59 let pkt = NAPacket::new(str, ts, kf, buf);
60 Ok(pkt)
61 }
62 fn fill_packet(&mut self, pkt: &mut NAPacket) -> DemuxerResult<()> {
63 let mut refbuf = pkt.get_buffer();
64 let buf = refbuf.as_mut().unwrap();
65 self.read_buf(buf.as_mut_slice())?;
66 Ok(())
67 }
68 }
69
70 /// An auxiliary structure for operations with individual streams inside the container.
71 #[derive(Default)]
72 pub struct StreamManager {
73 streams: Vec<NAStreamRef>,
74 ignored: Vec<bool>,
75 no_ign: bool,
76 }
77
78 impl StreamManager {
79 /// Constructs a new instance of `StreamManager`.
80 pub fn new() -> Self {
81 StreamManager {
82 streams: Vec::new(),
83 ignored: Vec::new(),
84 no_ign: true,
85 }
86 }
87 /// Returns stream iterator.
88 pub fn iter(&self) -> StreamIter { StreamIter::new(&self.streams) }
89
90 /// Adds a new stream.
91 pub fn add_stream(&mut self, stream: NAStream) -> Option<usize> {
92 let stream_num = self.streams.len();
93 let mut str = stream;
94 str.set_num(stream_num);
95 self.streams.push(str.into_ref());
96 self.ignored.push(false);
97 Some(stream_num)
98 }
99 /// Adds a new stream from reference-counted object.
100 pub fn add_stream_ref(&mut self, stream: NAStreamRef) -> Option<usize> {
101 let stream_num = self.streams.len();
102 self.streams.push(stream);
103 self.ignored.push(false);
104 Some(stream_num)
105 }
106 /// Returns stream with the requested index.
107 pub fn get_stream(&self, idx: usize) -> Option<NAStreamRef> {
108 if idx < self.streams.len() {
109 Some(self.streams[idx].clone())
110 } else {
111 None
112 }
113 }
114 /// Returns stream with the requested stream ID.
115 pub fn get_stream_by_id(&self, id: u32) -> Option<NAStreamRef> {
116 for i in 0..self.streams.len() {
117 if self.streams[i].get_id() == id {
118 return Some(self.streams[i].clone());
119 }
120 }
121 None
122 }
123 /// Returns the number of known streams.
124 pub fn get_num_streams(&self) -> usize { self.streams.len() }
125 /// Reports whether the stream is marked as ignored.
126 pub fn is_ignored(&self, idx: usize) -> bool {
127 if self.no_ign {
128 true
129 } else if idx < self.ignored.len() {
130 self.ignored[idx]
131 } else {
132 false
133 }
134 }
135 /// Reports whether the stream with certain ID is marked as ignored.
136 pub fn is_ignored_id(&self, id: u32) -> bool {
137 for i in 0..self.streams.len() {
138 if self.streams[i].get_id() == id {
139 return self.ignored[i];
140 }
141 }
142 false
143 }
144 /// Marks requested stream as ignored.
145 pub fn set_ignored(&mut self, idx: usize) {
146 if idx < self.ignored.len() {
147 self.ignored[idx] = true;
148 self.no_ign = false;
149 }
150 }
151 /// Clears the ignored mark for the requested stream.
152 pub fn set_unignored(&mut self, idx: usize) {
153 if idx < self.ignored.len() {
154 self.ignored[idx] = false;
155 }
156 }
157 }
158
159 /// Stream iterator.
160 pub struct StreamIter<'a> {
161 streams: &'a [NAStreamRef],
162 pos: usize,
163 }
164
165 impl<'a> StreamIter<'a> {
166 /// Constructs a new instance of `StreamIter`.
167 pub fn new(streams: &'a [NAStreamRef]) -> Self {
168 StreamIter { streams, pos: 0 }
169 }
170 }
171
172 impl<'a> Iterator for StreamIter<'a> {
173 type Item = NAStreamRef;
174
175 fn next(&mut self) -> Option<Self::Item> {
176 if self.pos >= self.streams.len() { return None; }
177 let ret = self.streams[self.pos].clone();
178 self.pos += 1;
179 Some(ret)
180 }
181 }
182
183 /// Seeking modes.
184 #[derive(Clone,Copy,PartialEq)]
185 pub enum SeekIndexMode {
186 /// No seeking index present.
187 None,
188 /// Seeking index is present.
189 Present,
190 /// Seeking index should be constructed by the demuxer if possible.
191 Automatic,
192 }
193
194 impl Default for SeekIndexMode {
195 fn default() -> Self { SeekIndexMode::None }
196 }
197
198 /// A structure holding seeking information.
199 #[derive(Clone,Copy,Default)]
200 pub struct SeekEntry {
201 /// Time in milliseconds.
202 pub time: u64,
203 /// PTS
204 pub pts: u64,
205 /// Position in file.
206 pub pos: u64,
207 }
208
209 /// Seeking information for individual streams.
210 #[derive(Clone)]
211 pub struct StreamSeekInfo {
212 /// Stream ID.
213 pub id: u32,
214 /// Index is present.
215 pub filled: bool,
216 /// Packet seeking information.
217 pub entries: Vec<SeekEntry>,
218 }
219
220 impl StreamSeekInfo {
221 /// Constructs a new `StreamSeekInfo` instance.
222 pub fn new(id: u32) -> Self {
223 Self {
224 id,
225 filled: false,
226 entries: Vec::new(),
227 }
228 }
229 /// Adds new seeking point.
230 pub fn add_entry(&mut self, entry: SeekEntry) {
231 self.entries.push(entry);
232 }
233 /// Searches for an appropriate seek position before requested time.
234 pub fn find_pos(&self, time: NATimePoint) -> Option<SeekEntry> {
235 if time == NATimePoint::None {
236 return None;
237 }
238 if !self.entries.is_empty() {
239 // todo something faster like binary search
240 let mut cand = None;
241 for entry in self.entries.iter() {
242 match time {
243 NATimePoint::Milliseconds(ms) => {
244 if entry.time <= ms {
245 cand = Some(*entry);
246 } else {
247 break;
248 }
249 },
250 NATimePoint::PTS(pts) => {
251 if entry.pts <= pts {
252 cand = Some(*entry);
253 } else {
254 break;
255 }
256 },
257 NATimePoint::None => unreachable!(),
258 };
259 }
260 cand
261 } else {
262 None
263 }
264 }
265 }
266
267 /// Structure for holding seeking point search results.
268 #[derive(Clone,Copy,Default)]
269 pub struct SeekIndexResult {
270 /// Packet PTS.
271 pub pts: u64,
272 /// Position in file.
273 pub pos: u64,
274 /// Stream ID.
275 pub str_id: u32,
276 }
277
278 /// Seek information for the whole container.
279 #[derive(Default)]
280 pub struct SeekIndex {
281 /// Seek information for individual streams.
282 pub seek_info: Vec<StreamSeekInfo>,
283 /// Seeking index mode.
284 pub mode: SeekIndexMode,
285 /// Ignore index flag.
286 pub skip_index: bool,
287 }
288
289 impl SeekIndex {
290 /// Constructs a new `SeekIndex` instance.
291 pub fn new() -> Self { Self::default() }
292 pub fn add_stream(&mut self, id: u32) -> usize {
293 let ret = self.stream_id_to_index(id);
294 if let Some(res) = ret {
295 res
296 } else {
297 self.seek_info.push(StreamSeekInfo::new(id));
298 self.seek_info.len() - 1
299 }
300 }
301 /// Adds a new stream to the index.
302 pub fn stream_id_to_index(&self, id: u32) -> Option<usize> {
303 for (idx, str) in self.seek_info.iter().enumerate() {
304 if str.id == id {
305 return Some(idx);
306 }
307 }
308 None
309 }
310 /// Returns stream reference for provided stream ID.
311 pub fn get_stream_index(&mut self, id: u32) -> Option<&mut StreamSeekInfo> {
312 for str in self.seek_info.iter_mut() {
313 if str.id == id {
314 return Some(str);
315 }
316 }
317 None
318 }
319 /// Adds seeking information to the index.
320 pub fn add_entry(&mut self, id: u32, entry: SeekEntry) {
321 let mut idx = self.stream_id_to_index(id);
322 if idx.is_none() {
323 idx = Some(self.add_stream(id));
324 }
325 self.seek_info[idx.unwrap()].add_entry(entry);
326 self.seek_info[idx.unwrap()].filled = true;
327 }
328 /// Searches for a seek position before requested time.
329 pub fn find_pos(&self, time: NATimePoint) -> Option<SeekIndexResult> {
330 let mut cand = None;
331 for str in self.seek_info.iter() {
332 if !str.filled { continue; }
333 let res = str.find_pos(time);
334 if res.is_none() { continue; }
335 let res = res.unwrap();
336 if cand.is_none() {
337 cand = Some(SeekIndexResult { pts: res.pts, pos: res.pos, str_id: str.id });
338 } else if let Some(entry) = cand {
339 if res.pos < entry.pos {
340 cand = Some(SeekIndexResult { pts: res.pts, pos: res.pos, str_id: str.id });
341 }
342 }
343 }
344 cand
345 }
346 }
347
348 /// Demuxer structure with auxiliary data.
349 pub struct Demuxer<'a> {
350 dmx: Box<dyn DemuxCore<'a> + 'a>,
351 streams: StreamManager,
352 seek_idx: SeekIndex,
353 }
354
355 impl<'a> Demuxer<'a> {
356 /// Constructs a new `Demuxer` instance.
357 fn new(dmx: Box<dyn DemuxCore<'a> + 'a>, str: StreamManager, seek_idx: SeekIndex) -> Self {
358 Demuxer {
359 dmx,
360 streams: str,
361 seek_idx,
362 }
363 }
364 /// Returns a stream reference by its number.
365 pub fn get_stream(&self, idx: usize) -> Option<NAStreamRef> {
366 self.streams.get_stream(idx)
367 }
368 /// Returns a stream reference by its ID.
369 pub fn get_stream_by_id(&self, id: u32) -> Option<NAStreamRef> {
370 self.streams.get_stream_by_id(id)
371 }
372 /// Reports the total number of streams.
373 pub fn get_num_streams(&self) -> usize {
374 self.streams.get_num_streams()
375 }
376 /// Returns a reference to the internal stream manager.
377 pub fn get_stream_manager(&self) -> &StreamManager {
378 &self.streams
379 }
380 /// Returns an iterator over streams.
381 pub fn get_streams(&self) -> StreamIter {
382 self.streams.iter()
383 }
384 /// Returns 'ignored' marker for requested stream.
385 pub fn is_ignored_stream(&self, idx: usize) -> bool {
386 self.streams.is_ignored(idx)
387 }
388 /// Sets 'ignored' marker for requested stream.
389 pub fn set_ignored_stream(&mut self, idx: usize) {
390 self.streams.set_ignored(idx)
391 }
392 /// Clears 'ignored' marker for requested stream.
393 pub fn set_unignored_stream(&mut self, idx: usize) {
394 self.streams.set_unignored(idx)
395 }
396
397 /// Demuxes a new packet from the container.
398 pub fn get_frame(&mut self) -> DemuxerResult<NAPacket> {
399 loop {
400 let res = self.dmx.get_frame(&mut self.streams);
401 if self.streams.no_ign || res.is_err() { return res; }
402 let res = res.unwrap();
403 let idx = res.get_stream().get_num();
404 if !self.is_ignored_stream(idx) {
405 return Ok(res);
406 }
407 }
408 }
409 /// Seeks to the requested time if possible.
410 pub fn seek(&mut self, time: NATimePoint) -> DemuxerResult<()> {
411 if self.seek_idx.skip_index {
412 return Err(DemuxerError::NotPossible);
413 }
414 self.dmx.seek(time, &self.seek_idx)
415 }
416 /// Returns internal seek index.
417 pub fn get_seek_index(&self) -> &SeekIndex {
418 &self.seek_idx
419 }
420 /// Returns media duration reported by container or its streams.
421 ///
422 /// Duration is in milliseconds and set to zero when it is not available.
423 pub fn get_duration(&self) -> u64 {
424 let duration = self.dmx.get_duration();
425 if duration != 0 {
426 return duration;
427 }
428 let mut duration = 0;
429 for stream in self.streams.iter() {
430 if stream.duration > 0 {
431 let dur = NATimeInfo::ts_to_time(stream.duration, 1000, stream.tb_num, stream.tb_den);
432 if duration < dur {
433 duration = dur;
434 }
435 }
436 }
437 duration
438 }
439 }
440
441 impl<'a> NAOptionHandler for Demuxer<'a> {
442 fn get_supported_options(&self) -> &[NAOptionDefinition] {
443 self.dmx.get_supported_options()
444 }
445 fn set_options(&mut self, options: &[NAOption]) {
446 self.dmx.set_options(options);
447 }
448 fn query_option_value(&self, name: &str) -> Option<NAValue> {
449 self.dmx.query_option_value(name)
450 }
451 }
452
453 impl From<ByteIOError> for DemuxerError {
454 fn from(_: ByteIOError) -> Self { DemuxerError::IOError }
455 }
456
457 /// The trait for creating demuxers.
458 pub trait DemuxerCreator {
459 /// Creates new demuxer instance that will use `ByteReader` source as an input.
460 fn new_demuxer<'a>(&self, br: &'a mut ByteReader<'a>) -> Box<dyn DemuxCore<'a> + 'a>;
461 /// Returns the name of current demuxer creator (equal to the container name it can demux).
462 fn get_name(&self) -> &'static str;
463 }
464
465 /// Creates demuxer for a provided bytestream.
466 pub fn create_demuxer<'a>(dmxcr: &dyn DemuxerCreator, br: &'a mut ByteReader<'a>) -> DemuxerResult<Demuxer<'a>> {
467 let mut dmx = dmxcr.new_demuxer(br);
468 let mut str = StreamManager::new();
469 let mut seek_idx = SeekIndex::new();
470 dmx.open(&mut str, &mut seek_idx)?;
471 Ok(Demuxer::new(dmx, str, seek_idx))
472 }
473
474 /// List of registered demuxers.
475 #[derive(Default)]
476 pub struct RegisteredDemuxers {
477 dmxs: Vec<&'static dyn DemuxerCreator>,
478 }
479
480 impl RegisteredDemuxers {
481 /// Constructs a new `RegisteredDemuxers` instance.
482 pub fn new() -> Self {
483 Self { dmxs: Vec::new() }
484 }
485 /// Registers a new demuxer.
486 pub fn add_demuxer(&mut self, dmx: &'static dyn DemuxerCreator) {
487 self.dmxs.push(dmx);
488 }
489 /// Searches for a demuxer that supports requested container format.
490 pub fn find_demuxer(&self, name: &str) -> Option<&dyn DemuxerCreator> {
491 for &dmx in self.dmxs.iter() {
492 if dmx.get_name() == name {
493 return Some(dmx);
494 }
495 }
496 None
497 }
498 /// Provides an iterator over currently registered demuxers.
499 pub fn iter(&self) -> std::slice::Iter<&dyn DemuxerCreator> {
500 self.dmxs.iter()
501 }
502 }
503
504 /// A trait for raw data demuxing operations.
505 pub trait RawDemuxCore<'a>: NAOptionHandler {
506 /// Opens the input stream, reads required headers and prepares everything for packet demuxing.
507 fn open(&mut self, strmgr: &mut StreamManager, seek_idx: &mut SeekIndex) -> DemuxerResult<()>;
508 /// Reads a piece of raw data.
509 fn get_data(&mut self, strmgr: &mut StreamManager) -> DemuxerResult<NARawData>;
510 /// Seeks to the requested time.
511 fn seek(&mut self, time: NATimePoint, seek_idx: &SeekIndex) -> DemuxerResult<()>;
512 /// Returns container duration in milliseconds (zero if not available).
513 fn get_duration(&self) -> u64;
514 }
515
516 /// Demuxer structure with auxiliary data.
517 pub struct RawDemuxer<'a> {
518 dmx: Box<dyn RawDemuxCore<'a> + 'a>,
519 streams: StreamManager,
520 seek_idx: SeekIndex,
521 }
522
523 impl<'a> RawDemuxer<'a> {
524 /// Constructs a new `Demuxer` instance.
525 fn new(dmx: Box<dyn RawDemuxCore<'a> + 'a>, strmgr: StreamManager, seek_idx: SeekIndex) -> Self {
526 Self {
527 dmx,
528 streams: strmgr,
529 seek_idx,
530 }
531 }
532 /// Returns a stream reference by its number.
533 pub fn get_stream(&self, idx: usize) -> Option<NAStreamRef> {
534 self.streams.get_stream(idx)
535 }
536 /// Returns a stream reference by its ID.
537 pub fn get_stream_by_id(&self, id: u32) -> Option<NAStreamRef> {
538 self.streams.get_stream_by_id(id)
539 }
540 /// Reports the total number of streams.
541 pub fn get_num_streams(&self) -> usize {
542 self.streams.get_num_streams()
543 }
544 /// Returns a reference to the internal stream manager.
545 pub fn get_stream_manager(&self) -> &StreamManager {
546 &self.streams
547 }
548 /// Returns an iterator over streams.
549 pub fn get_streams(&self) -> StreamIter {
550 self.streams.iter()
551 }
552 /// Returns 'ignored' marker for requested stream.
553 pub fn is_ignored_stream(&self, idx: usize) -> bool {
554 self.streams.is_ignored(idx)
555 }
556 /// Sets 'ignored' marker for requested stream.
557 pub fn set_ignored_stream(&mut self, idx: usize) {
558 self.streams.set_ignored(idx)
559 }
560 /// Clears 'ignored' marker for requested stream.
561 pub fn set_unignored_stream(&mut self, idx: usize) {
562 self.streams.set_unignored(idx)
563 }
564
565 /// Demuxes a new piece of data from the container.
566 pub fn get_data(&mut self) -> DemuxerResult<NARawData> {
567 loop {
568 let res = self.dmx.get_data(&mut self.streams);
569 if self.streams.no_ign || res.is_err() { return res; }
570 let res = res.unwrap();
571 let idx = res.get_stream().get_num();
572 if !self.is_ignored_stream(idx) {
573 return Ok(res);
574 }
575 }
576 }
577 /// Seeks to the requested time if possible.
578 pub fn seek(&mut self, time: NATimePoint) -> DemuxerResult<()> {
579 if self.seek_idx.skip_index {
580 return Err(DemuxerError::NotPossible);
581 }
582 self.dmx.seek(time, &self.seek_idx)
583 }
584 /// Returns internal seek index.
585 pub fn get_seek_index(&self) -> &SeekIndex {
586 &self.seek_idx
587 }
588 /// Returns media duration reported by container or its streams.
589 ///
590 /// Duration is in milliseconds and set to zero when it is not available.
591 pub fn get_duration(&self) -> u64 {
592 let duration = self.dmx.get_duration();
593 if duration != 0 {
594 return duration;
595 }
596 let mut duration = 0;
597 for stream in self.streams.iter() {
598 if stream.duration > 0 {
599 let dur = NATimeInfo::ts_to_time(stream.duration, 1000, stream.tb_num, stream.tb_den);
600 if duration < dur {
601 duration = dur;
602 }
603 }
604 }
605 duration
606 }
607 }
608
609 impl<'a> NAOptionHandler for RawDemuxer<'a> {
610 fn get_supported_options(&self) -> &[NAOptionDefinition] {
611 self.dmx.get_supported_options()
612 }
613 fn set_options(&mut self, options: &[NAOption]) {
614 self.dmx.set_options(options);
615 }
616 fn query_option_value(&self, name: &str) -> Option<NAValue> {
617 self.dmx.query_option_value(name)
618 }
619 }
620
621 /// The trait for creating raw data demuxers.
622 pub trait RawDemuxerCreator {
623 /// Creates new raw demuxer instance that will use `ByteReader` source as an input.
624 fn new_demuxer<'a>(&self, br: &'a mut ByteReader<'a>) -> Box<dyn RawDemuxCore<'a> + 'a>;
625 /// Tries to check whether the input can be demuxed with the demuxer.
626 fn check_format(&self, br: &mut ByteReader) -> bool;
627 /// Returns the name of current raw data demuxer creator (equal to the container name it can demux).
628 fn get_name(&self) -> &'static str;
629 }
630
631 /// Creates raw data demuxer for a provided bytestream.
632 pub fn create_raw_demuxer<'a>(dmxcr: &dyn RawDemuxerCreator, br: &'a mut ByteReader<'a>) -> DemuxerResult<RawDemuxer<'a>> {
633 let mut dmx = dmxcr.new_demuxer(br);
634 let mut str = StreamManager::new();
635 let mut seek_idx = SeekIndex::new();
636 dmx.open(&mut str, &mut seek_idx)?;
637 Ok(RawDemuxer::new(dmx, str, seek_idx))
638 }
639
640 /// List of registered demuxers.
641 #[derive(Default)]
642 pub struct RegisteredRawDemuxers {
643 dmxs: Vec<&'static dyn RawDemuxerCreator>,
644 }
645
646 impl RegisteredRawDemuxers {
647 /// Constructs a new `RegisteredDemuxers` instance.
648 pub fn new() -> Self {
649 Self { dmxs: Vec::new() }
650 }
651 /// Registers a new demuxer.
652 pub fn add_demuxer(&mut self, dmx: &'static dyn RawDemuxerCreator) {
653 self.dmxs.push(dmx);
654 }
655 /// Searches for a demuxer that supports requested container format.
656 pub fn find_demuxer(&self, name: &str) -> Option<&dyn RawDemuxerCreator> {
657 for &dmx in self.dmxs.iter() {
658 if dmx.get_name() == name {
659 return Some(dmx);
660 }
661 }
662 None
663 }
664 /// Provides an iterator over currently registered demuxers.
665 pub fn iter(&self) -> std::slice::Iter<&dyn RawDemuxerCreator> {
666 self.dmxs.iter()
667 }
668 }