1#![cfg_attr(not(feature = "std"), no_std)]
29#![cfg_attr(docsrs, feature(doc_cfg))]
30#![doc = "\n# Examples\n"]
33#![doc = "Two runnable examples ship with this crate (`cargo run -p dvb-conformance --example <name>`).\n"]
34#![doc = "\n## `monitor_stream`\n\n```rust,ignore"]
35#![doc = include_str!("../examples/monitor_stream.rs")]
36#![doc = "```\n\n## `priority_breakdown`\n\n```rust,ignore"]
37#![doc = include_str!("../examples/priority_breakdown.rs")]
38#![doc = "```"]
39extern crate alloc;
40
41use alloc::collections::BTreeMap;
42use alloc::format;
43use alloc::string::String;
44use alloc::vec::Vec;
45use core::time::Duration;
46
47use broadcast_common::Parse;
48use dvb_si::tables::pat::{PatSection, TABLE_ID as PAT_TABLE_ID};
49use dvb_si::tables::pmt::PmtSection;
50use mpeg_ts::section::Section;
51use mpeg_ts::ts::{SectionReassembler, TsPacket};
52
53const PID_PAT: u16 = 0x0000;
57const PID_CAT: u16 = 0x0001;
59const PID_NIT: u16 = 0x0010;
61const PID_SDT_BAT: u16 = 0x0011;
63const PID_EIT: u16 = 0x0012;
65const PID_TDT_TOT: u16 = 0x0014;
67const PID_NULL: u16 = 0x1FFF;
69
70const SYNC_BYTE: u8 = 0x47;
72
73const SI_PIDS: [u16; 6] = [PID_PAT, PID_CAT, PID_NIT, PID_SDT_BAT, PID_EIT, PID_TDT_TOT];
75
76const DEFAULT_PAT_MAX_INTERVAL_MS: u64 = 500;
81
82const DEFAULT_PMT_MAX_INTERVAL_MS: u64 = 500;
84
85const DEFAULT_PID_ERROR_PERIOD_SECS: u64 = 5;
87
88const DEFAULT_SYNC_ACQUIRE_PACKETS: u8 = 5;
91
92const DEFAULT_SYNC_LOSS_PACKETS: u8 = 2;
95
96const DEFAULT_PCR_REPETITION_LIMIT_MS: u64 = 100;
99
100const DEFAULT_PCR_DISCONTINUITY_LIMIT_MS: u64 = 100;
103
104const DEFAULT_PTS_REPETITION_LIMIT_MS: u64 = 700;
107
108const DEFAULT_SI_NIT_INTERVAL_SECS: u64 = 10;
111
112const DEFAULT_SI_SDT_INTERVAL_SECS: u64 = 2;
115
116const DEFAULT_SI_EIT_PF_INTERVAL_SECS: u64 = 2;
119
120const DEFAULT_SI_TDT_INTERVAL_SECS: u64 = 30;
123
124const PCR_MODULUS_27MHZ: u64 = (1u64 << 33) * 300;
129
130const CLOCK_27MHZ: u64 = 27_000_000;
132
133const PES_PREFIX_0: u8 = 0x00;
135const PES_PREFIX_1: u8 = 0x00;
137const PES_PREFIX_2: u8 = 0x01;
139
140const PES_FLAGS_OFFSET: usize = 6;
143
144const PES_PTS_DTS_FLAGS_MASK: u8 = 0b1100_0000;
147
148const PES_PTS_PRESENT: u8 = 0b1000_0000;
150
151const CAT_TABLE_ID: u8 = dvb_si::table_id::TableId::Cat as u8;
153
154const NIT_ACTUAL_TABLE_ID: u8 = dvb_si::table_id::TableId::NetworkInformationActual as u8;
156
157const SDT_ACTUAL_TABLE_ID: u8 = dvb_si::table_id::TableId::ServiceDescriptionActual as u8;
159
160const EIT_PF_ACTUAL_TABLE_ID: u8 = dvb_si::table_id::TableId::EventInformationPfActual as u8;
162
163const TDT_TABLE_ID: u8 = dvb_si::table_id::TableId::TimeAndDate as u8;
165
166#[derive(Debug, Clone, Copy, PartialEq, Eq)]
170#[cfg_attr(feature = "serde", derive(serde::Serialize))]
171#[non_exhaustive]
172pub enum Priority {
173 First,
175 Second,
177 Third,
179}
180
181impl Priority {
182 #[must_use]
184 pub fn name(&self) -> &'static str {
185 match self {
186 Self::First => "first priority",
187 Self::Second => "second priority",
188 Self::Third => "third priority",
189 }
190 }
191}
192broadcast_common::impl_spec_display!(Priority);
193
194#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
198#[cfg_attr(feature = "serde", derive(serde::Serialize))]
199#[non_exhaustive]
200pub enum Indicator {
201 TsSyncLoss,
205 SyncByteError,
207 PatError2,
209 ContinuityCountError,
211 PmtError2,
213 PidError,
215
216 TransportError,
219 CrcError,
221 PcrRepetitionError,
223 PcrDiscontinuityError,
226 PtsError,
228 CatError,
230
231 SiRepetitionError,
235}
236
237impl Indicator {
238 #[must_use]
240 pub fn priority(self) -> Priority {
241 match self {
242 Self::TsSyncLoss
243 | Self::SyncByteError
244 | Self::PatError2
245 | Self::ContinuityCountError
246 | Self::PmtError2
247 | Self::PidError => Priority::First,
248 Self::TransportError
249 | Self::CrcError
250 | Self::PcrRepetitionError
251 | Self::PcrDiscontinuityError
252 | Self::PtsError
253 | Self::CatError => Priority::Second,
254 Self::SiRepetitionError => Priority::Third,
255 }
256 }
257
258 #[must_use]
260 pub fn name(self) -> &'static str {
261 match self {
262 Self::TsSyncLoss => "TS_sync_loss",
263 Self::SyncByteError => "Sync_byte_error",
264 Self::PatError2 => "PAT_error_2",
265 Self::ContinuityCountError => "Continuity_count_error",
266 Self::PmtError2 => "PMT_error_2",
267 Self::PidError => "PID_error",
268 Self::TransportError => "Transport_error",
269 Self::CrcError => "CRC_error",
270 Self::PcrRepetitionError => "PCR_repetition_error",
271 Self::PcrDiscontinuityError => "PCR_discontinuity_indicator_error",
272 Self::PtsError => "PTS_error",
273 Self::CatError => "CAT_error",
274 Self::SiRepetitionError => "SI_repetition_error",
275 }
276 }
277
278 #[must_use]
280 pub fn clause(self) -> &'static str {
281 match self {
282 Self::TsSyncLoss => "TR 101 290 v1.4.1 Table 5.0a indicator 1.1",
283 Self::SyncByteError => "TR 101 290 v1.4.1 Table 5.0a indicator 1.2",
284 Self::PatError2 => "TR 101 290 v1.4.1 Table 5.0a indicator 1.3.a",
285 Self::ContinuityCountError => "TR 101 290 v1.4.1 Table 5.0a indicator 1.4",
286 Self::PmtError2 => "TR 101 290 v1.4.1 Table 5.0a indicator 1.5.a",
287 Self::PidError => "TR 101 290 v1.4.1 Table 5.0a indicator 1.6",
288 Self::TransportError => "TR 101 290 v1.4.1 Table 5.0b indicator 2.1",
289 Self::CrcError => "TR 101 290 v1.4.1 Table 5.0b indicator 2.2",
290 Self::PcrRepetitionError => "TR 101 290 v1.4.1 Table 5.0b indicator 2.3a",
291 Self::PcrDiscontinuityError => "TR 101 290 v1.4.1 Table 5.0b indicator 2.3b",
292 Self::PtsError => "TR 101 290 v1.4.1 Table 5.0b indicator 2.5",
293 Self::CatError => "TR 101 290 v1.4.1 Table 5.0b indicator 2.6",
294 Self::SiRepetitionError => "TR 101 290 v1.4.1 Table 5.0c indicator 3.2",
295 }
296 }
297}
298broadcast_common::impl_spec_display!(Indicator);
299
300#[derive(Debug, Clone, PartialEq, Eq)]
302#[cfg_attr(feature = "serde", derive(serde::Serialize))]
303#[non_exhaustive]
304pub struct ConformanceEvent {
305 pub indicator: Indicator,
307 pub priority: Priority,
309 pub pid: Option<u16>,
311 pub at: Duration,
313 pub detail: String,
315}
316
317#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
319#[cfg_attr(feature = "serde", derive(serde::Serialize))]
320#[non_exhaustive]
321pub struct Stats {
322 pub packets: u64,
324 pub events: u64,
326 pub in_sync: bool,
328}
329
330#[derive(Debug, Clone)]
332#[non_exhaustive]
333pub struct Config {
334 pub pat_max_interval: Duration,
337 pub pmt_max_interval: Duration,
340 pub pid_error_period: Duration,
343 pub sync_acquire_packets: u8,
346 pub sync_loss_packets: u8,
349 pub pcr_repetition_limit: Duration,
352 pub pcr_discontinuity_limit: Duration,
355 pub pts_repetition_limit: Duration,
358 pub si_nit_interval: Duration,
361 pub si_sdt_interval: Duration,
364 pub si_eit_pf_interval: Duration,
367 pub si_tdt_interval: Duration,
370}
371
372impl Default for Config {
373 fn default() -> Self {
374 Self {
375 pat_max_interval: Duration::from_millis(DEFAULT_PAT_MAX_INTERVAL_MS),
376 pmt_max_interval: Duration::from_millis(DEFAULT_PMT_MAX_INTERVAL_MS),
377 pid_error_period: Duration::from_secs(DEFAULT_PID_ERROR_PERIOD_SECS),
378 sync_acquire_packets: DEFAULT_SYNC_ACQUIRE_PACKETS,
379 sync_loss_packets: DEFAULT_SYNC_LOSS_PACKETS,
380 pcr_repetition_limit: Duration::from_millis(DEFAULT_PCR_REPETITION_LIMIT_MS),
381 pcr_discontinuity_limit: Duration::from_millis(DEFAULT_PCR_DISCONTINUITY_LIMIT_MS),
382 pts_repetition_limit: Duration::from_millis(DEFAULT_PTS_REPETITION_LIMIT_MS),
383 si_nit_interval: Duration::from_secs(DEFAULT_SI_NIT_INTERVAL_SECS),
384 si_sdt_interval: Duration::from_secs(DEFAULT_SI_SDT_INTERVAL_SECS),
385 si_eit_pf_interval: Duration::from_secs(DEFAULT_SI_EIT_PF_INTERVAL_SECS),
386 si_tdt_interval: Duration::from_secs(DEFAULT_SI_TDT_INTERVAL_SECS),
387 }
388 }
389}
390
391struct CcState {
395 last_cc: u8,
396 had_payload: bool,
397 dup_used: bool,
398 initialised: bool,
399}
400
401struct PresenceTimer {
403 last_seen: Duration,
404 reported: bool,
405}
406
407struct PmtTracking {
409 timer: PresenceTimer,
410 reassembler: SectionReassembler,
411}
412
413struct EsTracking {
415 timer: PresenceTimer,
416}
417
418struct PcrState {
420 last_pcr_27mhz: u64,
421 last_pcr_time: Duration,
422 initialised: bool,
423}
424
425struct PtsState {
427 last_pts_time: Duration,
428 armed: bool,
429}
430
431struct SiReassembly {
433 reassembler: SectionReassembler,
434}
435
436struct SiRepetitionTimer {
440 last_seen: Duration,
441 reported: bool,
442 armed: bool,
443}
444
445pub struct ConformanceMonitor {
453 config: Config,
454 events: Vec<ConformanceEvent>,
455 stats: Stats,
456
457 in_sync: bool,
459 good_run: u8,
460 bad_run: u8,
461
462 cc_states: BTreeMap<u16, CcState>,
464
465 pat_reassembler: SectionReassembler,
467 pat_timer: PresenceTimer,
468
469 pmt_trackings: BTreeMap<u16, PmtTracking>,
471
472 es_trackings: BTreeMap<u16, EsTracking>,
474
475 si_reassemblies: BTreeMap<u16, SiReassembly>,
477
478 pcr_states: BTreeMap<u16, PcrState>,
480
481 pts_states: BTreeMap<u16, PtsState>,
483
484 cat_seen: bool,
486 scrambled_without_cat_reported: bool,
487
488 si_timers: BTreeMap<u8, SiRepetitionTimer>,
490}
491
492impl ConformanceMonitor {
493 pub fn new() -> Self {
495 Self::with_config(Config::default())
496 }
497
498 pub fn with_config(config: Config) -> Self {
500 let mut si_reassemblies = BTreeMap::new();
501 for &pid in &SI_PIDS {
502 si_reassemblies.insert(
503 pid,
504 SiReassembly {
505 reassembler: SectionReassembler::default(),
506 },
507 );
508 }
509 Self {
510 config,
511 events: Vec::new(),
512 stats: Stats {
513 packets: 0,
514 events: 0,
515 in_sync: false,
516 },
517 in_sync: false,
518 good_run: 0,
519 bad_run: 0,
520 cc_states: BTreeMap::new(),
521 pat_reassembler: SectionReassembler::default(),
522 pat_timer: PresenceTimer {
523 last_seen: Duration::ZERO,
524 reported: false,
525 },
526 pmt_trackings: BTreeMap::new(),
527 es_trackings: BTreeMap::new(),
528 si_reassemblies,
529 pcr_states: BTreeMap::new(),
530 pts_states: BTreeMap::new(),
531 cat_seen: false,
532 scrambled_without_cat_reported: false,
533 si_timers: BTreeMap::new(),
534 }
535 }
536
537 pub fn feed(&mut self, ts_packet: &[u8], t: Duration) -> &[ConformanceEvent] {
543 self.events.clear();
544 self.stats.packets += 1;
545
546 let sync_ok = !ts_packet.is_empty() && ts_packet[0] == SYNC_BYTE;
548 if !sync_ok {
549 self.emit(Indicator::SyncByteError, None, t, "sync_byte != 0x47");
550 }
551
552 if sync_ok {
554 self.good_run = self.good_run.saturating_add(1);
555 self.bad_run = 0;
556 if !self.in_sync && self.good_run >= self.config.sync_acquire_packets {
557 self.in_sync = true;
558 }
559 } else {
560 self.bad_run = self.bad_run.saturating_add(1);
561 self.good_run = 0;
562 if self.in_sync && self.bad_run >= self.config.sync_loss_packets {
563 self.in_sync = false;
564 self.emit(
565 Indicator::TsSyncLoss,
566 None,
567 t,
568 "sync lost after hysteresis threshold",
569 );
570 }
571 }
572
573 if !self.in_sync {
577 return &self.events;
578 }
579
580 let packet = match TsPacket::parse(ts_packet) {
582 Ok(p) => p,
583 Err(_) => return &self.events,
584 };
585 let header = &packet.header;
586 let pid = header.pid;
587
588 if header.tei {
590 self.emit(
591 Indicator::TransportError,
592 Some(pid),
593 t,
594 format!("transport_error_indicator set on PID 0x{:04X}", pid),
595 );
596 }
597
598 if pid != PID_NULL {
600 self.check_cc(
601 pid,
602 header.continuity_counter,
603 header.has_payload,
604 t,
605 ts_packet,
606 );
607 }
608
609 if pid == PID_PAT && header.scrambling != 0 {
611 self.emit(
612 Indicator::PatError2,
613 Some(PID_PAT),
614 t,
615 format!(
616 "scrambling_control_field != 00 on PID 0x0000 (got {})",
617 header.scrambling
618 ),
619 );
620 }
621
622 if self.pmt_trackings.contains_key(&pid) && header.scrambling != 0 {
624 self.emit(
625 Indicator::PmtError2,
626 Some(pid),
627 t,
628 format!(
629 "scrambling_control_field != 00 on program_map_PID 0x{:04X}",
630 pid
631 ),
632 );
633 }
634
635 if header.scrambling != 0 && !self.cat_seen && !self.scrambled_without_cat_reported {
642 self.scrambled_without_cat_reported = true;
643 self.emit(
644 Indicator::CatError,
645 Some(pid),
646 t,
647 format!(
648 "scrambled packet on PID 0x{:04X} but no CAT seen on PID 0x0001",
649 pid
650 ),
651 );
652 }
653
654 if pid == PID_PAT && header.has_payload {
656 if let Some(payload) = packet.payload {
657 self.pat_reassembler.feed(payload, header.pusi);
658 }
659 self.pat_timer.last_seen = t;
660 self.pat_timer.reported = false;
661 while let Some(section_bytes) = self.pat_reassembler.pop_section() {
662 self.check_crc_and_process_pat(§ion_bytes, pid, t);
663 }
664 }
665
666 if self.pmt_trackings.contains_key(&pid) && header.has_payload {
668 if let Some(payload) = packet.payload {
669 if let Some(tracking) = self.pmt_trackings.get_mut(&pid) {
670 tracking.reassembler.feed(payload, header.pusi);
671 }
672 }
673 let sections: Vec<_> = if let Some(tracking) = self.pmt_trackings.get_mut(&pid) {
674 tracking.timer.last_seen = t;
675 tracking.timer.reported = false;
676 core::iter::from_fn(|| tracking.reassembler.pop_section()).collect()
677 } else {
678 Vec::new()
679 };
680 for section_bytes in §ions {
681 self.check_crc_and_process_pmt(section_bytes, pid, t);
682 }
683 }
684
685 if pid != PID_PAT
689 && !self.pmt_trackings.contains_key(&pid)
690 && self.si_reassemblies.contains_key(&pid)
691 && header.has_payload
692 {
693 if let Some(payload) = packet.payload {
694 if let Some(si_ra) = self.si_reassemblies.get_mut(&pid) {
695 si_ra.reassembler.feed(payload, header.pusi);
696 }
697 }
698 let sections: Vec<_> = if let Some(si_ra) = self.si_reassemblies.get_mut(&pid) {
699 core::iter::from_fn(|| si_ra.reassembler.pop_section()).collect()
700 } else {
701 Vec::new()
702 };
703 for section_bytes in §ions {
704 self.check_crc_for_si(section_bytes, pid, t);
705 self.check_cat_table_id(section_bytes, pid, t);
706 self.update_si_repetition(section_bytes, pid, t);
707 }
708 }
709 if let Some(tracking) = self.es_trackings.get_mut(&pid) {
717 tracking.timer.last_seen = t;
718 tracking.timer.reported = false;
719 }
720
721 if let Some(Ok(af)) = packet.adaptation_field() {
723 if let Some(pcr) = af.pcr {
724 self.check_pcr(pid, pcr.as_27mhz(), af.discontinuity_indicator, t);
725 }
726 }
727
728 if header.pusi
730 && header.scrambling == 0
731 && self.es_trackings.contains_key(&pid)
732 && header.has_payload
733 {
734 if let Some(payload) = packet.payload {
735 self.check_pts(pid, payload, t);
736 }
737 }
738
739 self.check_presence_timeouts(t);
741
742 &self.events
743 }
744
745 pub fn stats(&self) -> Stats {
747 Stats {
748 in_sync: self.in_sync,
749 ..self.stats
750 }
751 }
752
753 fn emit(
756 &mut self,
757 indicator: Indicator,
758 pid: Option<u16>,
759 at: Duration,
760 detail: impl Into<String>,
761 ) {
762 let event = ConformanceEvent {
763 indicator,
764 priority: indicator.priority(),
765 pid,
766 at,
767 detail: detail.into(),
768 };
769 self.stats.events += 1;
770 self.events.push(event);
771 }
772
773 fn check_cc(&mut self, pid: u16, cc: u8, has_payload: bool, t: Duration, raw: &[u8]) {
775 let discontinuity = if raw.len() >= 5 {
778 let b3 = raw[3];
779 let has_adaptation = (b3 & 0x20) != 0;
780 if has_adaptation {
781 let af_len = raw[4] as usize;
782 if af_len > 0 && raw.len() > 5 {
783 (raw[5] & 0x80) != 0
784 } else {
785 false
786 }
787 } else {
788 false
789 }
790 } else {
791 false
792 };
793
794 let (expected, is_duplicate, should_emit_dup, should_emit_cc) = {
796 let state = self.cc_states.entry(pid).or_insert_with(|| CcState {
797 last_cc: cc,
798 had_payload: has_payload,
799 dup_used: false,
800 initialised: false,
801 });
802
803 if !state.initialised {
804 state.last_cc = cc;
805 state.had_payload = has_payload;
806 state.dup_used = false;
807 state.initialised = true;
808 return;
809 }
810
811 if discontinuity {
812 (0u8, false, false, false)
814 } else {
815 let is_duplicate = cc == state.last_cc && has_payload;
816 let mut should_emit_dup = false;
817 let mut should_emit_cc = false;
818
819 if is_duplicate {
820 if state.dup_used {
821 should_emit_dup = true;
822 }
823 } else {
824 state.dup_used = false;
825 let expected = if has_payload {
826 (state.last_cc.wrapping_add(1)) & 0x0F
827 } else {
828 state.last_cc
829 };
830 if cc != expected {
831 should_emit_cc = true;
832 }
833 }
834
835 (
836 if has_payload {
837 (state.last_cc.wrapping_add(1)) & 0x0F
838 } else {
839 state.last_cc
840 },
841 is_duplicate,
842 should_emit_dup,
843 should_emit_cc,
844 )
845 }
846 };
847
848 if should_emit_dup {
850 self.emit(
851 Indicator::ContinuityCountError,
852 Some(pid),
853 t,
854 format!(
855 "second consecutive duplicate on PID 0x{:04X} (cc={})",
856 pid, cc
857 ),
858 );
859 }
860 if should_emit_cc {
861 self.emit(
862 Indicator::ContinuityCountError,
863 Some(pid),
864 t,
865 format!("expected cc={}, got {} on PID 0x{:04X}", expected, cc, pid),
866 );
867 }
868
869 let state = self.cc_states.get_mut(&pid).unwrap();
871 if discontinuity {
872 state.last_cc = cc;
873 state.had_payload = has_payload;
874 state.dup_used = false;
875 } else if is_duplicate {
876 state.dup_used = true;
878 } else {
879 state.dup_used = false;
880 state.last_cc = cc;
881 state.had_payload = has_payload;
882 }
883 }
884
885 fn check_crc_and_process_pat(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
887 self.check_crc_for_section(section_bytes, pid, t);
889
890 self.process_pat_section(section_bytes, t);
891 }
892
893 fn check_crc_and_process_pmt(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
895 self.check_crc_for_section(section_bytes, pid, t);
897
898 self.process_pmt_section(section_bytes, pid, t);
899 }
900
901 fn check_crc_for_section(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
906 let section = match Section::parse(section_bytes) {
907 Ok(s) => s,
908 Err(_) => return,
909 };
910
911 if let Err(mpeg_ts::error::Error::CrcMismatch { .. }) = section.validate_crc(section_bytes)
913 {
914 self.emit(
915 Indicator::CrcError,
916 Some(pid),
917 t,
918 format!(
919 "CRC-32 mismatch on PID 0x{:04X} (table_id 0x{:02X})",
920 pid, section.table_id
921 ),
922 );
923 }
924 }
925
926 fn check_crc_for_si(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
928 self.check_crc_for_section(section_bytes, pid, t);
929 }
930
931 fn check_cat_table_id(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
934 if pid != PID_CAT {
935 return;
936 }
937 let section = match Section::parse(section_bytes) {
938 Ok(s) => s,
939 Err(_) => return,
940 };
941 if section.table_id == CAT_TABLE_ID {
942 self.cat_seen = true;
944 self.scrambled_without_cat_reported = false;
947 } else {
948 self.emit(
949 Indicator::CatError,
950 Some(PID_CAT),
951 t,
952 format!(
953 "section with table_id 0x{:02X} on PID 0x0001 (expected 0x01 for CAT)",
954 section.table_id
955 ),
956 );
957 }
958 }
959
960 fn process_pat_section(&mut self, section_bytes: &[u8], t: Duration) {
962 let section = match Section::parse(section_bytes) {
963 Ok(s) => s,
964 Err(_) => return,
965 };
966
967 if section.table_id != PAT_TABLE_ID {
969 self.emit(
970 Indicator::PatError2,
971 Some(PID_PAT),
972 t,
973 format!(
974 "section with table_id 0x{:02X} on PID 0x0000 (expected 0x00)",
975 section.table_id
976 ),
977 );
978 return;
979 }
980
981 let pat = match PatSection::parse(section_bytes) {
983 Ok(p) => p,
984 Err(_) => return,
985 };
986
987 for entry in pat.programmes() {
989 let pmt_pid = entry.pid;
990 self.pmt_trackings
991 .entry(pmt_pid)
992 .or_insert_with(|| PmtTracking {
993 timer: PresenceTimer {
994 last_seen: t,
995 reported: false,
996 },
997 reassembler: SectionReassembler::default(),
998 });
999 }
1000 }
1001
1002 fn process_pmt_section(&mut self, section_bytes: &[u8], _pid: u16, t: Duration) {
1004 let section = match Section::parse(section_bytes) {
1005 Ok(s) => s,
1006 Err(_) => return,
1007 };
1008
1009 let pmt_table_id: u8 = dvb_si::tables::pmt::TABLE_ID;
1014 if section.table_id != pmt_table_id {
1015 return;
1016 }
1017
1018 let pmt = match PmtSection::parse(section_bytes) {
1020 Ok(p) => p,
1021 Err(_) => return,
1022 };
1023
1024 let mut new_es_pids: Vec<u16> = Vec::new();
1026 if pmt.pcr_pid != PID_NULL && !self.es_trackings.contains_key(&pmt.pcr_pid) {
1027 new_es_pids.push(pmt.pcr_pid);
1028 }
1029 for stream in &pmt.streams {
1030 let es_pid = stream.elementary_pid;
1031 if !self.es_trackings.contains_key(&es_pid) {
1032 new_es_pids.push(es_pid);
1033 }
1034 }
1035
1036 for es_pid in new_es_pids {
1037 self.es_trackings.insert(
1038 es_pid,
1039 EsTracking {
1040 timer: PresenceTimer {
1041 last_seen: t,
1042 reported: false,
1043 },
1044 },
1045 );
1046 }
1047 }
1048
1049 fn check_pcr(&mut self, pid: u16, pcr_27mhz: u64, discontinuity: bool, t: Duration) {
1051 let state = self.pcr_states.entry(pid).or_insert_with(|| PcrState {
1052 last_pcr_27mhz: 0,
1053 last_pcr_time: Duration::ZERO,
1054 initialised: false,
1055 });
1056
1057 if !state.initialised {
1058 state.last_pcr_27mhz = pcr_27mhz;
1059 state.last_pcr_time = t;
1060 state.initialised = true;
1061 return;
1062 }
1063
1064 let last_pcr_time = state.last_pcr_time;
1066 let last_pcr_27mhz = state.last_pcr_27mhz;
1067
1068 let rep_interval = t.saturating_sub(last_pcr_time);
1071 let should_emit_rep = rep_interval > self.config.pcr_repetition_limit;
1072
1073 let delta =
1076 (pcr_27mhz.wrapping_add(PCR_MODULUS_27MHZ) - last_pcr_27mhz) % PCR_MODULUS_27MHZ;
1077 let delta_ms = delta * 1000 / CLOCK_27MHZ;
1078 let limit_ms = self.config.pcr_discontinuity_limit.as_millis() as u64;
1079 let should_emit_disc = delta_ms > limit_ms && !discontinuity;
1080
1081 if should_emit_rep {
1083 self.emit(
1084 Indicator::PcrRepetitionError,
1085 Some(pid),
1086 t,
1087 format!(
1088 "PCR interval {} ms exceeds limit {} ms on PID 0x{:04X}",
1089 rep_interval.as_millis(),
1090 self.config.pcr_repetition_limit.as_millis(),
1091 pid
1092 ),
1093 );
1094 }
1095 if should_emit_disc {
1096 self.emit(
1097 Indicator::PcrDiscontinuityError,
1098 Some(pid),
1099 t,
1100 format!(
1101 "PCR delta {} ms exceeds limit {} ms on PID 0x{:04X} without discontinuity_indicator",
1102 delta_ms, limit_ms, pid
1103 ),
1104 );
1105 }
1106
1107 let state = self.pcr_states.get_mut(&pid).unwrap();
1109 state.last_pcr_27mhz = pcr_27mhz;
1110 state.last_pcr_time = t;
1111 }
1112
1113 fn check_pts(&mut self, pid: u16, payload: &[u8], t: Duration) {
1118 if payload.len() < PES_FLAGS_OFFSET + 2 {
1120 return;
1121 }
1122 if payload[0] != PES_PREFIX_0 || payload[1] != PES_PREFIX_1 || payload[2] != PES_PREFIX_2 {
1123 return;
1124 }
1125
1126 let flags_byte = payload[PES_FLAGS_OFFSET];
1128 if (flags_byte >> 6) != 0b10 {
1129 return;
1130 }
1131
1132 let pts_dts_flags = payload[PES_FLAGS_OFFSET + 1] & PES_PTS_DTS_FLAGS_MASK;
1134 let pts_present = (pts_dts_flags & PES_PTS_PRESENT) != 0;
1135 if !pts_present {
1136 return;
1137 }
1138
1139 let state = self.pts_states.entry(pid).or_insert_with(|| PtsState {
1140 last_pts_time: Duration::ZERO,
1141 armed: false,
1142 });
1143
1144 if !state.armed {
1145 state.last_pts_time = t;
1147 state.armed = true;
1148 return;
1149 }
1150
1151 let last_pts_time = state.last_pts_time;
1153 let pts_interval = t.saturating_sub(last_pts_time);
1154 let should_emit = pts_interval > self.config.pts_repetition_limit;
1155
1156 if should_emit {
1157 self.emit(
1158 Indicator::PtsError,
1159 Some(pid),
1160 t,
1161 format!(
1162 "PTS interval {} ms exceeds limit {} ms on PID 0x{:04X}",
1163 pts_interval.as_millis(),
1164 self.config.pts_repetition_limit.as_millis(),
1165 pid
1166 ),
1167 );
1168 }
1169
1170 let state = self.pts_states.get_mut(&pid).unwrap();
1172 state.last_pts_time = t;
1173 }
1174
1175 fn update_si_repetition(&mut self, section_bytes: &[u8], _pid: u16, t: Duration) {
1179 let table_id = match Section::parse(section_bytes) {
1180 Ok(s) => s.table_id,
1181 Err(_) => return,
1182 };
1183
1184 let is_tracked = table_id == NIT_ACTUAL_TABLE_ID
1185 || table_id == SDT_ACTUAL_TABLE_ID
1186 || table_id == EIT_PF_ACTUAL_TABLE_ID
1187 || table_id == TDT_TABLE_ID;
1188
1189 if !is_tracked {
1190 return;
1191 }
1192
1193 let timer = self
1194 .si_timers
1195 .entry(table_id)
1196 .or_insert_with(|| SiRepetitionTimer {
1197 last_seen: Duration::ZERO,
1198 reported: false,
1199 armed: false,
1200 });
1201
1202 timer.last_seen = t;
1203 timer.reported = false;
1204 timer.armed = true;
1205 }
1206
1207 fn check_presence_timeouts(&mut self, t: Duration) {
1209 if t.saturating_sub(self.pat_timer.last_seen) > self.config.pat_max_interval
1211 && !self.pat_timer.reported
1212 {
1213 self.pat_timer.reported = true;
1214 self.emit(
1215 Indicator::PatError2,
1216 Some(PID_PAT),
1217 t,
1218 format!(
1219 "no PAT section within {} ms",
1220 self.config.pat_max_interval.as_millis()
1221 ),
1222 );
1223 }
1224
1225 let pmt_timeouts: Vec<(u16, u64)> = self
1228 .pmt_trackings
1229 .iter()
1230 .filter_map(|(&pid, tracking)| {
1231 if t.saturating_sub(tracking.timer.last_seen) > self.config.pmt_max_interval
1232 && !tracking.timer.reported
1233 {
1234 Some((pid, self.config.pmt_max_interval.as_millis() as u64))
1235 } else {
1236 None
1237 }
1238 })
1239 .collect();
1240 for (pid, interval_ms) in pmt_timeouts {
1241 if let Some(tracking) = self.pmt_trackings.get_mut(&pid) {
1242 tracking.timer.reported = true;
1243 }
1244 self.emit(
1245 Indicator::PmtError2,
1246 Some(pid),
1247 t,
1248 format!(
1249 "no PMT section on program_map_PID 0x{:04X} within {} ms",
1250 pid, interval_ms
1251 ),
1252 );
1253 }
1254
1255 let pid_timeouts: Vec<(u16, u64)> = self
1257 .es_trackings
1258 .iter()
1259 .filter_map(|(&pid, tracking)| {
1260 if t.saturating_sub(tracking.timer.last_seen) > self.config.pid_error_period
1261 && !tracking.timer.reported
1262 {
1263 Some((pid, self.config.pid_error_period.as_secs()))
1264 } else {
1265 None
1266 }
1267 })
1268 .collect();
1269 for (pid, period_secs) in pid_timeouts {
1270 if let Some(tracking) = self.es_trackings.get_mut(&pid) {
1271 tracking.timer.reported = true;
1272 }
1273 self.emit(
1274 Indicator::PidError,
1275 Some(pid),
1276 t,
1277 format!(
1278 "referenced PID 0x{:04X} absent for > {} s",
1279 pid, period_secs
1280 ),
1281 );
1282 }
1283
1284 let si_timeouts: Vec<(u8, u64, u16, u64)> = self
1287 .si_timers
1288 .iter()
1289 .filter_map(|(&table_id, timer)| {
1290 if !timer.armed || timer.reported {
1291 return None;
1292 }
1293 let (limit, pid) = match table_id {
1294 NIT_ACTUAL_TABLE_ID => (self.config.si_nit_interval, PID_NIT),
1295 SDT_ACTUAL_TABLE_ID => (self.config.si_sdt_interval, PID_SDT_BAT),
1296 EIT_PF_ACTUAL_TABLE_ID => (self.config.si_eit_pf_interval, PID_EIT),
1297 TDT_TABLE_ID => (self.config.si_tdt_interval, PID_TDT_TOT),
1298 _ => return None,
1299 };
1300 let interval = t.saturating_sub(timer.last_seen);
1301 if interval > limit {
1302 Some((
1303 table_id,
1304 interval.as_millis() as u64,
1305 pid,
1306 limit.as_millis() as u64,
1307 ))
1308 } else {
1309 None
1310 }
1311 })
1312 .collect();
1313 for (table_id, interval_ms, pid, limit_ms) in si_timeouts {
1314 if let Some(timer) = self.si_timers.get_mut(&table_id) {
1315 timer.reported = true;
1316 }
1317 let table_name = match table_id {
1318 NIT_ACTUAL_TABLE_ID => "NIT_actual",
1319 SDT_ACTUAL_TABLE_ID => "SDT_actual",
1320 EIT_PF_ACTUAL_TABLE_ID => "EIT_P/F_actual",
1321 TDT_TABLE_ID => "TDT",
1322 _ => "unknown",
1323 };
1324 self.emit(
1325 Indicator::SiRepetitionError,
1326 Some(pid),
1327 t,
1328 format!(
1329 "{} repetition interval {} ms exceeds {} ms",
1330 table_name, interval_ms, limit_ms
1331 ),
1332 );
1333 }
1334 }
1335}
1336
1337impl Default for ConformanceMonitor {
1338 fn default() -> Self {
1339 Self::new()
1340 }
1341}
1342
1343#[cfg(test)]
1344mod tests;