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{pid:04X}"),
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!("scrambling_control_field != 00 on program_map_PID 0x{pid:04X}"),
629 );
630 }
631
632 if header.scrambling != 0 && !self.cat_seen && !self.scrambled_without_cat_reported {
639 self.scrambled_without_cat_reported = true;
640 self.emit(
641 Indicator::CatError,
642 Some(pid),
643 t,
644 format!("scrambled packet on PID 0x{pid:04X} but no CAT seen on PID 0x0001"),
645 );
646 }
647
648 if pid == PID_PAT && header.has_payload {
650 if let Some(payload) = packet.payload {
651 self.pat_reassembler.feed(payload, header.pusi);
652 }
653 self.pat_timer.last_seen = t;
654 self.pat_timer.reported = false;
655 while let Some(section_bytes) = self.pat_reassembler.pop_section() {
656 self.check_crc_and_process_pat(§ion_bytes, pid, t);
657 }
658 }
659
660 if self.pmt_trackings.contains_key(&pid) && header.has_payload {
662 if let Some(payload) = packet.payload {
663 if let Some(tracking) = self.pmt_trackings.get_mut(&pid) {
664 tracking.reassembler.feed(payload, header.pusi);
665 }
666 }
667 let sections: Vec<_> = if let Some(tracking) = self.pmt_trackings.get_mut(&pid) {
668 tracking.timer.last_seen = t;
669 tracking.timer.reported = false;
670 core::iter::from_fn(|| tracking.reassembler.pop_section()).collect()
671 } else {
672 Vec::new()
673 };
674 for section_bytes in §ions {
675 self.check_crc_and_process_pmt(section_bytes, pid, t);
676 }
677 }
678
679 if pid != PID_PAT
683 && !self.pmt_trackings.contains_key(&pid)
684 && self.si_reassemblies.contains_key(&pid)
685 && header.has_payload
686 {
687 if let Some(payload) = packet.payload {
688 if let Some(si_ra) = self.si_reassemblies.get_mut(&pid) {
689 si_ra.reassembler.feed(payload, header.pusi);
690 }
691 }
692 let sections: Vec<_> = if let Some(si_ra) = self.si_reassemblies.get_mut(&pid) {
693 core::iter::from_fn(|| si_ra.reassembler.pop_section()).collect()
694 } else {
695 Vec::new()
696 };
697 for section_bytes in §ions {
698 self.check_crc_for_si(section_bytes, pid, t);
699 self.check_cat_table_id(section_bytes, pid, t);
700 self.update_si_repetition(section_bytes, pid, t);
701 }
702 }
703 if let Some(tracking) = self.es_trackings.get_mut(&pid) {
711 tracking.timer.last_seen = t;
712 tracking.timer.reported = false;
713 }
714
715 if let Some(Ok(af)) = packet.adaptation_field() {
717 if let Some(pcr) = af.pcr {
718 self.check_pcr(pid, pcr.as_27mhz(), af.discontinuity_indicator, t);
719 }
720 }
721
722 if header.pusi
724 && header.scrambling == 0
725 && self.es_trackings.contains_key(&pid)
726 && header.has_payload
727 {
728 if let Some(payload) = packet.payload {
729 self.check_pts(pid, payload, t);
730 }
731 }
732
733 self.check_presence_timeouts(t);
735
736 &self.events
737 }
738
739 pub fn stats(&self) -> Stats {
741 Stats {
742 in_sync: self.in_sync,
743 ..self.stats
744 }
745 }
746
747 fn emit(
750 &mut self,
751 indicator: Indicator,
752 pid: Option<u16>,
753 at: Duration,
754 detail: impl Into<String>,
755 ) {
756 let event = ConformanceEvent {
757 indicator,
758 priority: indicator.priority(),
759 pid,
760 at,
761 detail: detail.into(),
762 };
763 self.stats.events += 1;
764 self.events.push(event);
765 }
766
767 fn check_cc(&mut self, pid: u16, cc: u8, has_payload: bool, t: Duration, raw: &[u8]) {
769 let discontinuity = if raw.len() >= 5 {
772 let b3 = raw[3];
773 let has_adaptation = (b3 & 0x20) != 0;
774 if has_adaptation {
775 let af_len = raw[4] as usize;
776 if af_len > 0 && raw.len() > 5 {
777 (raw[5] & 0x80) != 0
778 } else {
779 false
780 }
781 } else {
782 false
783 }
784 } else {
785 false
786 };
787
788 let (expected, is_duplicate, should_emit_dup, should_emit_cc) = {
790 let state = self.cc_states.entry(pid).or_insert_with(|| CcState {
791 last_cc: cc,
792 had_payload: has_payload,
793 dup_used: false,
794 initialised: false,
795 });
796
797 if !state.initialised {
798 state.last_cc = cc;
799 state.had_payload = has_payload;
800 state.dup_used = false;
801 state.initialised = true;
802 return;
803 }
804
805 if discontinuity {
806 (0u8, false, false, false)
808 } else {
809 let is_duplicate = cc == state.last_cc && has_payload;
810 let mut should_emit_dup = false;
811 let mut should_emit_cc = false;
812
813 if is_duplicate {
814 if state.dup_used {
815 should_emit_dup = true;
816 }
817 } else {
818 state.dup_used = false;
819 let expected = if has_payload {
820 (state.last_cc.wrapping_add(1)) & 0x0F
821 } else {
822 state.last_cc
823 };
824 if cc != expected {
825 should_emit_cc = true;
826 }
827 }
828
829 (
830 if has_payload {
831 (state.last_cc.wrapping_add(1)) & 0x0F
832 } else {
833 state.last_cc
834 },
835 is_duplicate,
836 should_emit_dup,
837 should_emit_cc,
838 )
839 }
840 };
841
842 if should_emit_dup {
844 self.emit(
845 Indicator::ContinuityCountError,
846 Some(pid),
847 t,
848 format!("second consecutive duplicate on PID 0x{pid:04X} (cc={cc})"),
849 );
850 }
851 if should_emit_cc {
852 self.emit(
853 Indicator::ContinuityCountError,
854 Some(pid),
855 t,
856 format!("expected cc={expected}, got {cc} on PID 0x{pid:04X}"),
857 );
858 }
859
860 let state = self.cc_states.get_mut(&pid).unwrap();
862 if discontinuity {
863 state.last_cc = cc;
864 state.had_payload = has_payload;
865 state.dup_used = false;
866 } else if is_duplicate {
867 state.dup_used = true;
869 } else {
870 state.dup_used = false;
871 state.last_cc = cc;
872 state.had_payload = has_payload;
873 }
874 }
875
876 fn check_crc_and_process_pat(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
878 self.check_crc_for_section(section_bytes, pid, t);
880
881 self.process_pat_section(section_bytes, t);
882 }
883
884 fn check_crc_and_process_pmt(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
886 self.check_crc_for_section(section_bytes, pid, t);
888
889 self.process_pmt_section(section_bytes, pid, t);
890 }
891
892 fn check_crc_for_section(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
897 let section = match Section::parse(section_bytes) {
898 Ok(s) => s,
899 Err(_) => return,
900 };
901
902 if let Err(mpeg_ts::error::Error::CrcMismatch { .. }) = section.validate_crc(section_bytes)
904 {
905 self.emit(
906 Indicator::CrcError,
907 Some(pid),
908 t,
909 format!(
910 "CRC-32 mismatch on PID 0x{:04X} (table_id 0x{:02X})",
911 pid, section.table_id
912 ),
913 );
914 }
915 }
916
917 fn check_crc_for_si(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
919 self.check_crc_for_section(section_bytes, pid, t);
920 }
921
922 fn check_cat_table_id(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
925 if pid != PID_CAT {
926 return;
927 }
928 let section = match Section::parse(section_bytes) {
929 Ok(s) => s,
930 Err(_) => return,
931 };
932 if section.table_id == CAT_TABLE_ID {
933 self.cat_seen = true;
935 self.scrambled_without_cat_reported = false;
938 } else {
939 self.emit(
940 Indicator::CatError,
941 Some(PID_CAT),
942 t,
943 format!(
944 "section with table_id 0x{:02X} on PID 0x0001 (expected 0x01 for CAT)",
945 section.table_id
946 ),
947 );
948 }
949 }
950
951 fn process_pat_section(&mut self, section_bytes: &[u8], t: Duration) {
953 let section = match Section::parse(section_bytes) {
954 Ok(s) => s,
955 Err(_) => return,
956 };
957
958 if section.table_id != PAT_TABLE_ID {
960 self.emit(
961 Indicator::PatError2,
962 Some(PID_PAT),
963 t,
964 format!(
965 "section with table_id 0x{:02X} on PID 0x0000 (expected 0x00)",
966 section.table_id
967 ),
968 );
969 return;
970 }
971
972 let pat = match PatSection::parse(section_bytes) {
974 Ok(p) => p,
975 Err(_) => return,
976 };
977
978 for entry in pat.programmes() {
980 let pmt_pid = entry.pid;
981 self.pmt_trackings
982 .entry(pmt_pid)
983 .or_insert_with(|| PmtTracking {
984 timer: PresenceTimer {
985 last_seen: t,
986 reported: false,
987 },
988 reassembler: SectionReassembler::default(),
989 });
990 }
991 }
992
993 fn process_pmt_section(&mut self, section_bytes: &[u8], _pid: u16, t: Duration) {
995 let section = match Section::parse(section_bytes) {
996 Ok(s) => s,
997 Err(_) => return,
998 };
999
1000 let pmt_table_id: u8 = dvb_si::tables::pmt::TABLE_ID;
1005 if section.table_id != pmt_table_id {
1006 return;
1007 }
1008
1009 let pmt = match PmtSection::parse(section_bytes) {
1011 Ok(p) => p,
1012 Err(_) => return,
1013 };
1014
1015 let mut new_es_pids: Vec<u16> = Vec::new();
1017 if pmt.pcr_pid != PID_NULL && !self.es_trackings.contains_key(&pmt.pcr_pid) {
1018 new_es_pids.push(pmt.pcr_pid);
1019 }
1020 for stream in &pmt.streams {
1021 let es_pid = stream.elementary_pid;
1022 if !self.es_trackings.contains_key(&es_pid) {
1023 new_es_pids.push(es_pid);
1024 }
1025 }
1026
1027 for es_pid in new_es_pids {
1028 self.es_trackings.insert(
1029 es_pid,
1030 EsTracking {
1031 timer: PresenceTimer {
1032 last_seen: t,
1033 reported: false,
1034 },
1035 },
1036 );
1037 }
1038 }
1039
1040 fn check_pcr(&mut self, pid: u16, pcr_27mhz: u64, discontinuity: bool, t: Duration) {
1042 let state = self.pcr_states.entry(pid).or_insert_with(|| PcrState {
1043 last_pcr_27mhz: 0,
1044 last_pcr_time: Duration::ZERO,
1045 initialised: false,
1046 });
1047
1048 if !state.initialised {
1049 state.last_pcr_27mhz = pcr_27mhz;
1050 state.last_pcr_time = t;
1051 state.initialised = true;
1052 return;
1053 }
1054
1055 let last_pcr_time = state.last_pcr_time;
1057 let last_pcr_27mhz = state.last_pcr_27mhz;
1058
1059 let rep_interval = t.saturating_sub(last_pcr_time);
1062 let should_emit_rep = rep_interval > self.config.pcr_repetition_limit;
1063
1064 let delta =
1067 (pcr_27mhz.wrapping_add(PCR_MODULUS_27MHZ) - last_pcr_27mhz) % PCR_MODULUS_27MHZ;
1068 let delta_ms = delta * 1000 / CLOCK_27MHZ;
1069 let limit_ms = self.config.pcr_discontinuity_limit.as_millis() as u64;
1070 let should_emit_disc = delta_ms > limit_ms && !discontinuity;
1071
1072 if should_emit_rep {
1074 self.emit(
1075 Indicator::PcrRepetitionError,
1076 Some(pid),
1077 t,
1078 format!(
1079 "PCR interval {} ms exceeds limit {} ms on PID 0x{:04X}",
1080 rep_interval.as_millis(),
1081 self.config.pcr_repetition_limit.as_millis(),
1082 pid
1083 ),
1084 );
1085 }
1086 if should_emit_disc {
1087 self.emit(
1088 Indicator::PcrDiscontinuityError,
1089 Some(pid),
1090 t,
1091 format!(
1092 "PCR delta {delta_ms} ms exceeds limit {limit_ms} ms on PID 0x{pid:04X} without discontinuity_indicator"
1093 ),
1094 );
1095 }
1096
1097 let state = self.pcr_states.get_mut(&pid).unwrap();
1099 state.last_pcr_27mhz = pcr_27mhz;
1100 state.last_pcr_time = t;
1101 }
1102
1103 fn check_pts(&mut self, pid: u16, payload: &[u8], t: Duration) {
1108 if payload.len() < PES_FLAGS_OFFSET + 2 {
1110 return;
1111 }
1112 if payload[0] != PES_PREFIX_0 || payload[1] != PES_PREFIX_1 || payload[2] != PES_PREFIX_2 {
1113 return;
1114 }
1115
1116 let flags_byte = payload[PES_FLAGS_OFFSET];
1118 if (flags_byte >> 6) != 0b10 {
1119 return;
1120 }
1121
1122 let pts_dts_flags = payload[PES_FLAGS_OFFSET + 1] & PES_PTS_DTS_FLAGS_MASK;
1124 let pts_present = (pts_dts_flags & PES_PTS_PRESENT) != 0;
1125 if !pts_present {
1126 return;
1127 }
1128
1129 let state = self.pts_states.entry(pid).or_insert_with(|| PtsState {
1130 last_pts_time: Duration::ZERO,
1131 armed: false,
1132 });
1133
1134 if !state.armed {
1135 state.last_pts_time = t;
1137 state.armed = true;
1138 return;
1139 }
1140
1141 let last_pts_time = state.last_pts_time;
1143 let pts_interval = t.saturating_sub(last_pts_time);
1144 let should_emit = pts_interval > self.config.pts_repetition_limit;
1145
1146 if should_emit {
1147 self.emit(
1148 Indicator::PtsError,
1149 Some(pid),
1150 t,
1151 format!(
1152 "PTS interval {} ms exceeds limit {} ms on PID 0x{:04X}",
1153 pts_interval.as_millis(),
1154 self.config.pts_repetition_limit.as_millis(),
1155 pid
1156 ),
1157 );
1158 }
1159
1160 let state = self.pts_states.get_mut(&pid).unwrap();
1162 state.last_pts_time = t;
1163 }
1164
1165 fn update_si_repetition(&mut self, section_bytes: &[u8], _pid: u16, t: Duration) {
1169 let table_id = match Section::parse(section_bytes) {
1170 Ok(s) => s.table_id,
1171 Err(_) => return,
1172 };
1173
1174 let is_tracked = table_id == NIT_ACTUAL_TABLE_ID
1175 || table_id == SDT_ACTUAL_TABLE_ID
1176 || table_id == EIT_PF_ACTUAL_TABLE_ID
1177 || table_id == TDT_TABLE_ID;
1178
1179 if !is_tracked {
1180 return;
1181 }
1182
1183 let timer = self
1184 .si_timers
1185 .entry(table_id)
1186 .or_insert_with(|| SiRepetitionTimer {
1187 last_seen: Duration::ZERO,
1188 reported: false,
1189 armed: false,
1190 });
1191
1192 timer.last_seen = t;
1193 timer.reported = false;
1194 timer.armed = true;
1195 }
1196
1197 fn check_presence_timeouts(&mut self, t: Duration) {
1199 if t.saturating_sub(self.pat_timer.last_seen) > self.config.pat_max_interval
1201 && !self.pat_timer.reported
1202 {
1203 self.pat_timer.reported = true;
1204 self.emit(
1205 Indicator::PatError2,
1206 Some(PID_PAT),
1207 t,
1208 format!(
1209 "no PAT section within {} ms",
1210 self.config.pat_max_interval.as_millis()
1211 ),
1212 );
1213 }
1214
1215 let pmt_timeouts: Vec<(u16, u64)> = self
1218 .pmt_trackings
1219 .iter()
1220 .filter_map(|(&pid, tracking)| {
1221 if t.saturating_sub(tracking.timer.last_seen) > self.config.pmt_max_interval
1222 && !tracking.timer.reported
1223 {
1224 Some((pid, self.config.pmt_max_interval.as_millis() as u64))
1225 } else {
1226 None
1227 }
1228 })
1229 .collect();
1230 for (pid, interval_ms) in pmt_timeouts {
1231 if let Some(tracking) = self.pmt_trackings.get_mut(&pid) {
1232 tracking.timer.reported = true;
1233 }
1234 self.emit(
1235 Indicator::PmtError2,
1236 Some(pid),
1237 t,
1238 format!("no PMT section on program_map_PID 0x{pid:04X} within {interval_ms} ms"),
1239 );
1240 }
1241
1242 let pid_timeouts: Vec<(u16, u64)> = self
1244 .es_trackings
1245 .iter()
1246 .filter_map(|(&pid, tracking)| {
1247 if t.saturating_sub(tracking.timer.last_seen) > self.config.pid_error_period
1248 && !tracking.timer.reported
1249 {
1250 Some((pid, self.config.pid_error_period.as_secs()))
1251 } else {
1252 None
1253 }
1254 })
1255 .collect();
1256 for (pid, period_secs) in pid_timeouts {
1257 if let Some(tracking) = self.es_trackings.get_mut(&pid) {
1258 tracking.timer.reported = true;
1259 }
1260 self.emit(
1261 Indicator::PidError,
1262 Some(pid),
1263 t,
1264 format!("referenced PID 0x{pid:04X} absent for > {period_secs} s"),
1265 );
1266 }
1267
1268 let si_timeouts: Vec<(u8, u64, u16, u64)> = self
1271 .si_timers
1272 .iter()
1273 .filter_map(|(&table_id, timer)| {
1274 if !timer.armed || timer.reported {
1275 return None;
1276 }
1277 let (limit, pid) = match table_id {
1278 NIT_ACTUAL_TABLE_ID => (self.config.si_nit_interval, PID_NIT),
1279 SDT_ACTUAL_TABLE_ID => (self.config.si_sdt_interval, PID_SDT_BAT),
1280 EIT_PF_ACTUAL_TABLE_ID => (self.config.si_eit_pf_interval, PID_EIT),
1281 TDT_TABLE_ID => (self.config.si_tdt_interval, PID_TDT_TOT),
1282 _ => return None,
1283 };
1284 let interval = t.saturating_sub(timer.last_seen);
1285 if interval > limit {
1286 Some((
1287 table_id,
1288 interval.as_millis() as u64,
1289 pid,
1290 limit.as_millis() as u64,
1291 ))
1292 } else {
1293 None
1294 }
1295 })
1296 .collect();
1297 for (table_id, interval_ms, pid, limit_ms) in si_timeouts {
1298 if let Some(timer) = self.si_timers.get_mut(&table_id) {
1299 timer.reported = true;
1300 }
1301 let table_name = match table_id {
1302 NIT_ACTUAL_TABLE_ID => "NIT_actual",
1303 SDT_ACTUAL_TABLE_ID => "SDT_actual",
1304 EIT_PF_ACTUAL_TABLE_ID => "EIT_P/F_actual",
1305 TDT_TABLE_ID => "TDT",
1306 _ => "unknown",
1307 };
1308 self.emit(
1309 Indicator::SiRepetitionError,
1310 Some(pid),
1311 t,
1312 format!("{table_name} repetition interval {interval_ms} ms exceeds {limit_ms} ms"),
1313 );
1314 }
1315 }
1316}
1317
1318impl Default for ConformanceMonitor {
1319 fn default() -> Self {
1320 Self::new()
1321 }
1322}
1323
1324#[cfg(test)]
1325mod tests;