dvb_conformance/lib.rs
1//! ETSI TR 101 290 v1.4.1 transport-stream conformance monitor.
2//!
3//! Implements the **first-priority** (Table 5.0a, indicators 1.1–1.6),
4//! **second-priority** (Table 5.0b, indicators 2.1–2.3b, 2.5–2.6), and
5//! **third-priority** (Table 5.0c, indicators 3.1–3.10) indicator sets —
6//! see `docs/tr_101_290.md` for the full spec transcription and the
7//! crate-coverage mapping.
8//!
9//! ## T-STD buffer model (indicators 3.3, 3.9, 3.10)
10//!
11//! A partial ISO/IEC 13818-1 T-STD buffer model (see `src/tstd.rs`) drives
12//! the buffer-model indicators:
13//!
14//! - **3.3 `BufferError`**: TBsys overflow detection (512-byte buffer at
15//! 1 Mbit/s drain, fed at PSI section completion). TBn overflow is deferred
16//! — it requires the coded bitrate `Rxn` from descriptors.
17//! - **3.9 `EmptyBufferError`**: TBn (per-PID) and TBsys (global) empty at
18//! least once per second. MBn empty check is deferred.
19//! - **3.10 `Data_delay_error`**: Data delay > 1 s through TBn and TBsys.
20//! Still-picture 60 s threshold is tracked but not yet differentiated.
21//! - **2.4 `PcrAccuracyError`**: not implemented — requires hardware arrival
22//! timestamps with ±500 ns resolution. The variant exists for documentation
23//! completeness only.
24//!
25//! MBn/EBn/Bn/Bsys buffer modelling is deferred — it requires codec-level
26//! buffer sizes from descriptors (multiplex_buffer_descriptor,
27//! smoothing_buffer_descriptor) not yet parsed by the monitor.
28//!
29//! Feasible but deferred: the 25 ms minimum-gap dimension shared by 3.1.a /
30//! 3.2 / 3.5.a / 3.6.a / 3.7 / 3.8 (needs per-`(table_id, section_number)`
31//! tracking to avoid false positives on dense multi-section tables); the
32//! `_other` repetition sub-clauses 3.1.b / 3.5.b / 3.6.b (need TR 101 211
33//! interval rules); the EIT P/F pairing check 3.6.c.
34//!
35//! # Caller-supplied time
36//!
37//! [`ConformanceMonitor::feed`] takes a [`core::time::Duration`] timestamp
38//! alongside each TS packet. All presence/absence timeout checks (1.3.a, 1.5.a,
39//! 1.6, 2.3a, 2.3b, 2.5, 3.2) are evaluated against this clock. The caller
40//! must ensure that timestamps are **monotonic non-decreasing** across calls;
41//! the monitor does not enforce this but non-monotonic timestamps will produce
42//! spurious events.
43//!
44//! # References
45//!
46//! - ETSI TR 101 290 v1.4.1 (2020-06), §5.2.1, Table 5.0a
47//! - ETSI TR 101 290 v1.4.1 (2020-06), §5.2.2, Table 5.0b
48//! - ETSI TR 101 290 v1.4.1 (2020-06), §5.2.3, Table 5.0c
49//! - ISO/IEC 13818-1 (MPEG-2 Systems)
50
51#![cfg_attr(not(feature = "std"), no_std)]
52#![cfg_attr(docsrs, feature(doc_cfg))]
53// Runnable examples, embedded so they render on docs.rs and stay in sync with
54// the actual `examples/*.rs` files (shown, not compiled).
55#![doc = "\n# Examples\n"]
56#![doc = "Two runnable examples ship with this crate (`cargo run -p dvb-conformance --example <name>`).\n"]
57#![doc = "\n## `monitor_stream`\n\n```rust,ignore"]
58#![doc = include_str!("../examples/monitor_stream.rs")]
59#![doc = "```\n\n## `priority_breakdown`\n\n```rust,ignore"]
60#![doc = include_str!("../examples/priority_breakdown.rs")]
61#![doc = "```"]
62extern crate alloc;
63
64mod tstd;
65
66use alloc::collections::BTreeMap;
67use alloc::format;
68use alloc::string::String;
69use alloc::vec::Vec;
70use core::time::Duration;
71
72use tstd::TstdModel;
73
74use broadcast_common::Parse;
75use dvb_si::tables::pat::{PatSection, TABLE_ID as PAT_TABLE_ID};
76use dvb_si::tables::pmt::PmtSection;
77use mpeg_ts::section::Section;
78use mpeg_ts::ts::{SectionReassembler, TsPacket};
79
80// ── Named PID constants ─────────────────────────────────────────────────────
81
82/// PID 0x0000 — Program Association Table (ISO/IEC 13818-1 §2.4.4.3).
83const PID_PAT: u16 = 0x0000;
84/// PID 0x0001 — Conditional Access Table (ISO/IEC 13818-1 §2.4.4.5).
85const PID_CAT: u16 = 0x0001;
86/// PID 0x0010 — Network Information Table (EN 300 468 §5.2.1).
87const PID_NIT: u16 = 0x0010;
88/// PID 0x0011 — SDT/BAT (EN 300 468 §5.2.2 / §5.2.3).
89const PID_SDT_BAT: u16 = 0x0011;
90/// PID 0x0012 — Event Information Table (EN 300 468 §5.2.4).
91const PID_EIT: u16 = 0x0012;
92/// PID 0x0013 — Running Status Table (EN 300 468 §5.2.8).
93const PID_RST: u16 = 0x0013;
94/// PID 0x0014 — TDT/TOT (EN 300 468 §5.2.5 / §5.2.6).
95const PID_TDT_TOT: u16 = 0x0014;
96/// PID 0x1FFF — Null/padding packets (ISO/IEC 13818-1 §2.4.3.3).
97const PID_NULL: u16 = 0x1FFF;
98
99/// Sync byte value (ISO/IEC 13818-1 §2.4.3.3).
100const SYNC_BYTE: u8 = 0x47;
101
102/// Well-known SI/PSI PIDs on which CRC-checked long-form sections appear.
103const SI_PIDS: [u16; 7] = [
104 PID_PAT,
105 PID_CAT,
106 PID_NIT,
107 PID_SDT_BAT,
108 PID_EIT,
109 PID_RST,
110 PID_TDT_TOT,
111];
112
113/// Reserved-for-future-use PID range (ISO/IEC 13818-1 Table 2-3) — exempt
114/// from Unreferenced_PID (TR 101 290 v1.4.1 Table 5.0c indicator 3.4).
115const RESERVED_PID_MIN: u16 = 0x0002;
116/// Upper (inclusive) bound of the reserved PID range.
117const RESERVED_PID_MAX: u16 = 0x000F;
118
119// ── Default timing constants ────────────────────────────────────────────────
120
121/// TR 101 290 v1.4.1 Table 5.0a note 3 / TS 101 154 §4.1.7 — PAT maximum
122/// interval (0.5 s per Table 5.0a row 1.3.a; TS 101 154 recommends ≤ 100 ms).
123const DEFAULT_PAT_MAX_INTERVAL_MS: u64 = 500;
124
125/// TR 101 290 v1.4.1 Table 5.0a row 1.5.a / note 3 — PMT maximum interval.
126const DEFAULT_PMT_MAX_INTERVAL_MS: u64 = 500;
127
128/// TR 101 290 v1.4.1 §5.2.1 accompanying text (1.6) — PID_error period.
129const DEFAULT_PID_ERROR_PERIOD_SECS: u64 = 5;
130
131/// TR 101 290 v1.4.1 §5.2.1 accompanying text (1.1) — sync acquisition
132/// threshold: five consecutive correct sync bytes.
133const DEFAULT_SYNC_ACQUIRE_PACKETS: u8 = 5;
134
135/// TR 101 290 v1.4.1 §5.2.1 accompanying text (1.1) — sync loss threshold:
136/// two or more consecutive corrupted sync bytes.
137const DEFAULT_SYNC_LOSS_PACKETS: u8 = 2;
138
139/// TR 101 290 v1.4.1 Table 5.0b indicator 2.3a / note 2 — PCR maximum
140/// repetition interval (100 ms; note 2 removed the 40 ms limit).
141const DEFAULT_PCR_REPETITION_LIMIT_MS: u64 = 100;
142
143/// TR 101 290 v1.4.1 Table 5.0b indicator 2.3b — PCR discontinuity indicator
144/// maximum interval (100 ms).
145const DEFAULT_PCR_DISCONTINUITY_LIMIT_MS: u64 = 100;
146
147/// TR 101 290 v1.4.1 Table 5.0b indicator 2.5 / note 3 — PTS maximum
148/// repetition interval (700 ms; not applied to still pictures).
149const DEFAULT_PTS_REPETITION_LIMIT_MS: u64 = 700;
150
151/// TR 101 290 v1.4.1 Table 5.0c indicator 3.2 — NIT_actual maximum repetition
152/// interval (10 s; EN 300 468 §5.2.1).
153const DEFAULT_SI_NIT_INTERVAL_SECS: u64 = 10;
154
155/// TR 101 290 v1.4.1 Table 5.0c indicator 3.2 — SDT_actual maximum repetition
156/// interval (2 s; EN 300 468 §5.2.2).
157const DEFAULT_SI_SDT_INTERVAL_SECS: u64 = 2;
158
159/// TR 101 290 v1.4.1 Table 5.0c indicator 3.2 — EIT P/F actual maximum
160/// repetition interval (2 s; EN 300 468 §5.2.4).
161const DEFAULT_SI_EIT_PF_INTERVAL_SECS: u64 = 2;
162
163/// TR 101 290 v1.4.1 Table 5.0c indicator 3.2 — TDT maximum repetition
164/// interval (30 s; EN 300 468 §5.2.5).
165const DEFAULT_SI_TDT_INTERVAL_SECS: u64 = 30;
166
167/// TR 101 290 v1.4.1 Table 5.0c indicator 3.4 / note 1 — Unreferenced_PID
168/// persistence threshold (0,5 s). Transitions shorter than this (e.g. a PID
169/// seen briefly before its PMT arrives) are not errors.
170const DEFAULT_UNREFERENCED_PID_PERIOD_MS: u64 = 500;
171
172// ── PCR / PES constants ─────────────────────────────────────────────────────
173
174/// PCR modulus on the 27 MHz clock: `2^33 × 300` (33-bit base × 300 ticks).
175/// ISO/IEC 13818-1 §2.4.3.5 — PCR wraps modulo this value.
176const PCR_MODULUS_27MHZ: u64 = (1u64 << 33) * 300;
177
178/// 27 MHz clock rate (ticks per second).
179const CLOCK_27MHZ: u64 = 27_000_000;
180
181/// PES start-code prefix byte 0 (ISO/IEC 13818-1 §2.4.3.7 Table 2-18).
182const PES_PREFIX_0: u8 = 0x00;
183/// PES start-code prefix byte 1.
184const PES_PREFIX_1: u8 = 0x00;
185/// PES start-code prefix byte 2.
186const PES_PREFIX_2: u8 = 0x01;
187
188/// Offset of the PES header `marker_bits + flags` byte relative to the PES
189/// packet start (byte 6: `'10' + PES_scrambling_control + …`).
190const PES_FLAGS_OFFSET: usize = 6;
191
192/// Mask for the `PTS_DTS_flags` field within the PES header byte at offset 7
193/// (bits `[7:6]` — `0b10` means PTS present, `0b11` means PTS+DTS).
194const PES_PTS_DTS_FLAGS_MASK: u8 = 0b1100_0000;
195
196/// Value indicating PTS is present in `PTS_DTS_flags` (bit 7 set).
197const PES_PTS_PRESENT: u8 = 0b1000_0000;
198
199/// CAT `table_id` value (ISO/IEC 13818-1 §2.4.4.5).
200const CAT_TABLE_ID: u8 = dvb_si::table_id::TableId::Cat as u8;
201
202/// NIT_actual `table_id` (EN 300 468 §5.2.1, table_id 0x40).
203const NIT_ACTUAL_TABLE_ID: u8 = dvb_si::table_id::TableId::NetworkInformationActual as u8;
204
205/// SDT_actual `table_id` (EN 300 468 §5.2.2, table_id 0x42).
206const SDT_ACTUAL_TABLE_ID: u8 = dvb_si::table_id::TableId::ServiceDescriptionActual as u8;
207
208/// EIT P/F actual `table_id` (EN 300 468 §5.2.4, table_id 0x4E).
209const EIT_PF_ACTUAL_TABLE_ID: u8 = dvb_si::table_id::TableId::EventInformationPfActual as u8;
210
211/// TDT `table_id` (EN 300 468 §5.2.5, table_id 0x70).
212const TDT_TABLE_ID: u8 = dvb_si::table_id::TableId::TimeAndDate as u8;
213
214/// NIT_other `table_id` (EN 300 468 §5.2.1, table_id 0x41).
215const NIT_OTHER_TABLE_ID: u8 = dvb_si::table_id::TableId::NetworkInformationOther as u8;
216
217/// SDT_other `table_id` (EN 300 468 §5.2.3, table_id 0x46).
218const SDT_OTHER_TABLE_ID: u8 = dvb_si::table_id::TableId::ServiceDescriptionOther as u8;
219
220/// BAT `table_id` (EN 300 468 §5.2.2, table_id 0x4A — allowed alongside SDT
221/// on PID 0x0011).
222const BAT_TABLE_ID: u8 = dvb_si::table_id::TableId::BouquetAssociation as u8;
223
224/// EIT P/F other `table_id` (EN 300 468 §5.2.4, table_id 0x4F).
225const EIT_PF_OTHER_TABLE_ID: u8 = dvb_si::table_id::TableId::EventInformationPfOther as u8;
226
227/// RST `table_id` (EN 300 468 §5.2.8, table_id 0x71).
228const RST_TABLE_ID: u8 = dvb_si::table_id::TableId::RunningStatus as u8;
229
230/// Stuffing table (`ST`) `table_id` — allowed filler on every SI PID covered
231/// by Table 5.0c (EN 300 468 §5.2.9, table_id 0x72).
232const STUFFING_TABLE_ID: u8 = dvb_si::table_id::TableId::Stuffing as u8;
233
234/// TOT `table_id` (EN 300 468 §5.2.6, table_id 0x73 — allowed alongside TDT
235/// on PID 0x0014).
236const TOT_TABLE_ID: u8 = dvb_si::table_id::TableId::TimeOffset as u8;
237
238// ── Public types ─────────────────────────────────────────────────────────────
239
240/// Severity tier per TR 101 290 §5.2 (Tables 5.0a/5.0b/5.0c).
241#[derive(Debug, Clone, Copy, PartialEq, Eq)]
242#[cfg_attr(feature = "serde", derive(serde::Serialize))]
243#[non_exhaustive]
244pub enum Priority {
245 /// Table 5.0a — necessary for de-codability.
246 First,
247 /// Table 5.0b — recommended for continuous or periodic monitoring.
248 Second,
249 /// Table 5.0c — application-dependant monitoring.
250 Third,
251}
252
253impl Priority {
254 /// Human-readable spec label (TR 101 290 §5.2, Tables 5.0a/5.0b/5.0c).
255 #[must_use]
256 pub fn name(&self) -> &'static str {
257 match self {
258 Self::First => "first priority",
259 Self::Second => "second priority",
260 Self::Third => "third priority",
261 }
262 }
263}
264broadcast_common::impl_spec_display!(Priority);
265
266/// A TR 101 290 measurement indicator.
267///
268/// `#[non_exhaustive]` — additional Priority-3 variants may be added later.
269#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
270#[cfg_attr(feature = "serde", derive(serde::Serialize))]
271#[non_exhaustive]
272pub enum Indicator {
273 // ── Priority 1 (Table 5.0a) ──────────────────────────────────────────
274 /// TR 101 290 v1.4.1 Table 5.0a indicator 1.1 — loss of synchronisation
275 /// with hysteresis.
276 TsSyncLoss,
277 /// TR 101 290 v1.4.1 Table 5.0a indicator 1.2 — sync_byte not equal 0x47.
278 SyncByteError,
279 /// TR 101 290 v1.4.1 Table 5.0a indicator 1.3.a — PAT_error_2.
280 PatError2,
281 /// TR 101 290 v1.4.1 Table 5.0a indicator 1.4 — Continuity_count_error.
282 ContinuityCountError,
283 /// TR 101 290 v1.4.1 Table 5.0a indicator 1.5.a — PMT_error_2.
284 PmtError2,
285 /// TR 101 290 v1.4.1 Table 5.0a indicator 1.6 — PID_error.
286 PidError,
287
288 // ── Priority 2 (Table 5.0b) ──────────────────────────────────────────
289 /// TR 101 290 v1.4.1 Table 5.0b indicator 2.1 — Transport_error.
290 TransportError,
291 /// TR 101 290 v1.4.1 Table 5.0b indicator 2.2 — CRC_error.
292 CrcError,
293 /// TR 101 290 v1.4.1 Table 5.0b indicator 2.3a — PCR_repetition_error.
294 PcrRepetitionError,
295 /// TR 101 290 v1.4.1 Table 5.0b indicator 2.3b —
296 /// PCR_discontinuity_indicator_error.
297 PcrDiscontinuityError,
298 /// TR 101 290 v1.4.1 Table 5.0b indicator 2.5 — PTS_error.
299 PtsError,
300 /// TR 101 290 v1.4.1 Table 5.0b indicator 2.6 — CAT_error.
301 CatError,
302
303 // ── Priority 3 (Table 5.0c) ──────────────────────────────────────────
304 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.1 — NIT_error (bad table_id
305 /// on PID 0x0010 + NIT_actual absence; the 25 ms min-gap dimension of
306 /// 3.1.a is deferred).
307 NitError,
308 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.2 — SI_repetition_error
309 /// (maximum interval dimension; minimum-gap deferred).
310 SiRepetitionError,
311 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.4 — Unreferenced_PID: a PID
312 /// persists longer than the presence threshold without being referenced
313 /// by the PAT/CAT/a PMT or one of the well-known SI PIDs.
314 UnreferencedPid,
315 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.5 — SDT_error (bad table_id
316 /// on PID 0x0011 + SDT_actual absence; the 25 ms min-gap dimension of
317 /// 3.5.a is deferred).
318 SdtError,
319 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.6 — EIT_error (bad table_id
320 /// on PID 0x0012 + EIT P/F actual absence; the 25 ms min-gap dimension of
321 /// 3.6.a and the P/F pairing check 3.6.c are deferred).
322 EitError,
323 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.7 — RST_error (bad table_id
324 /// on PID 0x0013; the 25 ms min-gap dimension is deferred).
325 RstError,
326 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.8 — TDT_error (bad table_id
327 /// on PID 0x0014 + TDT absence; the 25 ms min-gap dimension is deferred).
328 TdtError,
329
330 // ── Priority 2 (Table 5.0b) — T-STD ──────────────────────────────────
331 /// TR 101 290 v1.4.1 Table 5.0b indicator 2.4 — PCR_accuracy_error.
332 ///
333 /// **Not implemented**: requires hardware arrival timing with ±500 ns
334 /// resolution (ISO/IEC 13818-1 §2.4.2.2). A packet-index-derived arrival
335 /// estimate cannot honestly resolve 500 ns, and a false positive is worse
336 /// than a gap (c.f. the withdrawn `PtsCheck`). This variant exists for
337 /// documentation completeness only and is never emitted by the monitor.
338 PcrAccuracyError,
339
340 // ── Priority 3 (Table 5.0c) — T-STD ──────────────────────────────────
341 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.3 — Buffer_error.
342 ///
343 /// Currently checks `TB_buffering_error` (overflow of transport buffer
344 /// TBn, 512 bytes per ISO/IEC 13818-1 §2.4.2.3) and
345 /// `TBsys_buffering_error` (overflow of TBsys, 512 bytes). The remaining
346 /// sub-checks — `MB_buffering_error`, `EB_buffering_error`,
347 /// `B_buffering_error`, `Bsys_buffering_error` — require codec-dependent
348 /// buffer sizes from descriptors (multiplex_buffer_descriptor,
349 /// smoothing_buffer_descriptor) and are deferred until the monitor
350 /// parses those descriptors.
351 BufferError,
352
353 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.9 — Empty_buffer_error.
354 ///
355 /// Checks that TBn (transport buffer for each elementary stream) and
356 /// TBsys (system-information transport buffer) are empty at least once
357 /// per second. The MBn check (multiplexing buffer, leak method) is
358 /// deferred — it requires the full MBn buffer model with leak-rate
359 /// parameters from descriptors.
360 EmptyBufferError,
361
362 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.10 — Data_delay_error.
363 ///
364 /// Checks that data delay through the T-STD transport buffers (TBn,
365 /// TBsys) does not exceed 1 second (60 s for still-picture video).
366 /// The full end-to-end delay through MBn/EBn/Bn is not modelled
367 /// (deferred — see `Buffer_error` doc).
368 DataDelayError,
369}
370
371impl Indicator {
372 /// The priority tier this indicator belongs to.
373 #[must_use]
374 pub fn priority(self) -> Priority {
375 match self {
376 Self::TsSyncLoss
377 | Self::SyncByteError
378 | Self::PatError2
379 | Self::ContinuityCountError
380 | Self::PmtError2
381 | Self::PidError => Priority::First,
382 Self::TransportError
383 | Self::CrcError
384 | Self::PcrRepetitionError
385 | Self::PcrDiscontinuityError
386 | Self::PtsError
387 | Self::CatError
388 | Self::PcrAccuracyError => Priority::Second,
389 Self::NitError
390 | Self::SiRepetitionError
391 | Self::UnreferencedPid
392 | Self::SdtError
393 | Self::EitError
394 | Self::RstError
395 | Self::TdtError
396 | Self::BufferError
397 | Self::EmptyBufferError
398 | Self::DataDelayError => Priority::Third,
399 }
400 }
401
402 /// Verbatim indicator name from the TR 101 290 tables.
403 #[must_use]
404 pub fn name(self) -> &'static str {
405 match self {
406 Self::TsSyncLoss => "TS_sync_loss",
407 Self::SyncByteError => "Sync_byte_error",
408 Self::PatError2 => "PAT_error_2",
409 Self::ContinuityCountError => "Continuity_count_error",
410 Self::PmtError2 => "PMT_error_2",
411 Self::PidError => "PID_error",
412 Self::TransportError => "Transport_error",
413 Self::CrcError => "CRC_error",
414 Self::PcrRepetitionError => "PCR_repetition_error",
415 Self::PcrDiscontinuityError => "PCR_discontinuity_indicator_error",
416 Self::PtsError => "PTS_error",
417 Self::CatError => "CAT_error",
418 Self::NitError => "NIT_error",
419 Self::SiRepetitionError => "SI_repetition_error",
420 Self::UnreferencedPid => "Unreferenced_PID",
421 Self::SdtError => "SDT_error",
422 Self::EitError => "EIT_error",
423 Self::RstError => "RST_error",
424 Self::TdtError => "TDT_error",
425 Self::PcrAccuracyError => "PCR_accuracy_error",
426 Self::BufferError => "Buffer_error",
427 Self::EmptyBufferError => "Empty_buffer_error",
428 Self::DataDelayError => "Data_delay_error",
429 }
430 }
431
432 /// Clause citation from the spec.
433 #[must_use]
434 pub fn clause(self) -> &'static str {
435 match self {
436 Self::TsSyncLoss => "TR 101 290 v1.4.1 Table 5.0a indicator 1.1",
437 Self::SyncByteError => "TR 101 290 v1.4.1 Table 5.0a indicator 1.2",
438 Self::PatError2 => "TR 101 290 v1.4.1 Table 5.0a indicator 1.3.a",
439 Self::ContinuityCountError => "TR 101 290 v1.4.1 Table 5.0a indicator 1.4",
440 Self::PmtError2 => "TR 101 290 v1.4.1 Table 5.0a indicator 1.5.a",
441 Self::PidError => "TR 101 290 v1.4.1 Table 5.0a indicator 1.6",
442 Self::TransportError => "TR 101 290 v1.4.1 Table 5.0b indicator 2.1",
443 Self::CrcError => "TR 101 290 v1.4.1 Table 5.0b indicator 2.2",
444 Self::PcrRepetitionError => "TR 101 290 v1.4.1 Table 5.0b indicator 2.3a",
445 Self::PcrDiscontinuityError => "TR 101 290 v1.4.1 Table 5.0b indicator 2.3b",
446 Self::PtsError => "TR 101 290 v1.4.1 Table 5.0b indicator 2.5",
447 Self::CatError => "TR 101 290 v1.4.1 Table 5.0b indicator 2.6",
448 Self::NitError => "TR 101 290 v1.4.1 Table 5.0c indicator 3.1",
449 Self::SiRepetitionError => "TR 101 290 v1.4.1 Table 5.0c indicator 3.2",
450 Self::UnreferencedPid => "TR 101 290 v1.4.1 Table 5.0c indicator 3.4",
451 Self::SdtError => "TR 101 290 v1.4.1 Table 5.0c indicator 3.5",
452 Self::EitError => "TR 101 290 v1.4.1 Table 5.0c indicator 3.6",
453 Self::RstError => "TR 101 290 v1.4.1 Table 5.0c indicator 3.7",
454 Self::TdtError => "TR 101 290 v1.4.1 Table 5.0c indicator 3.8",
455 Self::PcrAccuracyError => "TR 101 290 v1.4.1 Table 5.0b indicator 2.4",
456 Self::BufferError => "TR 101 290 v1.4.1 Table 5.0c indicator 3.3",
457 Self::EmptyBufferError => "TR 101 290 v1.4.1 Table 5.0c indicator 3.9",
458 Self::DataDelayError => "TR 101 290 v1.4.1 Table 5.0c indicator 3.10",
459 }
460 }
461}
462broadcast_common::impl_spec_display!(Indicator);
463
464/// One raised conformance error.
465#[derive(Debug, Clone, PartialEq, Eq)]
466#[cfg_attr(feature = "serde", derive(serde::Serialize))]
467#[non_exhaustive]
468pub struct ConformanceEvent {
469 /// The indicator that was raised.
470 pub indicator: Indicator,
471 /// Priority tier of the indicator.
472 pub priority: Priority,
473 /// PID the error concerns, when applicable.
474 pub pid: Option<u16>,
475 /// Caller timestamp of the packet that raised it.
476 pub at: Duration,
477 /// Human-readable specifics (e.g. "expected cc=5, got 7").
478 pub detail: String,
479}
480
481/// Diagnostic counters.
482#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
483#[cfg_attr(feature = "serde", derive(serde::Serialize))]
484#[non_exhaustive]
485pub struct Stats {
486 /// Total TS packets fed.
487 pub packets: u64,
488 /// Total conformance events raised.
489 pub events: u64,
490 /// Whether the monitor is currently in sync.
491 pub in_sync: bool,
492}
493
494/// Configurable hysteresis and timeout parameters.
495#[derive(Debug, Clone)]
496#[non_exhaustive]
497pub struct Config {
498 /// Maximum interval between PAT sections (Table 5.0a 1.3.a / note 3).
499 /// Default: 500 ms.
500 pub pat_max_interval: Duration,
501 /// Maximum interval between PMT sections per program_map_PID (1.5.a).
502 /// Default: 500 ms.
503 pub pmt_max_interval: Duration,
504 /// Period after which a referenced PID is considered absent (1.6).
505 /// Default: 5 s.
506 pub pid_error_period: Duration,
507 /// Consecutive good sync bytes to acquire sync (1.1).
508 /// Default: 5.
509 pub sync_acquire_packets: u8,
510 /// Consecutive bad sync bytes to declare sync loss (1.1).
511 /// Default: 2.
512 pub sync_loss_packets: u8,
513 /// Maximum interval between consecutive PCR values on a single PID
514 /// (Table 5.0b 2.3a / note 2). Default: 100 ms.
515 pub pcr_repetition_limit: Duration,
516 /// Maximum legal PCR delta (in time) without a signalled discontinuity
517 /// (Table 5.0b 2.3b). Default: 100 ms.
518 pub pcr_discontinuity_limit: Duration,
519 /// Maximum interval between consecutive PTS values on an elementary-stream
520 /// PID (Table 5.0b 2.5 / note 3). Default: 700 ms.
521 pub pts_repetition_limit: Duration,
522 /// Maximum repetition interval for NIT_actual sections (Table 5.0c 3.2 /
523 /// EN 300 468 §5.2.1). Default: 10 s.
524 pub si_nit_interval: Duration,
525 /// Maximum repetition interval for SDT_actual sections (Table 5.0c 3.2 /
526 /// EN 300 468 §5.2.2). Default: 2 s.
527 pub si_sdt_interval: Duration,
528 /// Maximum repetition interval for EIT P/F actual sections (Table 5.0c
529 /// 3.2 / EN 300 468 §5.2.4). Default: 2 s.
530 pub si_eit_pf_interval: Duration,
531 /// Maximum repetition interval for TDT sections (Table 5.0c 3.2 /
532 /// EN 300 468 §5.2.5). Default: 30 s.
533 pub si_tdt_interval: Duration,
534 /// Persistence threshold before an unreferenced PID is flagged
535 /// (Table 5.0c 3.4 / note 1). Default: 500 ms.
536 pub unreferenced_pid_period: Duration,
537}
538
539impl Default for Config {
540 fn default() -> Self {
541 Self {
542 pat_max_interval: Duration::from_millis(DEFAULT_PAT_MAX_INTERVAL_MS),
543 pmt_max_interval: Duration::from_millis(DEFAULT_PMT_MAX_INTERVAL_MS),
544 pid_error_period: Duration::from_secs(DEFAULT_PID_ERROR_PERIOD_SECS),
545 sync_acquire_packets: DEFAULT_SYNC_ACQUIRE_PACKETS,
546 sync_loss_packets: DEFAULT_SYNC_LOSS_PACKETS,
547 pcr_repetition_limit: Duration::from_millis(DEFAULT_PCR_REPETITION_LIMIT_MS),
548 pcr_discontinuity_limit: Duration::from_millis(DEFAULT_PCR_DISCONTINUITY_LIMIT_MS),
549 pts_repetition_limit: Duration::from_millis(DEFAULT_PTS_REPETITION_LIMIT_MS),
550 si_nit_interval: Duration::from_secs(DEFAULT_SI_NIT_INTERVAL_SECS),
551 si_sdt_interval: Duration::from_secs(DEFAULT_SI_SDT_INTERVAL_SECS),
552 si_eit_pf_interval: Duration::from_secs(DEFAULT_SI_EIT_PF_INTERVAL_SECS),
553 si_tdt_interval: Duration::from_secs(DEFAULT_SI_TDT_INTERVAL_SECS),
554 unreferenced_pid_period: Duration::from_millis(DEFAULT_UNREFERENCED_PID_PERIOD_MS),
555 }
556 }
557}
558
559// ── Internal per-PID state ───────────────────────────────────────────────────
560
561/// Per-PID continuity-counter tracking state.
562struct CcState {
563 last_cc: u8,
564 had_payload: bool,
565 dup_used: bool,
566 initialised: bool,
567}
568
569/// Timer state for a presence/absence check (shared by 1.3.a, 1.5.a, 1.6).
570struct PresenceTimer {
571 last_seen: Duration,
572 reported: bool,
573}
574
575/// State tracked for each program_map_PID signalled by the PAT.
576struct PmtTracking {
577 timer: PresenceTimer,
578 reassembler: SectionReassembler,
579}
580
581/// State tracked for each elementary-stream PID referenced by a PMT.
582struct EsTracking {
583 timer: PresenceTimer,
584}
585
586/// Per-PID PCR tracking state (indicators 2.3a, 2.3b).
587struct PcrState {
588 last_pcr_27mhz: u64,
589 last_pcr_time: Duration,
590 initialised: bool,
591}
592
593/// Per-PID PTS tracking state (indicator 2.5).
594struct PtsState {
595 last_pts_time: Duration,
596 armed: bool,
597}
598
599/// Per-PID section reassembly state for the well-known SI/PSI PIDs.
600struct SiReassembly {
601 reassembler: SectionReassembler,
602}
603
604/// Timer state for an SI table repetition-interval check (indicator 3.2).
605/// Lazily armed — only starts checking after the first section of that
606/// table_id is seen.
607struct SiRepetitionTimer {
608 last_seen: Duration,
609 reported: bool,
610 armed: bool,
611}
612
613/// Tracking state for a candidate Unreferenced_PID (indicator 3.4): the time
614/// this PID was first observed while NOT part of the referenced set.
615struct UnreferencedPidTracking {
616 first_seen: Duration,
617 reported: bool,
618}
619
620// ── ConformanceMonitor ───────────────────────────────────────────────────────
621
622/// ETSI TR 101 290 transport-stream conformance monitor.
623///
624/// Feed one TS packet at a time via [`feed`](Self::feed); each call returns
625/// the events raised by that packet. The monitor is synchronous and
626/// single-threaded — no interior mutability, no async.
627pub struct ConformanceMonitor {
628 config: Config,
629 events: Vec<ConformanceEvent>,
630 stats: Stats,
631
632 // Sync hysteresis state machine (1.1)
633 in_sync: bool,
634 good_run: u8,
635 bad_run: u8,
636
637 // Per-PID continuity counter (1.4)
638 cc_states: BTreeMap<u16, CcState>,
639
640 // PAT section reassembly + timing (1.3.a)
641 pat_reassembler: SectionReassembler,
642 pat_timer: PresenceTimer,
643
644 // PMT section reassembly + timing per program_map_PID (1.5.a)
645 pmt_trackings: BTreeMap<u16, PmtTracking>,
646
647 // Referenced ES PID timing (1.6)
648 es_trackings: BTreeMap<u16, EsTracking>,
649
650 // Well-known SI/PSI section reassembly + CRC checking (2.2)
651 si_reassemblies: BTreeMap<u16, SiReassembly>,
652
653 // Per-PID PCR tracking (2.3a, 2.3b)
654 pcr_states: BTreeMap<u16, PcrState>,
655
656 // Per-PID PTS tracking (2.5)
657 pts_states: BTreeMap<u16, PtsState>,
658
659 // CAT tracking (2.6)
660 cat_seen: bool,
661 scrambled_without_cat_reported: bool,
662
663 // SI repetition-interval timers keyed by table_id (3.2)
664 si_timers: BTreeMap<u8, SiRepetitionTimer>,
665
666 // Unreferenced_PID candidate tracking (3.4)
667 unreferenced_pid_timers: BTreeMap<u16, UnreferencedPidTracking>,
668
669 // T-STD buffer model (3.3, 3.9, 3.10)
670 tstd: TstdModel,
671}
672
673impl ConformanceMonitor {
674 /// Create a monitor with default configuration.
675 pub fn new() -> Self {
676 Self::with_config(Config::default())
677 }
678
679 /// Create a monitor with the given configuration.
680 pub fn with_config(config: Config) -> Self {
681 let mut si_reassemblies = BTreeMap::new();
682 for &pid in &SI_PIDS {
683 si_reassemblies.insert(
684 pid,
685 SiReassembly {
686 reassembler: SectionReassembler::default(),
687 },
688 );
689 }
690 Self {
691 config,
692 events: Vec::new(),
693 stats: Stats {
694 packets: 0,
695 events: 0,
696 in_sync: false,
697 },
698 in_sync: false,
699 good_run: 0,
700 bad_run: 0,
701 cc_states: BTreeMap::new(),
702 pat_reassembler: SectionReassembler::default(),
703 pat_timer: PresenceTimer {
704 last_seen: Duration::ZERO,
705 reported: false,
706 },
707 pmt_trackings: BTreeMap::new(),
708 es_trackings: BTreeMap::new(),
709 si_reassemblies,
710 pcr_states: BTreeMap::new(),
711 pts_states: BTreeMap::new(),
712 cat_seen: false,
713 scrambled_without_cat_reported: false,
714 si_timers: BTreeMap::new(),
715 unreferenced_pid_timers: BTreeMap::new(),
716 tstd: TstdModel::new(Duration::ZERO),
717 }
718 }
719
720 /// Feed ONE TS packet (any length; 188 expected) with its caller-supplied
721 /// arrival time `t`.
722 ///
723 /// `t` must be monotonic non-decreasing across calls (documented but not
724 /// enforced). Returns the events raised by this packet.
725 pub fn feed(&mut self, ts_packet: &[u8], t: Duration) -> &[ConformanceEvent] {
726 self.events.clear();
727 self.stats.packets += 1;
728
729 // ── Step 2: Sync byte check (1.2) ─────────────────────────────────
730 let sync_ok = !ts_packet.is_empty() && ts_packet[0] == SYNC_BYTE;
731 if !sync_ok {
732 self.emit(Indicator::SyncByteError, None, t, "sync_byte != 0x47");
733 }
734
735 // ── Step 3: Sync hysteresis state machine (1.1) ──────────────────
736 if sync_ok {
737 self.good_run = self.good_run.saturating_add(1);
738 self.bad_run = 0;
739 if !self.in_sync && self.good_run >= self.config.sync_acquire_packets {
740 self.in_sync = true;
741 }
742 } else {
743 self.bad_run = self.bad_run.saturating_add(1);
744 self.good_run = 0;
745 if self.in_sync && self.bad_run >= self.config.sync_loss_packets {
746 self.in_sync = false;
747 self.emit(
748 Indicator::TsSyncLoss,
749 None,
750 t,
751 "sync lost after hysteresis threshold",
752 );
753 }
754 }
755
756 // Per the doc: "If indicator 1.1 is activated then all other
757 // indicators are invalid." While not in sync, suppress all other
758 // indicators.
759 if !self.in_sync {
760 return &self.events;
761 }
762
763 // ── Step 4: Parse TS packet ───────────────────────────────────────
764 let packet = match TsPacket::parse(ts_packet) {
765 Ok(p) => p,
766 Err(_) => return &self.events,
767 };
768 let header = &packet.header;
769 let pid = header.pid;
770
771 // ── 2.1 Transport_error (Table 5.0b indicator 2.1) ──────────────
772 if header.tei {
773 self.emit(
774 Indicator::TransportError,
775 Some(pid),
776 t,
777 format!("transport_error_indicator set on PID 0x{pid:04X}"),
778 );
779 }
780
781 // ── Step 5: Continuity_count_error (1.4) ─────────────────────────
782 if pid != PID_NULL {
783 self.check_cc(
784 pid,
785 header.continuity_counter,
786 header.has_payload,
787 t,
788 ts_packet,
789 );
790 }
791
792 // ── Step 7: PAT_error_2 — scrambling check (1.3.a) ──────────────
793 if pid == PID_PAT && header.scrambling != 0 {
794 self.emit(
795 Indicator::PatError2,
796 Some(PID_PAT),
797 t,
798 format!(
799 "scrambling_control_field != 00 on PID 0x0000 (got {})",
800 header.scrambling
801 ),
802 );
803 }
804
805 // ── Step 8: PMT_error_2 — scrambling check (1.5.a) ──────────────
806 if self.pmt_trackings.contains_key(&pid) && header.scrambling != 0 {
807 self.emit(
808 Indicator::PmtError2,
809 Some(pid),
810 t,
811 format!("scrambling_control_field != 00 on program_map_PID 0x{pid:04X}"),
812 );
813 }
814
815 // ── 2.6 CAT_error — scrambled packet with no CAT (Table 5.0b 2.6)
816 //
817 // At stream start, scrambled packets may arrive before a CAT section
818 // has been acquired; this check fires once in that case. It re-arms
819 // (see `check_cat_table_id`) when a CAT later appears, so the error is
820 // re-detectable after a CAT section is seen.
821 if header.scrambling != 0 && !self.cat_seen && !self.scrambled_without_cat_reported {
822 self.scrambled_without_cat_reported = true;
823 self.emit(
824 Indicator::CatError,
825 Some(pid),
826 t,
827 format!("scrambled packet on PID 0x{pid:04X} but no CAT seen on PID 0x0001"),
828 );
829 }
830
831 // ── Step 6: Section reassembly — PAT ─────────────────────────────
832 if pid == PID_PAT && header.has_payload {
833 if let Some(payload) = packet.payload {
834 self.pat_reassembler.feed(payload, header.pusi);
835 }
836 self.pat_timer.last_seen = t;
837 self.pat_timer.reported = false;
838 while let Some(section_bytes) = self.pat_reassembler.pop_section() {
839 self.check_crc_and_process_pat(§ion_bytes, pid, t);
840 }
841 }
842
843 // ── Step 6b: Section reassembly — PMT PIDs ───────────────────────
844 if self.pmt_trackings.contains_key(&pid) && header.has_payload {
845 if let Some(payload) = packet.payload {
846 if let Some(tracking) = self.pmt_trackings.get_mut(&pid) {
847 tracking.reassembler.feed(payload, header.pusi);
848 }
849 }
850 let sections: Vec<_> = if let Some(tracking) = self.pmt_trackings.get_mut(&pid) {
851 tracking.timer.last_seen = t;
852 tracking.timer.reported = false;
853 core::iter::from_fn(|| tracking.reassembler.pop_section()).collect()
854 } else {
855 Vec::new()
856 };
857 for section_bytes in §ions {
858 self.check_crc_and_process_pmt(section_bytes, pid, t);
859 }
860 }
861
862 // ── Step 6c: Section reassembly — well-known SI/PSI PIDs (2.2) ───
863 // PAT and PMT PIDs are handled above (they have separate reassembly
864 // for P1 logic). Only process the non-PAT, non-PMT SI PIDs here.
865 if pid != PID_PAT
866 && !self.pmt_trackings.contains_key(&pid)
867 && self.si_reassemblies.contains_key(&pid)
868 && header.has_payload
869 {
870 if let Some(payload) = packet.payload {
871 if let Some(si_ra) = self.si_reassemblies.get_mut(&pid) {
872 si_ra.reassembler.feed(payload, header.pusi);
873 }
874 }
875 let sections: Vec<_> = if let Some(si_ra) = self.si_reassemblies.get_mut(&pid) {
876 core::iter::from_fn(|| si_ra.reassembler.pop_section()).collect()
877 } else {
878 Vec::new()
879 };
880 for section_bytes in §ions {
881 self.check_crc_for_si(section_bytes, pid, t);
882 self.check_cat_table_id(section_bytes, pid, t);
883 self.check_nit_table_id(section_bytes, pid, t);
884 self.check_sdt_table_id(section_bytes, pid, t);
885 self.check_eit_table_id(section_bytes, pid, t);
886 self.check_rst_table_id(section_bytes, pid, t);
887 self.check_tdt_table_id(section_bytes, pid, t);
888 self.update_si_repetition(section_bytes, pid, t);
889 }
890 }
891 // Also CRC-check completed PAT/PMT sections via the si_reassemblies
892 // map (these share the same PID). PAT and PMT already have their own
893 // reassemblers above — the si_reassemblies entries for those PIDs are
894 // not fed again. CRC checking for PAT/PMT is done inside
895 // check_crc_and_process_pat / check_crc_and_process_pmt.
896
897 // ── Step 9: PID_error — update last_seen for referenced PIDs ─────
898 if let Some(tracking) = self.es_trackings.get_mut(&pid) {
899 tracking.timer.last_seen = t;
900 tracking.timer.reported = false;
901 }
902
903 // ── 2.3a / 2.3b: PCR checks (Table 5.0b indicators 2.3a, 2.3b) ──
904 if let Some(Ok(af)) = packet.adaptation_field() {
905 if let Some(pcr) = af.pcr {
906 self.check_pcr(pid, pcr.as_27mhz(), af.discontinuity_indicator, t);
907 }
908 }
909
910 // ── 2.5: PTS check (Table 5.0b indicator 2.5) ───────────────────
911 if header.pusi
912 && header.scrambling == 0
913 && self.es_trackings.contains_key(&pid)
914 && header.has_payload
915 {
916 if let Some(payload) = packet.payload {
917 self.check_pts(pid, payload, t);
918 }
919 }
920
921 // ── 3.4: Unreferenced_PID bookkeeping ───────────────────────────
922 if pid != PID_NULL {
923 self.track_unreferenced_pid(pid, t);
924 }
925
926 // ── T-STD buffer model (3.3, 3.9, 3.10) ─────────────────────────
927 // ts_packet is a raw 188-byte (or larger) TS packet.
928 self.process_tstd(header, ts_packet, t);
929
930 // ── Presence-timeout evaluation (1.3.a, 1.5.a, 1.6, 3.4) ────────
931 self.check_presence_timeouts(t);
932
933 &self.events
934 }
935
936 /// Diagnostic counters.
937 pub fn stats(&self) -> Stats {
938 Stats {
939 in_sync: self.in_sync,
940 ..self.stats
941 }
942 }
943
944 // ── Internal helpers ──────────────────────────────────────────────────
945
946 fn emit(
947 &mut self,
948 indicator: Indicator,
949 pid: Option<u16>,
950 at: Duration,
951 detail: impl Into<String>,
952 ) {
953 let event = ConformanceEvent {
954 indicator,
955 priority: indicator.priority(),
956 pid,
957 at,
958 detail: detail.into(),
959 };
960 self.stats.events += 1;
961 self.events.push(event);
962 }
963
964 /// Continuity_count_error (1.4) check.
965 fn check_cc(&mut self, pid: u16, cc: u8, has_payload: bool, t: Duration, raw: &[u8]) {
966 // Check for discontinuity_indicator in the adaptation field BEFORE
967 // mutating cc_states (avoids holding the entry borrow across self.emit).
968 let discontinuity = if raw.len() >= 5 {
969 let b3 = raw[3];
970 let has_adaptation = (b3 & 0x20) != 0;
971 if has_adaptation {
972 let af_len = raw[4] as usize;
973 if af_len > 0 && raw.len() > 5 {
974 (raw[5] & 0x80) != 0
975 } else {
976 false
977 }
978 } else {
979 false
980 }
981 } else {
982 false
983 };
984
985 // Compute what we need from the existing state, then decide.
986 let (expected, is_duplicate, should_emit_dup, should_emit_cc) = {
987 let state = self.cc_states.entry(pid).or_insert_with(|| CcState {
988 last_cc: cc,
989 had_payload: has_payload,
990 dup_used: false,
991 initialised: false,
992 });
993
994 if !state.initialised {
995 state.last_cc = cc;
996 state.had_payload = has_payload;
997 state.dup_used = false;
998 state.initialised = true;
999 return;
1000 }
1001
1002 if discontinuity {
1003 // Will update state below — just signal no emit.
1004 (0u8, false, false, false)
1005 } else {
1006 let is_duplicate = cc == state.last_cc && has_payload;
1007 let mut should_emit_dup = false;
1008 let mut should_emit_cc = false;
1009
1010 if is_duplicate {
1011 if state.dup_used {
1012 should_emit_dup = true;
1013 }
1014 } else {
1015 state.dup_used = false;
1016 let expected = if has_payload {
1017 (state.last_cc.wrapping_add(1)) & 0x0F
1018 } else {
1019 state.last_cc
1020 };
1021 if cc != expected {
1022 should_emit_cc = true;
1023 }
1024 }
1025
1026 (
1027 if has_payload {
1028 (state.last_cc.wrapping_add(1)) & 0x0F
1029 } else {
1030 state.last_cc
1031 },
1032 is_duplicate,
1033 should_emit_dup,
1034 should_emit_cc,
1035 )
1036 }
1037 };
1038
1039 // Now emit events without holding a borrow on cc_states.
1040 if should_emit_dup {
1041 self.emit(
1042 Indicator::ContinuityCountError,
1043 Some(pid),
1044 t,
1045 format!("second consecutive duplicate on PID 0x{pid:04X} (cc={cc})"),
1046 );
1047 }
1048 if should_emit_cc {
1049 self.emit(
1050 Indicator::ContinuityCountError,
1051 Some(pid),
1052 t,
1053 format!("expected cc={expected}, got {cc} on PID 0x{pid:04X}"),
1054 );
1055 }
1056
1057 // Finally, update state.
1058 let state = self.cc_states.get_mut(&pid).unwrap();
1059 if discontinuity {
1060 state.last_cc = cc;
1061 state.had_payload = has_payload;
1062 state.dup_used = false;
1063 } else if is_duplicate {
1064 // First duplicate is legal; mark dup_used but do NOT update last_cc.
1065 state.dup_used = true;
1066 } else {
1067 state.dup_used = false;
1068 state.last_cc = cc;
1069 state.had_payload = has_payload;
1070 }
1071 }
1072
1073 /// CRC-check a completed section and, if on PID_PAT, process it.
1074 fn check_crc_and_process_pat(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
1075 // 2.2: CRC check on PAT section.
1076 self.check_crc_for_section(section_bytes, pid, t);
1077
1078 self.process_pat_section(section_bytes, t);
1079 }
1080
1081 /// CRC-check a completed section and, if on a PMT PID, process it.
1082 fn check_crc_and_process_pmt(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
1083 // 2.2: CRC check on PMT section.
1084 self.check_crc_for_section(section_bytes, pid, t);
1085
1086 self.process_pmt_section(section_bytes, pid, t);
1087 }
1088
1089 /// TR 101 290 v1.4.1 Table 5.0b indicator 2.2 — CRC_error.
1090 ///
1091 /// On any tracked PID, if a completed long-form section has a CRC
1092 /// mismatch, emit `CrcError`. Also feeds section bytes into TBsys
1093 /// for T-STD Buffer_error tracking (indicator 3.3).
1094 fn check_crc_for_section(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
1095 let section = match Section::parse(section_bytes) {
1096 Ok(s) => s,
1097 Err(_) => return,
1098 };
1099
1100 // Feed section bytes into TBsys for T-STD buffer tracking
1101 // (ISO/IEC 13818-1 §2.4.2.3: PSI sections pass through TBsys
1102 // before entering Bsys). TBsys has 512 bytes capacity and drains
1103 // at 1 Mbit/s. Overflow fires Buffer_error (3.3).
1104 self.feed_tbsys_section_bytes(section_bytes, pid, t);
1105
1106 // validate_crc returns Ok for short-form sections (no CRC to check).
1107 if let Err(mpeg_ts::error::Error::CrcMismatch { .. }) = section.validate_crc(section_bytes)
1108 {
1109 self.emit(
1110 Indicator::CrcError,
1111 Some(pid),
1112 t,
1113 format!(
1114 "CRC-32 mismatch on PID 0x{:04X} (table_id 0x{:02X})",
1115 pid, section.table_id
1116 ),
1117 );
1118 }
1119 }
1120
1121 /// Feed PSI section bytes into TBsys and check for overflow.
1122 ///
1123 /// TBsys (ISO/IEC 13818-1 §2.4.2.3): 512-byte transport buffer for system
1124 /// information. Receives PSI section bytes from the well-known SI/PSI PIDs.
1125 /// Drains at 1 Mbit/s (125 000 bytes/s). Overflow fires Buffer_error (3.3).
1126 fn feed_tbsys_section_bytes(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
1127 let section_len = section_bytes.len() as u64;
1128
1129 // Drain TBsys to current time using the fixed 1 Mbit/s leak rate.
1130 self.tstd.drain_tb_sys(t);
1131
1132 // Check if adding these section bytes would overflow TBsys.
1133 // TBsys has 512 bytes capacity at 1 Mbit/s drain.
1134 if self.tstd.tb_sys.would_overflow(section_len, t) {
1135 self.emit(
1136 Indicator::BufferError,
1137 Some(pid),
1138 t,
1139 format!(
1140 "TBsys overflow on PID 0x{pid:04X}: section {section_len} bytes exceeds {} byte capacity at 1 Mbit/s drain",
1141 tstd::TB_SYS_SIZE,
1142 ),
1143 );
1144 }
1145
1146 // Feed the section bytes into TBsys regardless (the buffer
1147 // model tracks occupancy for empty-interval and delay checks).
1148 let _overflow = self.tstd.tb_sys.feed(section_len, t);
1149
1150 // Indicator 3.9: TBsys empty at least once per second?
1151 if self.tstd.tb_sys.check_empty_interval(t) && !self.tstd.tb_sys_empty_reported {
1152 self.tstd.tb_sys_empty_reported = true;
1153 self.emit(
1154 Indicator::EmptyBufferError,
1155 Some(pid),
1156 t,
1157 format!(
1158 "TBsys not empty in the last {} s",
1159 tstd::TB_SYS_EMPTY_INTERVAL_SECS
1160 ),
1161 );
1162 }
1163
1164 // Indicator 3.10: TBsys data delay > 1 s?
1165 if let Some(delay) = self.tstd.tb_sys.delay_secs(t) {
1166 if delay > tstd::DATA_DELAY_LIMIT_SECS as f64 {
1167 self.emit(
1168 Indicator::DataDelayError,
1169 Some(pid),
1170 t,
1171 format!(
1172 "TBsys data delay {delay:.2} s exceeds {} s on PID 0x{pid:04X}",
1173 tstd::DATA_DELAY_LIMIT_SECS,
1174 ),
1175 );
1176 }
1177 }
1178 }
1179
1180 /// CRC-check for SI PIDs that are not PAT/PMT (handled via si_reassemblies).
1181 fn check_crc_for_si(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
1182 self.check_crc_for_section(section_bytes, pid, t);
1183 }
1184
1185 /// TR 101 290 v1.4.1 Table 5.0b indicator 2.6 — CAT_error, condition 1:
1186 /// section with `table_id != 0x01` on PID_CAT.
1187 fn check_cat_table_id(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
1188 if pid != PID_CAT {
1189 return;
1190 }
1191 let section = match Section::parse(section_bytes) {
1192 Ok(s) => s,
1193 Err(_) => return,
1194 };
1195 if section.table_id == CAT_TABLE_ID {
1196 // Valid CAT section — mark as seen.
1197 self.cat_seen = true;
1198 // Re-arm the "scrambled without CAT" check so that if a CAT was
1199 // previously absent and then appears, the check resets.
1200 self.scrambled_without_cat_reported = false;
1201 } else {
1202 self.emit(
1203 Indicator::CatError,
1204 Some(PID_CAT),
1205 t,
1206 format!(
1207 "section with table_id 0x{:02X} on PID 0x0001 (expected 0x01 for CAT)",
1208 section.table_id
1209 ),
1210 );
1211 }
1212 }
1213
1214 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.1 — NIT_error, bad-table_id
1215 /// dimension. `docs/tr_101_290.md` clause 3.1: allowed table_ids on
1216 /// PID 0x0010 are NIT_actual (0x40), NIT_other (0x41), stuffing/ST
1217 /// (0x72). The absence dimension (NIT_actual missing for the
1218 /// EN 300 468 §5.1.4 repetition interval) is raised from the shared 3.2
1219 /// SI timer in `check_presence_timeouts`.
1220 fn check_nit_table_id(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
1221 if pid != PID_NIT {
1222 return;
1223 }
1224 let section = match Section::parse(section_bytes) {
1225 Ok(s) => s,
1226 Err(_) => return,
1227 };
1228 let allowed = section.table_id == NIT_ACTUAL_TABLE_ID
1229 || section.table_id == NIT_OTHER_TABLE_ID
1230 || section.table_id == STUFFING_TABLE_ID;
1231 if !allowed {
1232 self.emit(
1233 Indicator::NitError,
1234 Some(PID_NIT),
1235 t,
1236 format!(
1237 "section with table_id 0x{:02X} on PID 0x0010 (expected NIT_actual/NIT_other/ST)",
1238 section.table_id
1239 ),
1240 );
1241 }
1242 }
1243
1244 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.5 — SDT_error, bad-table_id
1245 /// dimension. `docs/tr_101_290.md` clause 3.5: allowed table_ids on
1246 /// PID 0x0011 are SDT_actual (0x42), SDT_other (0x46), BAT (0x4A),
1247 /// stuffing/ST (0x72). The absence dimension (SDT_actual missing) is
1248 /// raised from the shared 3.2 SI timer in `check_presence_timeouts`.
1249 fn check_sdt_table_id(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
1250 if pid != PID_SDT_BAT {
1251 return;
1252 }
1253 let section = match Section::parse(section_bytes) {
1254 Ok(s) => s,
1255 Err(_) => return,
1256 };
1257 let allowed = section.table_id == SDT_ACTUAL_TABLE_ID
1258 || section.table_id == SDT_OTHER_TABLE_ID
1259 || section.table_id == BAT_TABLE_ID
1260 || section.table_id == STUFFING_TABLE_ID;
1261 if !allowed {
1262 self.emit(
1263 Indicator::SdtError,
1264 Some(PID_SDT_BAT),
1265 t,
1266 format!(
1267 "section with table_id 0x{:02X} on PID 0x0011 (expected SDT_actual/SDT_other/BAT/ST)",
1268 section.table_id
1269 ),
1270 );
1271 }
1272 }
1273
1274 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.6 — EIT_error, bad-table_id
1275 /// dimension. `docs/tr_101_290.md` clause 3.6: allowed table_ids on
1276 /// PID 0x0012 are EIT P/F actual (0x4E), EIT P/F other (0x4F), EIT
1277 /// schedule actual (0x50..=0x5F), EIT schedule other (0x60..=0x6F),
1278 /// stuffing/ST (0x72). The absence dimension (EIT P/F actual missing) is
1279 /// raised from the shared 3.2 SI timer in `check_presence_timeouts`.
1280 fn check_eit_table_id(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
1281 if pid != PID_EIT {
1282 return;
1283 }
1284 let section = match Section::parse(section_bytes) {
1285 Ok(s) => s,
1286 Err(_) => return,
1287 };
1288 let table_id = section.table_id;
1289 let allowed = table_id == EIT_PF_ACTUAL_TABLE_ID
1290 || table_id == EIT_PF_OTHER_TABLE_ID
1291 || (dvb_si::tables::eit::TABLE_ID_SCHEDULE_ACTUAL_FIRST
1292 ..=dvb_si::tables::eit::TABLE_ID_SCHEDULE_ACTUAL_LAST)
1293 .contains(&table_id)
1294 || (dvb_si::tables::eit::TABLE_ID_SCHEDULE_OTHER_FIRST
1295 ..=dvb_si::tables::eit::TABLE_ID_SCHEDULE_OTHER_LAST)
1296 .contains(&table_id)
1297 || table_id == STUFFING_TABLE_ID;
1298 if !allowed {
1299 self.emit(
1300 Indicator::EitError,
1301 Some(PID_EIT),
1302 t,
1303 format!(
1304 "section with table_id 0x{table_id:02X} on PID 0x0012 (expected EIT P/F or schedule range or ST)"
1305 ),
1306 );
1307 }
1308 }
1309
1310 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.7 — RST_error, bad-table_id
1311 /// dimension. `docs/tr_101_290.md` clause 3.7: allowed table_ids on
1312 /// PID 0x0013 are RST (0x71), stuffing/ST (0x72). RST has no documented
1313 /// absence threshold in Table 5.0c, so there is no presence dimension.
1314 fn check_rst_table_id(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
1315 if pid != PID_RST {
1316 return;
1317 }
1318 let section = match Section::parse(section_bytes) {
1319 Ok(s) => s,
1320 Err(_) => return,
1321 };
1322 let allowed = section.table_id == RST_TABLE_ID || section.table_id == STUFFING_TABLE_ID;
1323 if !allowed {
1324 self.emit(
1325 Indicator::RstError,
1326 Some(PID_RST),
1327 t,
1328 format!(
1329 "section with table_id 0x{:02X} on PID 0x0013 (expected RST/ST)",
1330 section.table_id
1331 ),
1332 );
1333 }
1334 }
1335
1336 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.8 — TDT_error, bad-table_id
1337 /// dimension. `docs/tr_101_290.md` clause 3.8: allowed table_ids on
1338 /// PID 0x0014 are TDT (0x70), stuffing/ST (0x72), TOT (0x73). The
1339 /// absence dimension (TDT missing) is raised from the shared 3.2 SI
1340 /// timer in `check_presence_timeouts`.
1341 fn check_tdt_table_id(&mut self, section_bytes: &[u8], pid: u16, t: Duration) {
1342 if pid != PID_TDT_TOT {
1343 return;
1344 }
1345 let section = match Section::parse(section_bytes) {
1346 Ok(s) => s,
1347 Err(_) => return,
1348 };
1349 let allowed = section.table_id == TDT_TABLE_ID
1350 || section.table_id == TOT_TABLE_ID
1351 || section.table_id == STUFFING_TABLE_ID;
1352 if !allowed {
1353 self.emit(
1354 Indicator::TdtError,
1355 Some(PID_TDT_TOT),
1356 t,
1357 format!(
1358 "section with table_id 0x{:02X} on PID 0x0014 (expected TDT/TOT/ST)",
1359 section.table_id
1360 ),
1361 );
1362 }
1363 }
1364
1365 /// Whether `pid` is (currently) part of the TR 101 290 v1.4.1 Table 5.0c
1366 /// indicator 3.4 referenced set: PAT, CAT, well-known SI PIDs (NIT/
1367 /// SDT-BAT/EIT/RST/TDT-TOT), the null PID, the reserved-for-future-use
1368 /// range, PMT_PIDs referenced by the PAT, and ES/PCR PIDs referenced by
1369 /// a PMT.
1370 ///
1371 /// Not tracked: CAT-referenced EMM PIDs (this monitor does not decode CA
1372 /// descriptors) and PIDs "user defined as private data streams" (the
1373 /// spec's own carve-out — not distinguishable from the wire alone). Both
1374 /// are documented limitations; see the crate `//!` and README.
1375 fn is_referenced_or_reserved_pid(&self, pid: u16) -> bool {
1376 pid == PID_PAT
1377 || pid == PID_CAT
1378 || pid == PID_NIT
1379 || pid == PID_SDT_BAT
1380 || pid == PID_EIT
1381 || pid == PID_RST
1382 || pid == PID_TDT_TOT
1383 || pid == PID_NULL
1384 || (RESERVED_PID_MIN..=RESERVED_PID_MAX).contains(&pid)
1385 || self.pmt_trackings.contains_key(&pid)
1386 || self.es_trackings.contains_key(&pid)
1387 }
1388
1389 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.4 — Unreferenced_PID
1390 /// bookkeeping. Records the first-seen time of a PID that is not
1391 /// (currently) part of the referenced set. If the PID is already
1392 /// referenced, drops any stale tracking entry. Absence-timeout
1393 /// evaluation happens in `check_presence_timeouts`; a PID that later
1394 /// becomes referenced (see `process_pat_section` / `process_pmt_section`)
1395 /// has its entry removed there.
1396 fn track_unreferenced_pid(&mut self, pid: u16, t: Duration) {
1397 if self.is_referenced_or_reserved_pid(pid) {
1398 self.unreferenced_pid_timers.remove(&pid);
1399 return;
1400 }
1401 self.unreferenced_pid_timers
1402 .entry(pid)
1403 .or_insert_with(|| UnreferencedPidTracking {
1404 first_seen: t,
1405 reported: false,
1406 });
1407 }
1408
1409 /// Process a completed section on PID_PAT.
1410 fn process_pat_section(&mut self, section_bytes: &[u8], t: Duration) {
1411 let section = match Section::parse(section_bytes) {
1412 Ok(s) => s,
1413 Err(_) => return,
1414 };
1415
1416 // 1.3.a: section with table_id other than 0x00 found on PID 0x0000.
1417 if section.table_id != PAT_TABLE_ID {
1418 self.emit(
1419 Indicator::PatError2,
1420 Some(PID_PAT),
1421 t,
1422 format!(
1423 "section with table_id 0x{:02X} on PID 0x0000 (expected 0x00)",
1424 section.table_id
1425 ),
1426 );
1427 return;
1428 }
1429
1430 // Parse the PAT proper.
1431 let pat = match PatSection::parse(section_bytes) {
1432 Ok(p) => p,
1433 Err(_) => return,
1434 };
1435
1436 // Discover program_map_PIDs and start tracking them.
1437 for entry in pat.programmes() {
1438 let pmt_pid = entry.pid;
1439 self.pmt_trackings
1440 .entry(pmt_pid)
1441 .or_insert_with(|| PmtTracking {
1442 timer: PresenceTimer {
1443 last_seen: t,
1444 reported: false,
1445 },
1446 reassembler: SectionReassembler::default(),
1447 });
1448 // 3.4: a program_map_PID is now referenced — it is no longer an
1449 // Unreferenced_PID candidate even if its own packets have not
1450 // been observed yet.
1451 self.unreferenced_pid_timers.remove(&pmt_pid);
1452 }
1453 }
1454
1455 /// Process a completed section on a program_map_PID.
1456 fn process_pmt_section(&mut self, section_bytes: &[u8], _pid: u16, t: Duration) {
1457 let section = match Section::parse(section_bytes) {
1458 Ok(s) => s,
1459 Err(_) => return,
1460 };
1461
1462 // 1.5.a only checks presence and scrambling of table_id 0x02 sections.
1463 // If table_id is not 0x02, skip — we don't emit PMT_error_2 for a
1464 // wrong table_id on a program_map_PID (that's not in the spec for
1465 // 1.5.a).
1466 let pmt_table_id: u8 = dvb_si::tables::pmt::TABLE_ID;
1467 if section.table_id != pmt_table_id {
1468 return;
1469 }
1470
1471 // Parse the PMT proper.
1472 let pmt = match PmtSection::parse(section_bytes) {
1473 Ok(p) => p,
1474 Err(_) => return,
1475 };
1476
1477 // Collect new ES PIDs to add.
1478 let mut new_es_pids: Vec<u16> = Vec::new();
1479 if pmt.pcr_pid != PID_NULL && !self.es_trackings.contains_key(&pmt.pcr_pid) {
1480 new_es_pids.push(pmt.pcr_pid);
1481 }
1482 for stream in &pmt.streams {
1483 let es_pid = stream.elementary_pid;
1484 if !self.es_trackings.contains_key(&es_pid) {
1485 new_es_pids.push(es_pid);
1486 }
1487 }
1488
1489 for es_pid in new_es_pids {
1490 self.es_trackings.insert(
1491 es_pid,
1492 EsTracking {
1493 timer: PresenceTimer {
1494 last_seen: t,
1495 reported: false,
1496 },
1497 },
1498 );
1499 // 3.4: an ES/PCR PID is now referenced by a PMT — no longer an
1500 // Unreferenced_PID candidate.
1501 self.unreferenced_pid_timers.remove(&es_pid);
1502 }
1503 }
1504
1505 /// TR 101 290 v1.4.1 Table 5.0b indicators 2.3a / 2.3b — PCR checks.
1506 fn check_pcr(&mut self, pid: u16, pcr_27mhz: u64, discontinuity: bool, t: Duration) {
1507 let state = self.pcr_states.entry(pid).or_insert_with(|| PcrState {
1508 last_pcr_27mhz: 0,
1509 last_pcr_time: Duration::ZERO,
1510 initialised: false,
1511 });
1512
1513 if !state.initialised {
1514 state.last_pcr_27mhz = pcr_27mhz;
1515 state.last_pcr_time = t;
1516 state.initialised = true;
1517 return;
1518 }
1519
1520 // Snapshot state for decision-making before any emit.
1521 let last_pcr_time = state.last_pcr_time;
1522 let last_pcr_27mhz = state.last_pcr_27mhz;
1523
1524 // 2.3a: PCR_repetition_error — interval between consecutive PCR
1525 // values exceeds the configured limit.
1526 let rep_interval = t.saturating_sub(last_pcr_time);
1527 let should_emit_rep = rep_interval > self.config.pcr_repetition_limit;
1528
1529 // 2.3b: PCR_discontinuity_indicator_error — PCR delta exceeds 100 ms
1530 // without a signalled discontinuity.
1531 let delta =
1532 (pcr_27mhz.wrapping_add(PCR_MODULUS_27MHZ) - last_pcr_27mhz) % PCR_MODULUS_27MHZ;
1533 let delta_ms = delta * 1000 / CLOCK_27MHZ;
1534 let limit_ms = self.config.pcr_discontinuity_limit.as_millis() as u64;
1535 let should_emit_disc = delta_ms > limit_ms && !discontinuity;
1536
1537 // Emit outside the HashMap borrow.
1538 if should_emit_rep {
1539 self.emit(
1540 Indicator::PcrRepetitionError,
1541 Some(pid),
1542 t,
1543 format!(
1544 "PCR interval {} ms exceeds limit {} ms on PID 0x{:04X}",
1545 rep_interval.as_millis(),
1546 self.config.pcr_repetition_limit.as_millis(),
1547 pid
1548 ),
1549 );
1550 }
1551 if should_emit_disc {
1552 self.emit(
1553 Indicator::PcrDiscontinuityError,
1554 Some(pid),
1555 t,
1556 format!(
1557 "PCR delta {delta_ms} ms exceeds limit {limit_ms} ms on PID 0x{pid:04X} without discontinuity_indicator"
1558 ),
1559 );
1560 }
1561
1562 // Update state.
1563 let state = self.pcr_states.get_mut(&pid).unwrap();
1564 state.last_pcr_27mhz = pcr_27mhz;
1565 state.last_pcr_time = t;
1566 }
1567
1568 /// TR 101 290 v1.4.1 Table 5.0b indicator 2.5 — PTS_error.
1569 ///
1570 /// Peeks the PES header on an elementary-stream PID for PTS_DTS_flags.
1571 /// Only checks PIDs that have been "armed" by seeing at least one PTS.
1572 fn check_pts(&mut self, pid: u16, payload: &[u8], t: Duration) {
1573 // PES start-code prefix: 00 00 01.
1574 if payload.len() < PES_FLAGS_OFFSET + 2 {
1575 return;
1576 }
1577 if payload[0] != PES_PREFIX_0 || payload[1] != PES_PREFIX_1 || payload[2] != PES_PREFIX_2 {
1578 return;
1579 }
1580
1581 // Byte 6: `'10' + flags` — the top two bits must be `10`.
1582 let flags_byte = payload[PES_FLAGS_OFFSET];
1583 if (flags_byte >> 6) != 0b10 {
1584 return;
1585 }
1586
1587 // Byte 7: PTS_DTS_flags in bits `[7:6]`.
1588 let pts_dts_flags = payload[PES_FLAGS_OFFSET + 1] & PES_PTS_DTS_FLAGS_MASK;
1589 let pts_present = (pts_dts_flags & PES_PTS_PRESENT) != 0;
1590 if !pts_present {
1591 return;
1592 }
1593
1594 let state = self.pts_states.entry(pid).or_insert_with(|| PtsState {
1595 last_pts_time: Duration::ZERO,
1596 armed: false,
1597 });
1598
1599 if !state.armed {
1600 // First PTS on this PID — arm the check, no error yet.
1601 state.last_pts_time = t;
1602 state.armed = true;
1603 return;
1604 }
1605
1606 // Snapshot state for decision-making before any emit.
1607 let last_pts_time = state.last_pts_time;
1608 let pts_interval = t.saturating_sub(last_pts_time);
1609 let should_emit = pts_interval > self.config.pts_repetition_limit;
1610
1611 if should_emit {
1612 self.emit(
1613 Indicator::PtsError,
1614 Some(pid),
1615 t,
1616 format!(
1617 "PTS interval {} ms exceeds limit {} ms on PID 0x{:04X}",
1618 pts_interval.as_millis(),
1619 self.config.pts_repetition_limit.as_millis(),
1620 pid
1621 ),
1622 );
1623 }
1624
1625 // Update state.
1626 let state = self.pts_states.get_mut(&pid).unwrap();
1627 state.last_pts_time = t;
1628 }
1629
1630 /// TR 101 290 v1.4.1 Table 5.0c indicator 3.2 — update SI repetition timer
1631 /// when a completed section on a well-known SI PID matches one of the four
1632 /// tracked table_ids.
1633 fn update_si_repetition(&mut self, section_bytes: &[u8], _pid: u16, t: Duration) {
1634 let table_id = match Section::parse(section_bytes) {
1635 Ok(s) => s.table_id,
1636 Err(_) => return,
1637 };
1638
1639 let is_tracked = table_id == NIT_ACTUAL_TABLE_ID
1640 || table_id == SDT_ACTUAL_TABLE_ID
1641 || table_id == EIT_PF_ACTUAL_TABLE_ID
1642 || table_id == TDT_TABLE_ID;
1643
1644 if !is_tracked {
1645 return;
1646 }
1647
1648 let timer = self
1649 .si_timers
1650 .entry(table_id)
1651 .or_insert_with(|| SiRepetitionTimer {
1652 last_seen: Duration::ZERO,
1653 reported: false,
1654 armed: false,
1655 });
1656
1657 timer.last_seen = t;
1658 timer.reported = false;
1659 timer.armed = true;
1660 }
1661
1662 /// T-STD buffer model processing: run the buffer simulation for one TS
1663 /// packet and evaluate indicators 3.3 (Buffer_error), 3.9
1664 /// (Empty_buffer_error), and 3.10 (Data_delay_error).
1665 ///
1666 /// # Buffer model
1667 ///
1668 /// - **TBn** (per-PID, 512 bytes): each payload PID gets a transport
1669 /// buffer. The full 188-byte packet (minus sync) enters the buffer
1670 /// instantaneously at the caller's timestamp `t`. The buffer drains at
1671 /// the stream's effective bitrate (estimated from accumulated bytes
1672 /// divided by elapsed wall-clock time, clamped to a floor of
1673 /// 125 kbit/s).
1674 /// - **TBsys** (global, 512 bytes, 1 Mbit/s leak): the PSI/SI PIDs (PAT,
1675 /// CAT, NIT, SDT/BAT, EIT, RST, TDT/TOT) feed into TBsys. When a
1676 /// completed section is assembled and passed to the SI/PSI decoder, the
1677 /// section bytes are removed from TBsys.
1678 ///
1679 /// # Known limitation
1680 ///
1681 /// The effective bitrate estimate from accumulated bytes / wall-clock time
1682 /// is inherently approximate when the caller's timestamps do not match the
1683 /// actual transport rate. The monitor documents this: the caller is
1684 /// responsible for supplying realistic timestamps.
1685 fn process_tstd(&mut self, header: &mpeg_ts::ts::TsHeader, ts_packet: &[u8], t: Duration) {
1686 let pid = header.pid;
1687
1688 // Periodic TBsys drain and empty-interval check. TBsys feeds at
1689 // section-completion, but we must drain it on every packet to
1690 // accurately model the 1 Mbit/s continuous drain rate.
1691 self.tstd.drain_tb_sys(t);
1692
1693 // ── TBsys: PSI/SI PIDs ──────────────────────────────────────────
1694 let is_si_pid = pid == PID_PAT
1695 || pid == PID_CAT
1696 || pid == PID_NIT
1697 || pid == PID_SDT_BAT
1698 || pid == PID_EIT
1699 || pid == PID_RST
1700 || pid == PID_TDT_TOT;
1701
1702 // Count the packet bytes for T-STD accounting.
1703 // The T-STD model uses the full 188-byte TS packet as the arrival
1704 // unit (ISO/IEC 13818-1 §2.4.2.3: the transport buffer receives TS
1705 // packets and strips the TS header + adaptation field before passing
1706 // bytes downstream; but for TBn overflow, the relevant input unit
1707 // is the packet's contribution to buffer occupancy).
1708 let packet_bytes = ts_packet.len() as u64;
1709
1710 if is_si_pid && header.has_payload {
1711 // TBsys is fed at section-completion time (see
1712 // `feed_tbsys_section_bytes` called from `check_crc_for_section`).
1713 // The per-packet path only drains TBsys and does empty/delay
1714 // checks, which are also done at section-completion.
1715 }
1716
1717 // ── TBn: per-PID buffers for all non-SI, non-null PIDs ──────────
1718 if pid == PID_NULL || is_si_pid {
1719 return;
1720 }
1721
1722 // Compute T-STD checks without holding a mutable borrow on self.tstd
1723 // across any self.emit() calls. We collect the events to emit, then
1724 // emit them after releasing the borrow.
1725 struct TstdCheck {
1726 buffer_overflow: bool,
1727 empty_interval: bool,
1728 delay_exceeded: bool,
1729 }
1730
1731 let check = {
1732 let entry = self.tstd.pid_buffers.entry(pid).or_insert_with(|| {
1733 // Initial leak rate conservatively high: the effective
1734 // bitrate will be estimated once we have timing data.
1735 tstd::PidStdState::new(1_000_000u64, t)
1736 });
1737
1738 // Accumulate bytes for effective bitrate estimation.
1739 entry.total_bytes += packet_bytes;
1740
1741 // Estimate the effective transport rate from accumulated bytes
1742 // and elapsed wall-clock time. Wait until we have at least
1743 // 1 ms of history before estimating; use a default of
1744 // ~6 Mbit/s (750 KB/s) until then.
1745 let leak = {
1746 let elapsed_us = t.saturating_sub(entry.first_seen).as_micros() as u64;
1747 if elapsed_us >= 1_000 {
1748 // bytes per second = total_bytes * 1_000_000 / elapsed_us
1749 (entry.total_bytes * 1_000_000 / elapsed_us).max(tstd::TB_LEAK_RATE_FLOOR)
1750 } else {
1751 // Default: generous initial rate (~40 Mbit/s).
1752 // This prevents false overflows during the first
1753 // millisecond before we have enough data to estimate
1754 // the effective bitrate. For a real ~38 Mbit/s stream
1755 // with 40 µs inter-packet, this is close enough to
1756 // prevent overflow during the convergence period.
1757 5_000_000u64
1758 }
1759 };
1760 entry.tb.set_leak_rate(leak);
1761
1762 // Drain the buffer TO the current time (i.e. drain up to the
1763 // arrival instant of this packet, before the packet data enters).
1764 entry.tb.drain_to(t);
1765
1766 let buffer_overflow = {
1767 // Drain, then feed (for occupancy tracking). We do NOT flag
1768 // TBn overflow as BufferError — accurate overflow detection
1769 // requires the coded bitrate Rxn from descriptors
1770 // (multiplex_buffer_descriptor). Without that, the
1771 // estimated leak rate is approximate and produces false
1772 // positives on clean streams. TBn occupancy is still
1773 // tracked for empty-interval and data-delay checks.
1774 entry.tb.drain_to(t);
1775 let _overflow = entry.tb.feed(packet_bytes, t);
1776 false
1777 };
1778 let empty_interval = entry.tb.check_empty_interval(t) && !entry.empty_reported;
1779 let delay_exceeded = if let Some(delay) = entry.tb.delay_secs(t) {
1780 delay > tstd::DATA_DELAY_LIMIT_SECS as f64 && !entry.delay_reported
1781 } else {
1782 false
1783 };
1784
1785 if empty_interval {
1786 entry.empty_reported = true;
1787 }
1788 if delay_exceeded {
1789 entry.delay_reported = true;
1790 }
1791 entry.last_packet_time = t;
1792
1793 TstdCheck {
1794 buffer_overflow,
1795 empty_interval,
1796 delay_exceeded,
1797 }
1798 };
1799
1800 // Emit events outside the borrow.
1801 if check.buffer_overflow {
1802 self.emit(
1803 Indicator::BufferError,
1804 Some(pid),
1805 t,
1806 format!(
1807 "TBn overflow on PID 0x{pid:04X}: {} byte capacity exceeded",
1808 tstd::TB_SIZE,
1809 ),
1810 );
1811 }
1812
1813 if check.empty_interval {
1814 self.emit(
1815 Indicator::EmptyBufferError,
1816 Some(pid),
1817 t,
1818 format!(
1819 "TBn not empty in the last {} s on PID 0x{pid:04X}",
1820 tstd::TB_EMPTY_INTERVAL_SECS,
1821 ),
1822 );
1823 }
1824
1825 if check.delay_exceeded {
1826 self.emit(
1827 Indicator::DataDelayError,
1828 Some(pid),
1829 t,
1830 format!(
1831 "TBn data delay exceeds {} s on PID 0x{pid:04X}",
1832 tstd::DATA_DELAY_LIMIT_SECS,
1833 ),
1834 );
1835 }
1836 }
1837
1838 /// Evaluate all presence/absence timeouts against the current time `t`.
1839 fn check_presence_timeouts(&mut self, t: Duration) {
1840 // 1.3.a: PAT presence timeout
1841 if t.saturating_sub(self.pat_timer.last_seen) > self.config.pat_max_interval
1842 && !self.pat_timer.reported
1843 {
1844 self.pat_timer.reported = true;
1845 self.emit(
1846 Indicator::PatError2,
1847 Some(PID_PAT),
1848 t,
1849 format!(
1850 "no PAT section within {} ms",
1851 self.config.pat_max_interval.as_millis()
1852 ),
1853 );
1854 }
1855
1856 // 1.5.a: PMT presence timeout per program_map_PID
1857 // Collect PIDs that need events, then emit outside the iteration.
1858 let pmt_timeouts: Vec<(u16, u64)> = self
1859 .pmt_trackings
1860 .iter()
1861 .filter_map(|(&pid, tracking)| {
1862 if t.saturating_sub(tracking.timer.last_seen) > self.config.pmt_max_interval
1863 && !tracking.timer.reported
1864 {
1865 Some((pid, self.config.pmt_max_interval.as_millis() as u64))
1866 } else {
1867 None
1868 }
1869 })
1870 .collect();
1871 for (pid, interval_ms) in pmt_timeouts {
1872 if let Some(tracking) = self.pmt_trackings.get_mut(&pid) {
1873 tracking.timer.reported = true;
1874 }
1875 self.emit(
1876 Indicator::PmtError2,
1877 Some(pid),
1878 t,
1879 format!("no PMT section on program_map_PID 0x{pid:04X} within {interval_ms} ms"),
1880 );
1881 }
1882
1883 // 1.6: PID_error — referenced PID absence
1884 let pid_timeouts: Vec<(u16, u64)> = self
1885 .es_trackings
1886 .iter()
1887 .filter_map(|(&pid, tracking)| {
1888 if t.saturating_sub(tracking.timer.last_seen) > self.config.pid_error_period
1889 && !tracking.timer.reported
1890 {
1891 Some((pid, self.config.pid_error_period.as_secs()))
1892 } else {
1893 None
1894 }
1895 })
1896 .collect();
1897 for (pid, period_secs) in pid_timeouts {
1898 if let Some(tracking) = self.es_trackings.get_mut(&pid) {
1899 tracking.timer.reported = true;
1900 }
1901 self.emit(
1902 Indicator::PidError,
1903 Some(pid),
1904 t,
1905 format!("referenced PID 0x{pid:04X} absent for > {period_secs} s"),
1906 );
1907 }
1908
1909 // 3.2: SI_repetition_error — maximum interval for tracked SI tables.
1910 // Collect table_ids that need events, then emit outside the iteration.
1911 let si_timeouts: Vec<(u8, u64, u16, u64)> = self
1912 .si_timers
1913 .iter()
1914 .filter_map(|(&table_id, timer)| {
1915 if !timer.armed || timer.reported {
1916 return None;
1917 }
1918 let (limit, pid) = match table_id {
1919 NIT_ACTUAL_TABLE_ID => (self.config.si_nit_interval, PID_NIT),
1920 SDT_ACTUAL_TABLE_ID => (self.config.si_sdt_interval, PID_SDT_BAT),
1921 EIT_PF_ACTUAL_TABLE_ID => (self.config.si_eit_pf_interval, PID_EIT),
1922 TDT_TABLE_ID => (self.config.si_tdt_interval, PID_TDT_TOT),
1923 _ => return None,
1924 };
1925 let interval = t.saturating_sub(timer.last_seen);
1926 if interval > limit {
1927 Some((
1928 table_id,
1929 interval.as_millis() as u64,
1930 pid,
1931 limit.as_millis() as u64,
1932 ))
1933 } else {
1934 None
1935 }
1936 })
1937 .collect();
1938 for (table_id, interval_ms, pid, limit_ms) in si_timeouts {
1939 if let Some(timer) = self.si_timers.get_mut(&table_id) {
1940 timer.reported = true;
1941 }
1942 // The four `_actual` table_ids share this ONE lazily-armed
1943 // timer: an absence past the interval is simultaneously
1944 // SI_repetition_error (3.2, the general repetition-rate
1945 // indicator, emitted unconditionally below) AND the presence
1946 // dimension of the corresponding table-specific indicator
1947 // (NIT_error 3.1 / SDT_error 3.5 / EIT_error 3.6 / TDT_error 3.8)
1948 // — the same underlying absence, per the spec's historical
1949 // combined indicator vs. the split `_actual` variants (see
1950 // docs/tr_101_290.md notes 2-4).
1951 let (table_name, group_indicator) = match table_id {
1952 NIT_ACTUAL_TABLE_ID => ("NIT_actual", Some(Indicator::NitError)),
1953 SDT_ACTUAL_TABLE_ID => ("SDT_actual", Some(Indicator::SdtError)),
1954 EIT_PF_ACTUAL_TABLE_ID => ("EIT_P/F_actual", Some(Indicator::EitError)),
1955 TDT_TABLE_ID => ("TDT", Some(Indicator::TdtError)),
1956 _ => ("unknown", None),
1957 };
1958 self.emit(
1959 Indicator::SiRepetitionError,
1960 Some(pid),
1961 t,
1962 format!("{table_name} repetition interval {interval_ms} ms exceeds {limit_ms} ms"),
1963 );
1964 if let Some(indicator) = group_indicator {
1965 self.emit(
1966 indicator,
1967 Some(pid),
1968 t,
1969 format!("no {table_name} section on PID 0x{pid:04X} within {limit_ms} ms"),
1970 );
1971 }
1972 }
1973
1974 // 3.4: Unreferenced_PID — a PID persisting beyond the presence
1975 // threshold without being referenced by the PAT/CAT/a PMT or one of
1976 // the well-known SI PIDs.
1977 let unref_timeouts: Vec<(u16, u64)> = self
1978 .unreferenced_pid_timers
1979 .iter()
1980 .filter_map(|(&pid, timer)| {
1981 if timer.reported {
1982 return None;
1983 }
1984 let elapsed = t.saturating_sub(timer.first_seen);
1985 if elapsed > self.config.unreferenced_pid_period {
1986 Some((pid, self.config.unreferenced_pid_period.as_millis() as u64))
1987 } else {
1988 None
1989 }
1990 })
1991 .collect();
1992 for (pid, period_ms) in unref_timeouts {
1993 if let Some(timer) = self.unreferenced_pid_timers.get_mut(&pid) {
1994 timer.reported = true;
1995 }
1996 self.emit(
1997 Indicator::UnreferencedPid,
1998 Some(pid),
1999 t,
2000 format!(
2001 "PID 0x{pid:04X} present for > {period_ms} ms without being referenced by PAT/CAT/a PMT or a well-known SI PID"
2002 ),
2003 );
2004 }
2005 }
2006}
2007
2008impl Default for ConformanceMonitor {
2009 fn default() -> Self {
2010 Self::new()
2011 }
2012}
2013
2014#[cfg(test)]
2015mod tests;