From f9271c047a3bd7f3e106af39b9f7cae27ab9c0e2 Mon Sep 17 00:00:00 2001 From: Simon Date: Sat, 11 Jul 2026 11:11:43 +0000 Subject: [PATCH] Add passive DMR data observability --- FreeDMR.cfg | 3 + bridge_master.py | 63 ++++++- config.py | 1 + docker-configs/freedmr.cfg | 3 + docs/dmr-data-observability.md | 100 +++++++++++ docs/testing.md | 12 +- docs/v1x-codex-changelog.md | 16 ++ freedmr.cfg | 3 + freedmr_dmr_codec.py | 265 ++++++++++++++++++++++++++++ tests/harness/deterministic.py | 1 + tests/harness/udp_blackbox.py | 5 + tests/test_deterministic_harness.py | 200 ++++++++++++++++++++- tests/test_freedmr_dmr_codec.py | 119 +++++++++++++ tests/test_udp_blackbox_harness.py | 54 +++++- 14 files changed, 834 insertions(+), 11 deletions(-) create mode 100644 docs/dmr-data-observability.md diff --git a/FreeDMR.cfg b/FreeDMR.cfg index 67c5c90..1c23673 100644 --- a/FreeDMR.cfg +++ b/FreeDMR.cfg @@ -7,6 +7,9 @@ [GLOBAL] # If you join the FreeDMR network, add your server ID here. SERVER_ID: 0 +# Passive DMR data logging: decoded content at INFO, raw bursts at DEBUG. +# May expose private message or location content. +DATA_PACKET_LOGGING: False [REPORTS] diff --git a/bridge_master.py b/bridge_master.py index 2f507fe..47ea7bb 100644 --- a/bridge_master.py +++ b/bridge_master.py @@ -147,6 +147,46 @@ def is_group_data_control(_call_type, _dtype_vseq): def group_data_report_name(_call_type, _dtype_vseq): return group_data_event_name(_call_type, _dtype_vseq) or 'OTHER DATA' +def log_dmr_data_observation(owner, system, transport, peer_id, rf_src, dst_id, seq, slot, call_type, frame_type, dtype_vseq, stream_id, payload): + if not CONFIG['GLOBAL'].get('DATA_PACKET_LOGGING', False): + return + if frame_type != HBPF_DATA_SYNC or dtype_vseq not in dmr_codec.DMR_DATA_BURST_TYPES: + return + + try: + observation = dmr_codec.inspect_data_burst(payload, dtype_vseq) + except dmr_codec.DmrDataError as error: + logger.debug('(%s) *DMR DATA OBSERVE ERROR* TRANSPORT: %s STREAM ID: %s SEQ: %s TS: %s CALL: %s SRC: %s DST: %s PEER: %s ERROR: %s', + system, transport, int_id(stream_id), seq, slot, call_type, int_id(rf_src), int_id(dst_id), int_id(peer_id), error) + return + + if logger.isEnabledFor(logging.DEBUG): + logical = observation.logical_pdu.hex() if observation.logical_pdu is not None else '-' + dpf = '{}({})'.format(observation.data_packet_format, observation.data_packet_format_name) if observation.data_packet_format is not None else '-' + csbko = observation.csbko if observation.csbko is not None else '-' + fid = observation.fid if observation.fid is not None else '-' + logger.debug('(%s) *DMR DATA OBSERVE* TRANSPORT: %s STREAM ID: %s SEQ: %s TS: %s CALL: %s SRC: %s DST: %s PEER: %s TYPE: %s(%s) AIR_TYPE: %s(%s) MATCH: %s CC: %s CODING: %s CRC: %s SYNC: %s DPF: %s CSBKO: %s FID: %s LOGICAL: %s RAW: %s', + system, transport, int_id(stream_id), seq, slot, call_type, int_id(rf_src), int_id(dst_id), int_id(peer_id), + observation.wrapper_data_type, observation.wrapper_name, observation.air_data_type, observation.air_data_type_name, + observation.type_matches, observation.color_code, observation.coding, observation.crc_status, observation.sync.hex(), + dpf, csbko, fid, logical, observation.raw_payload.hex()) + + collector = getattr(owner, '_dmr_data_human_collector', None) + if collector is None: + collector = dmr_codec.DmrDataHumanCollector() + owner._dmr_data_human_collector = collector + key = (transport, system, int_id(stream_id), int_id(rf_src), int_id(dst_id), slot) + human = collector.observe(key, observation) + for text in human.texts: + logger.info('(%s) *DMR DATA TEXT* TRANSPORT: %s STREAM ID: %s TS: %s SRC: %s DST: %s APPLICATION: %s PROFILE: %s ENCODING: %s TEXT: %r', + system, transport, int_id(stream_id), slot, int_id(rf_src), int_id(dst_id), + text.application, text.profile_hint, text.encoding, text.text) + if human.location is not None: + logger.info('(%s) *DMR DATA GPS* TRANSPORT: %s STREAM ID: %s TS: %s SRC: %s DST: %s PROFILE: NMEA-RMC TYPE: %s STATUS: %s UTC: %s LAT: %.6f LON: %.6f SPEED_KNOTS: %s COURSE: %s', + system, transport, int_id(stream_id), slot, int_id(rf_src), int_id(dst_id), + human.location.sentence_type, human.location.status, human.location.utc, + human.location.latitude, human.location.longitude, human.location.speed_knots, human.location.course) + def dmrd_seq_delta(seq, last_seq): if last_seq is False or last_seq is None: return None @@ -385,11 +425,18 @@ def _log_talker_alias(_system, _status, _rf_src, _dst_id, _slot, _stream_id, _lc text = _decode_ta_text(ta_state) data_format = ta_state.get('FORMAT', 0) length = ta_state.get('LENGTH', 0) - logger.info( - '(%s) *IN-CALL TA* STREAM ID: %s SUB: %s TGID: %s TS: %s %s FORMAT: %s LENGTH: %s TEXT: %r LC: %s', + logger.debug( + '(%s) *IN-CALL TA DETAIL* STREAM ID: %s SUB: %s TGID: %s TS: %s %s FORMAT: %s LENGTH: %s TEXT: %r LC: %s', _system, int_id(_stream_id), int_id(_rf_src), int_id(_dst_id), _slot, block_name, TA_FORMAT_NAMES.get(data_format, 'UNKNOWN'), length, text, ahex(_lc).decode() ) + if text and (not length or len(text) >= length) and text != ta_state.get('LAST_LOGGED_TEXT'): + ta_state['LAST_LOGGED_TEXT'] = text + logger.info( + '(%s) *IN-CALL TA* STREAM ID: %s SUB: %s TGID: %s TS: %s FORMAT: %s TEXT: %r', + _system, int_id(_stream_id), int_id(_rf_src), int_id(_dst_id), _slot, + TA_FORMAT_NAMES.get(data_format, 'UNKNOWN'), text + ) def _log_gps_info(_system, _rf_src, _dst_id, _slot, _stream_id, _lc): lc_bits = bitarray(endian='big') @@ -399,10 +446,14 @@ def _log_gps_info(_system, _rf_src, _dst_id, _slot, _stream_id, _lc): latitude_raw = _signed_value(_bits_to_int(lc_bits[48:72]), 24) longitude = longitude_raw * 360.0 / (1 << 25) latitude = latitude_raw * 180.0 / (1 << 24) + logger.debug( + '(%s) *IN-CALL GPS DETAIL* STREAM ID: %s SUB: %s TGID: %s TS: %s LC: %s', + _system, int_id(_stream_id), int_id(_rf_src), int_id(_dst_id), _slot, ahex(_lc).decode() + ) logger.info( - '(%s) *IN-CALL GPS* STREAM ID: %s SUB: %s TGID: %s TS: %s LAT: %.6f LON: %.6f POS_ERR: %s LC: %s', + '(%s) *IN-CALL GPS* STREAM ID: %s SUB: %s TGID: %s TS: %s LAT: %.6f LON: %.6f POS_ERR: %s', _system, int_id(_stream_id), int_id(_rf_src), int_id(_dst_id), _slot, - latitude, longitude, position_error, ahex(_lc).decode() + latitude, longitude, position_error ) def log_embedded_lc_observations( @@ -1965,6 +2016,8 @@ class routerOBP(OPENBRIDGE): dmrpkt = _data[20:53] _bits = _data[15] + log_dmr_data_observation(self, self._system, 'FBP/OBP', _peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, dmrpkt) + #pkt_crc = Crc32.calc(_data[4:53]) #_pkt_crc = Crc32.calc(dmrpkt) @@ -2665,6 +2718,8 @@ class routerHBP(HBSYSTEM): pkt_time = time() dmrpkt = _data[20:53] + + log_dmr_data_observation(self, self._system, 'HBP', _peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, dmrpkt) _ber = _data[53:54] _rssi = _data[54:55] diff --git a/config.py b/config.py index 5a7b942..3fc83f0 100755 --- a/config.py +++ b/config.py @@ -195,6 +195,7 @@ def build_config(_config_file): 'ANNOUNCEMENT_LANGUAGES': config.get(section, 'ANNOUNCEMENT_LANGUAGES', fallback=''), 'SERVER_ID': config.getint(section, 'SERVER_ID', fallback=0).to_bytes(4, 'big'), 'DATA_GATEWAY': config.getboolean(section, 'DATA_GATEWAY', fallback=False), + 'DATA_PACKET_LOGGING': config.getboolean(section, 'DATA_PACKET_LOGGING', fallback=False), 'VALIDATE_SERVER_IDS': config.getboolean(section, 'VALIDATE_SERVER_IDS', fallback=True), 'DEBUG_BRIDGES' : config.getboolean(section, 'DEBUG_BRIDGES', fallback=False), 'ENABLE_API' : config.getboolean(section, 'ENABLE_API', fallback=False), diff --git a/docker-configs/freedmr.cfg b/docker-configs/freedmr.cfg index 67c5c90..1c23673 100644 --- a/docker-configs/freedmr.cfg +++ b/docker-configs/freedmr.cfg @@ -7,6 +7,9 @@ [GLOBAL] # If you join the FreeDMR network, add your server ID here. SERVER_ID: 0 +# Passive DMR data logging: decoded content at INFO, raw bursts at DEBUG. +# May expose private message or location content. +DATA_PACKET_LOGGING: False [REPORTS] diff --git a/docs/dmr-data-observability.md b/docs/dmr-data-observability.md new file mode 100644 index 0000000..7b290a9 --- /dev/null +++ b/docs/dmr-data-observability.md @@ -0,0 +1,100 @@ +# DMR Data Observability + +FreeDMR can emit passive, per-burst metadata for received DMR data/control +traffic. The observer does not change routing, packet bytes, acknowledgement +state or delivery behavior. + +## Enable + +The feature is disabled by default. Enable it in `[GLOBAL]`: + +```ini +DATA_PACKET_LOGGING: True +``` + +Restart FreeDMR after changing the setting. + +The setting applies to both HBP and FBP/OpenBridge receive paths. Voice header, +voice payload and voice terminator bursts never enter the data observer. + +At normal `INFO` log level, FreeDMR emits only decoded human-readable records: + +- `*DMR DATA TEXT*` for conservative ASCII, UTF-16BE or UTF-16LE text; +- `*DMR DATA GPS*` for complete NMEA GPRMC/GNRMC positions; +- existing `*IN-CALL TA*` records for complete Talker Alias text; +- existing `*IN-CALL GPS*` records for decoded embedded-LC positions. + +Set the server log level to `DEBUG` when raw burst and protocol detail is also +required. + +## Logged fields + +At `DEBUG`, each eligible RX burst produces one bounded +`*DMR DATA OBSERVE*` record: + +| Field | Meaning | +| --- | --- | +| `TRANSPORT` | Source receive path: `HBP` or `FBP/OBP`. | +| `STREAM ID`, `SEQ`, `TS`, `CALL` | Wrapper stream and burst metadata. | +| `SRC`, `DST`, `PEER` | Wrapper source, destination and peer IDs. | +| `TYPE` | Data type reported by the HBP/FBP wrapper. | +| `AIR_TYPE`, `CC` | Uncorrected Slot Type data bits carried in the 33-byte DMR burst. | +| `MATCH` | Whether wrapper and air-burst data types agree. | +| `CODING` | Expected coding for the wrapper data type. | +| `SYNC` | Raw 48-bit center sync field. | +| `DPF` | Data Packet Format for a BPTC-decoded Data Header. | +| `CSBKO`, `FID` | Opcode and feature-set ID for a BPTC-decoded CSBK. | +| `LOGICAL` | Generic 12-byte BPTC(196,96) information extraction where applicable. | +| `RAW` | Complete original 33-byte DMR burst payload. | +| `CRC` | Validation status. The first implementation reports `NOT_CHECKED`. | + +Supported DPF labels are UDT, response, unconfirmed, confirmed, defined short +data, raw/status short data and proprietary. Unknown values remain numeric and +are labelled `UNKNOWN`. + +CSBK, MBC header/continuation, Data Header and Rate 1/2 bursts receive generic +BPTC extraction. Rate 3/4 and Rate 1 payloads are kept raw; this observer does +not guess at trellis or rate-1 decoding. Idle and reserved burst types are +deliberately skipped to avoid noise and log-volume overhead. + +## Human-readable reconstruction + +The observer keeps a small per-router least-recently-used collector. It retains +at most 32 data streams and 256 logical bytes per stream. Only decoded Rate 1/2 +logical blocks are accumulated; voice, CSBK, Data Header, Idle, Rate 3/4 and +Rate 1 payloads are not added to the human-content buffer. + +Obvious text runs of at least four characters are reported. The known UDP port +signature `5016` (`1398` hexadecimal) gives an `ETSI/ANYTONE` SMS profile hint, +and port `4007` (`0fa7`) gives a `MOTOROLA` SMS hint. These are hints for capture +analysis, not proof that every payload follows that vendor profile. + +Complete `$GPRMC` and `$GNRMC` sentences are converted to decimal latitude and +longitude and include status, UTC, speed in knots and course. Duplicate text or +location observations within the bounded stream buffer are suppressed. + +## Deliberate limits + +This is a capture aid, not a DMR data endpoint. It does not currently: + +- validate or correct BPTC, CRC16, CRC9 or data CRC32; +- perform standards-complete DMR fragmentation/pad/CRC reassembly; +- send confirmed-data ACK/NACK/SACK responses; +- infer a vendor profile; +- decode binary MD-380, Motorola LRRP or Hytera LP locations; +- parse ARS, IP packets or vendor application ACKs; +- write new report-socket event formats. + +Those features need capture-backed fixtures for ETSI/Anytone, MD-380-like, +Hytera and Motorola profiles before they can be implemented reliably. + +## Performance and privacy + +When disabled, the receive path performs only the configuration check and exits. +When enabled, per-burst work and collector memory are bounded. Human-readable +records are concise at `INFO`; detailed `DEBUG` logging can be voluminous, so +use it only while collecting protocol captures on a system with appropriate log +capacity. + +`RAW` and `LOGICAL` may contain private message, location or application data. +Treat these logs as sensitive and disable the feature after capture work. diff --git a/docs/testing.md b/docs/testing.md index bd72919..323e61a 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -130,6 +130,15 @@ OpenBridge/FBP targets; DATA-GATEWAY remains present for protocol-v1 SMS/GPS handling and is not treated as an FBP peer. OBP-originated unit data forwarded to another FBP peer is covered for source server, source repeater, hops, BER and RSSI metadata preservation. +Passive DMR data observability coverage verifies the feature is disabled by +default, configuration parsing can enable it, HBP and FBP/OBP RX bursts are +labelled separately, generic BPTC logical bytes and conservative Data Header or +CSBK fields are logged at DEBUG, while reconstructed UTF-16 SMS-like text and +NMEA RMC coordinates are surfaced at INFO. Collector tests bound stream count +and retained bytes, and routed payload bytes remain unchanged. A dedicated +voice test verifies enabled data logging emits no observer record for voice +frames. Codec coverage also verifies Rate 3/4 traffic remains raw rather than +being subjected to an unimplemented trellis decode. The OpenBridge parser seam is covered for truncated `DMRE` packets so malformed UDP input is logged and discarded before fixed-offset parsing can raise. UDP black-box coverage includes FBP-originated network direct dial to a @@ -166,7 +175,8 @@ FBP-to-HBP and FBP-to-FBP same-TG forwarding. Those tests assert voice bursts B-E keep their DMR payload bytes unless the target TG is intentionally rewritten. Embedded-LC observability coverage verifies accepted in-call Talker Alias and GPS LC cycles are decoded through the standalone validated codec to system log -messages without changing packet routing or mutation behaviour. +messages without changing packet routing or mutation behaviour. Their raw LC +details are DEBUG-only while decoded TA text and GPS coordinates remain INFO. OBP group voice rate-drop coverage verifies per-stream packet-rate protection is calculated from elapsed stream duration rather than the absolute stream start timestamp, ignores short FBP bursts, and only drops sustained over-rate FBP diff --git a/docs/v1x-codex-changelog.md b/docs/v1x-codex-changelog.md index 8bbe891..5febec9 100644 --- a/docs/v1x-codex-changelog.md +++ b/docs/v1x-codex-changelog.md @@ -64,6 +64,22 @@ ## Data Path +- Added optional, reporting-only DMR data/control burst observability behind + global `DATA_PACKET_LOGGING`, disabled by default. Records include HBP/FBP + source metadata, wrapper and air Slot Type, Color Code, sync, raw payload and + generic BPTC logical bytes where applicable. +- Added conservative Data Header DPF and CSBK CSBKO/FID labels without + reassembly, vendor-profile guesses, delivery responses or packet mutation. + CRC/FEC status is explicitly logged as `NOT_CHECKED`. +- Moved raw per-burst metadata to DEBUG. INFO now surfaces bounded reconstructed + ASCII/UTF-16 text, ETSI/Anytone or Motorola SMS port hints, and complete NMEA + RMC coordinates. The collector is capped at 32 streams and 256 bytes per + stream. +- Kept decoded in-call Talker Alias and embedded-LC GPS at INFO while moving + their raw LC bytes and partial TA block detail to DEBUG. +- Kept the observer out of voice bursts and bounded to one RX burst with no + retained reassembly state. Raw/logical payload logging may expose SMS/GPS + content and must be enabled explicitly. - Preserved DMR data forwarding support. - Kept DATA-GATEWAY behavior for protocol-v1 SMS/GPS style handling. - Reported group-addressed data as data/control, not as voice lifecycle. diff --git a/freedmr.cfg b/freedmr.cfg index 67c5c90..1c23673 100644 --- a/freedmr.cfg +++ b/freedmr.cfg @@ -7,6 +7,9 @@ [GLOBAL] # If you join the FreeDMR network, add your server ID here. SERVER_ID: 0 +# Passive DMR data logging: decoded content at INFO, raw bursts at DEBUG. +# May expose private message or location content. +DATA_PACKET_LOGGING: False [REPORTS] diff --git a/freedmr_dmr_codec.py b/freedmr_dmr_codec.py index e1265e2..83c4264 100644 --- a/freedmr_dmr_codec.py +++ b/freedmr_dmr_codec.py @@ -9,7 +9,9 @@ from __future__ import annotations +from collections import OrderedDict from dataclasses import dataclass +import re from bitarray import bitarray @@ -142,6 +144,25 @@ SLOT_TYPE_NAMES = { 0xE: "RES_4", 0xF: "RES_5", } +DMR_DATA_BURST_TYPES = frozenset((0x3, 0x4, 0x5, 0x6, 0x7, 0x8, 0xA)) +DMR_DATA_CODING = { + 0x3: "BPTC_196_96", + 0x4: "BPTC_196_96", + 0x5: "BPTC_196_96", + 0x6: "BPTC_196_96", + 0x7: "BPTC_196_96", + 0x8: "TRELLIS_3_4", + 0xA: "RATE_1", +} +DATA_PACKET_FORMAT_NAMES = { + 0x0: "UDT", + 0x1: "RESPONSE", + 0x2: "UNCONFIRMED", + 0x3: "CONFIRMED", + 0xD: "DEFINED_SHORT_DATA", + 0xE: "RAW_OR_STATUS_SHORT_DATA", + 0xF: "PROPRIETARY", +} EMBEDDED_LC_PAYLOAD_RANGES = ( (0, 11), (16, 27), @@ -194,6 +215,50 @@ class SlotType: corrected: int = 0 +@dataclass(frozen=True) +class DmrDataObservation: + wrapper_data_type: int + wrapper_name: str + color_code: int + air_data_type: int + air_data_type_name: str + type_matches: bool + coding: str + sync: bytes + raw_payload: bytes + logical_pdu: bytes | None = None + data_packet_format: int | None = None + data_packet_format_name: str | None = None + csbko: int | None = None + fid: int | None = None + crc_status: str = "NOT_CHECKED" + + +@dataclass(frozen=True) +class DmrHumanText: + text: str + encoding: str + profile_hint: str + application: str + + +@dataclass(frozen=True) +class DmrHumanLocation: + sentence_type: str + latitude: float + longitude: float + status: str + utc: str + speed_knots: float | None = None + course: float | None = None + + +@dataclass(frozen=True) +class DmrHumanObservation: + texts: tuple[DmrHumanText, ...] = () + location: DmrHumanLocation | None = None + + class EmbeddedLCError(ValueError): pass @@ -206,6 +271,135 @@ class SlotTypeError(ValueError): pass +class DmrDataError(ValueError): + pass + + +def _profile_hint(data: bytes) -> tuple[str, str]: + if b"\x13\x98\x13\x98" in data: + return "ETSI/ANYTONE", "SMS" + if b"\x0f\xa7\x0f\xa7" in data: + return "MOTOROLA", "SMS" + return "UNKNOWN", "TEXT" + + +def _useful_text(text: str) -> bool: + return len(text) >= 4 and any(char.isalpha() for char in text) + + +def _ascii_texts(data: bytes) -> list[tuple[str, str]]: + texts = [] + for match in re.finditer(rb"[\x20-\x7e]{4,}", data): + if match.end() == len(data): + continue + text = match.group().decode("ascii") + if _useful_text(text) and "$GPRMC" not in text and "$GNRMC" not in text: + texts.append((text, "ASCII")) + return texts + + +def _utf16_texts(data: bytes, little_endian: bool) -> list[tuple[str, str]]: + texts = [] + encoding = "UTF-16LE" if little_endian else "UTF-16BE" + pattern = rb"(?:[\x20-\x7e]\x00){4,}" if little_endian else rb"(?:\x00[\x20-\x7e]){4,}" + for match in re.finditer(pattern, data): + if match.end() == len(data): + continue + text = match.group().decode("utf-16-le" if little_endian else "utf-16-be") + if _useful_text(text): + texts.append((text, encoding)) + return texts + + +def _nmea_coordinate(value: str, hemisphere: str, degree_digits: int) -> float: + degrees = float(value[:degree_digits]) + minutes = float(value[degree_digits:]) + coordinate = degrees + (minutes / 60.0) + return -coordinate if hemisphere in ("S", "W") else coordinate + + +def _nmea_rmc_location(data: bytes) -> DmrHumanLocation | None: + match = re.search(rb"\$(?:GP|GN)RMC,[^\r\n\x00]*?(?:\*[0-9A-Fa-f]{2})?(?=[\r\n\x00])", data) + if not match: + return None + try: + fields = match.group().decode("ascii").split(',') + if len(fields) < 9 or not fields[3] or not fields[5]: + return None + return DmrHumanLocation( + sentence_type=fields[0][1:], + utc=fields[1], + status=fields[2], + latitude=_nmea_coordinate(fields[3], fields[4], 2), + longitude=_nmea_coordinate(fields[5], fields[6], 3), + speed_knots=float(fields[7]) if fields[7] else None, + course=float(fields[8]) if fields[8] else None, + ) + except (ValueError, IndexError, UnicodeDecodeError): + return None + + +class DmrDataHumanCollector: + """Bounded, passive reconstruction for obvious text and NMEA data blocks.""" + + def __init__(self, max_streams: int = 32, max_bytes: int = 256): + if max_streams < 1 or max_bytes < 12: + raise ValueError("DMR human collector limits are too small") + self.max_streams = max_streams + self.max_bytes = max_bytes + self._streams = OrderedDict() + + @property + def stream_count(self) -> int: + return len(self._streams) + + @property + def largest_buffer(self) -> int: + return max((len(state['data']) for state in self._streams.values()), default=0) + + def _state(self, key): + state = self._streams.pop(key, None) + if state is None: + state = {'data': bytearray(), 'emitted': set(), 'location': None} + self._streams[key] = state + while len(self._streams) > self.max_streams: + self._streams.popitem(last=False) + return state + + def observe(self, key, observation: DmrDataObservation) -> DmrHumanObservation: + if observation.wrapper_data_type == 0x6: + self._streams.pop(key, None) + self._state(key) + return DmrHumanObservation() + if observation.wrapper_data_type != 0x7 or observation.logical_pdu is None: + return DmrHumanObservation() + + state = self._state(key) + state['data'].extend(observation.logical_pdu) + if len(state['data']) > self.max_bytes: + del state['data'][:-self.max_bytes] + data = bytes(state['data']) + profile_hint, application = _profile_hint(data) + + found = _ascii_texts(data) + found.extend(_utf16_texts(data, little_endian=False)) + found.extend(_utf16_texts(data, little_endian=True)) + texts = [] + for text, encoding in found: + identity = text + if identity in state['emitted']: + continue + state['emitted'].add(identity) + texts.append(DmrHumanText(text, encoding, profile_hint, application)) + + location = _nmea_rmc_location(data) + if location == state['location']: + location = None + elif location is not None: + state['location'] = location + return DmrHumanObservation(tuple(texts), location) + + def bytes_to_bits(data: bytes) -> bitarray: bits = bitarray(endian="big") bits.frombytes(data) @@ -329,6 +523,77 @@ def encode_full_lc(data: bytes, terminator: bool = False) -> bitarray: return interleave_19696(_encode_19696(code)) +def encode_bptc_19696(data: bytes) -> bitarray: + """Encode one generic 12-byte logical PDU with BPTC(196,96).""" + + return interleave_19696(_encode_19696(data)) + + +def decode_bptc_19696(bits: bitarray) -> bytes: + """Extract the 96 information bits without claiming FEC/CRC validation.""" + + if len(bits) != FULL_LC_BITS: + raise DmrDataError("BPTC(196,96) data must be 196 bits") + + deinterleaved = bitarray(FULL_LC_BITS, endian="big") + for index, target in enumerate(FULL_LC_INTERLEAVE_19696): + deinterleaved[index] = bits[target] + + row_data = bitarray(endian="big") + for row in range(9): + start = 1 + (row * 15) + row_data.extend(deinterleaved[start:start + 11]) + + return row_data[3:].tobytes() + + +def inspect_data_burst(payload: bytes, wrapper_data_type: int) -> DmrDataObservation: + """Return bounded, read-only metadata for one non-voice DMR data burst.""" + + if len(payload) != 33: + raise DmrDataError("DMR burst payload must be exactly 33 bytes") + if wrapper_data_type not in DMR_DATA_BURST_TYPES: + raise DmrDataError("wrapper data type is not a DMR data/control burst") + + bits = bytes_to_bits(payload) + info = bits[0:98] + bits[166:264] + slot_type_bits = bits[98:108] + bits[156:166] + air_slot_value = bits_to_int(slot_type_bits[:8]) + color_code = air_slot_value >> 4 + air_data_type = air_slot_value & 0x0F + logical_pdu = None + if DMR_DATA_CODING[wrapper_data_type] == "BPTC_196_96": + logical_pdu = decode_bptc_19696(info) + + data_packet_format = None + data_packet_format_name = None + csbko = None + fid = None + if wrapper_data_type == 0x6 and logical_pdu is not None: + data_packet_format = logical_pdu[0] & 0x0F + data_packet_format_name = DATA_PACKET_FORMAT_NAMES.get(data_packet_format, "UNKNOWN") + elif wrapper_data_type == 0x3 and logical_pdu is not None: + csbko = logical_pdu[0] & 0x3F + fid = logical_pdu[1] + + return DmrDataObservation( + wrapper_data_type=wrapper_data_type, + wrapper_name=SLOT_TYPE_NAMES[wrapper_data_type], + color_code=color_code, + air_data_type=air_data_type, + air_data_type_name=SLOT_TYPE_NAMES[air_data_type], + type_matches=air_data_type == wrapper_data_type, + coding=DMR_DATA_CODING[wrapper_data_type], + sync=bits[108:156].tobytes(), + raw_payload=payload, + logical_pdu=logical_pdu, + data_packet_format=data_packet_format, + data_packet_format_name=data_packet_format_name, + csbko=csbko, + fid=fid, + ) + + def encode_header_lc(data: bytes) -> bitarray: return encode_full_lc(data, terminator=False) diff --git a/tests/harness/deterministic.py b/tests/harness/deterministic.py index f7fa21a..12d13e1 100644 --- a/tests/harness/deterministic.py +++ b/tests/harness/deterministic.py @@ -274,6 +274,7 @@ def minimal_config(system_names: tuple[str, ...] = ("MASTER-A", "MASTER-B")) -> "SUB_ACL": acl_permit_all(), "GEN_STAT_BRIDGES": False, "DATA_GATEWAY": False, + "DATA_PACKET_LOGGING": False, "VALIDATE_SERVER_IDS": False, }, "REPORTS": {"REPORT": False}, diff --git a/tests/harness/udp_blackbox.py b/tests/harness/udp_blackbox.py index 3d788df..c4fb3c2 100644 --- a/tests/harness/udp_blackbox.py +++ b/tests/harness/udp_blackbox.py @@ -504,11 +504,13 @@ def write_bridge_master_config( dial_a_tg: bool = True, dynamic_tg_routing: bool = True, network_direct_dial_slot: int = 1, + data_packet_logging: bool = False, master_extra_config: str = "", ) -> None: global_use_acl_text = "True" if global_use_acl else "False" dial_a_tg_text = "True" if dial_a_tg else "False" dynamic_tg_routing_text = "True" if dynamic_tg_routing else "False" + data_packet_logging_text = "True" if data_packet_logging else "False" systems = [] for name, port in system_ports.items(): systems.append( @@ -586,6 +588,7 @@ ALLOW_NULL_PASSPHRASE: True ANNOUNCEMENT_LANGUAGES: en_GB SERVER_ID: 9990 DATA_GATEWAY: False +DATA_PACKET_LOGGING: {data_packet_logging_text} VALIDATE_SERVER_IDS: False DEBUG_BRIDGES: False ENABLE_API: False @@ -1070,6 +1073,7 @@ class UdpBlackBoxScenario: dial_a_tg: bool = True, dynamic_tg_routing: bool = True, network_direct_dial_slot: int = 1, + data_packet_logging: bool = False, master_extra_config: str = "", fbp_systems: dict[str, int] | None = None, fbp_proto_versions: dict[str, int] | None = None, @@ -1119,6 +1123,7 @@ class UdpBlackBoxScenario: "dial_a_tg": dial_a_tg, "dynamic_tg_routing": dynamic_tg_routing, "network_direct_dial_slot": network_direct_dial_slot, + "data_packet_logging": data_packet_logging, "master_extra_config": master_extra_config, } diff --git a/tests/test_deterministic_harness.py b/tests/test_deterministic_harness.py index d8099d9..c4df88d 100644 --- a/tests/test_deterministic_harness.py +++ b/tests/test_deterministic_harness.py @@ -5,7 +5,13 @@ import tempfile import unittest import config as freedmr_config -from freedmr_dmr_codec import encode_embedded_lc, payload_with_embedded_lc_fragment +from freedmr_dmr_codec import ( + BS_DATA_SYNC, + encode_bptc_19696, + encode_embedded_lc, + encode_slot_type, + payload_with_embedded_lc_fragment, +) from tests.harness.deterministic import ( HBPF_DATA_SYNC, @@ -31,6 +37,12 @@ def embedded_lc_payloads(lc): ) +def data_burst_payload(logical_pdu, data_type, color_code=1): + info = encode_bptc_19696(logical_pdu) + slot_type = encode_slot_type(color_code, data_type) + return (info[:98] + slot_type[:10] + BS_DATA_SYNC + slot_type[10:] + info[98:]).tobytes() + + class PacketSpecTest(unittest.TestCase): def test_packet_spec_builds_parseable_dmrd_payload(self): packet = PacketSpec( @@ -61,6 +73,7 @@ class DeterministicHarnessTest(unittest.TestCase): config_text = """ [GLOBAL] USE_ACL: False +DATA_PACKET_LOGGING: True [ALIASES] TRY_DOWNLOAD: False @@ -81,6 +94,7 @@ DEFAULT_REFLECTOR: 0 os.unlink(path) self.assertIs(parsed["GLOBAL"]["USE_ACL"], False) + self.assertIs(parsed["GLOBAL"]["DATA_PACKET_LOGGING"], True) self.assertEqual(parsed["ALIASES"]["STALE_TIME"], 86400) self.assertTrue(parsed["SYSTEMS"]["MASTER-A"]["DIAL_A_TG"]) self.assertTrue(parsed["SYSTEMS"]["MASTER-A"]["DYNAMIC_TG_ROUTING"]) @@ -114,6 +128,7 @@ DEFAULT_REFLECTOR: 91 self.assertEqual(parsed["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS1"], 0) self.assertEqual(parsed["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"], 91) self.assertEqual(parsed["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"], 91) + self.assertIs(parsed["GLOBAL"]["DATA_PACKET_LOGGING"], False) def test_config_default_dial_ts2_takes_precedence_over_deprecated_default_reflector(self): config_text = """ @@ -2319,6 +2334,183 @@ NETWORK_DIRECT_DIAL_SLOT: {value} self.assertTrue(config["STUN"]) self.assertNotIn("_STUN", config["SYSTEMS"]["OBP-1"]) + def test_data_packet_observer_is_disabled_by_default(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + bridges = active_bridge( + "123", + 123, + (("MASTER-A", 2), ("MASTER-B", 2)), + ) + packet = PacketSpec( + dst_id=123, + slot=2, + call_type="group", + frame_type=HBPF_DATA_SYNC, + dtype_vseq=6, + ) + + with DeterministicScenario(config=config, bridges=bridges) as scenario: + with self.assertLogs(scenario.bm.logger, level="INFO") as logs: + scenario.inject_hbp("MASTER-A", packet) + + self.assertNotIn("*DMR DATA OBSERVE*", "\n".join(logs.output)) + + def test_hbp_data_packet_observer_logs_metadata_and_preserves_payload(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + config["GLOBAL"]["DATA_PACKET_LOGGING"] = True + bridges = active_bridge( + "123", + 123, + (("MASTER-A", 2), ("MASTER-B", 2)), + ) + logical_pdu = bytes.fromhex("8304123456789abcdef00123") + payload = data_burst_payload(logical_pdu, data_type=6, color_code=7) + packet = PacketSpec( + dst_id=123, + slot=2, + call_type="group", + frame_type=HBPF_DATA_SYNC, + dtype_vseq=6, + payload=payload, + ) + + with DeterministicScenario(config=config, bridges=bridges) as scenario: + with self.assertLogs(scenario.bm.logger, level="DEBUG") as logs: + scenario.inject_hbp("MASTER-A", packet) + captured = scenario.capture.for_system("MASTER-B") + + log_output = "\n".join(logs.output) + self.assertIn("*DMR DATA OBSERVE*", log_output) + detail_records = [record for record in logs.records if "*DMR DATA OBSERVE*" in record.getMessage()] + self.assertEqual([record.levelname for record in detail_records], ["DEBUG"]) + self.assertIn("TRANSPORT: HBP", log_output) + self.assertIn("TYPE: 6(DATA_HEAD)", log_output) + self.assertIn("CC: 7", log_output) + self.assertIn("DPF: 3(CONFIRMED)", log_output) + self.assertIn("LOGICAL: " + logical_pdu.hex(), log_output) + self.assertIn("RAW: " + payload.hex(), log_output) + self.assertEqual(captured[0].fields["dmr_payload"], payload) + + def test_hbp_data_packet_observer_surfaces_reassembled_sms_text_at_info(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + config["GLOBAL"]["DATA_PACKET_LOGGING"] = True + bridges = active_bridge("123", 123, (("MASTER-A", 2), ("MASTER-B", 2))) + stream_id = 0x01020341 + blocks = ( + bytes.fromhex("1398139800480045004c004c"), + bytes.fromhex("004f00200042004f00420000"), + ) + + with DeterministicScenario(config=config, bridges=bridges) as scenario: + with self.assertLogs(scenario.bm.logger, level="INFO") as logs: + for seq, block in enumerate(blocks, start=1): + scenario.inject_hbp( + "MASTER-A", + PacketSpec( + dst_id=123, + slot=2, + stream_id=stream_id, + seq=seq, + call_type="group", + frame_type=HBPF_DATA_SYNC, + dtype_vseq=7, + payload=data_burst_payload(block, data_type=7), + ), + ) + + log_output = "\n".join(logs.output) + self.assertIn("*DMR DATA TEXT*", log_output) + self.assertIn("APPLICATION: SMS", log_output) + self.assertIn("PROFILE: ETSI/ANYTONE", log_output) + self.assertIn("ENCODING: UTF-16BE", log_output) + self.assertIn("TEXT: 'HELLO BOB'", log_output) + self.assertNotIn("TEXT: 'HELL'", log_output) + + def test_hbp_data_packet_observer_surfaces_nmea_gps_at_info(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + config["GLOBAL"]["DATA_PACKET_LOGGING"] = True + bridges = active_bridge("123", 123, (("MASTER-A", 2), ("MASTER-B", 2))) + stream_id = 0x01020342 + sentence = b"$GPRMC,123519,A,4807.038,N,01131.000,E,022.4,084.4,230394,,,A*00\r\n" + + with DeterministicScenario(config=config, bridges=bridges) as scenario: + with self.assertLogs(scenario.bm.logger, level="INFO") as logs: + for seq, offset in enumerate(range(0, len(sentence), 12), start=1): + block = sentence[offset:offset + 12].ljust(12, b"\x00") + scenario.inject_hbp( + "MASTER-A", + PacketSpec( + dst_id=123, + slot=2, + stream_id=stream_id, + seq=seq, + call_type="group", + frame_type=HBPF_DATA_SYNC, + dtype_vseq=7, + payload=data_burst_payload(block, data_type=7), + ), + ) + + log_output = "\n".join(logs.output) + self.assertIn("*DMR DATA GPS*", log_output) + self.assertIn("PROFILE: NMEA-RMC", log_output) + self.assertIn("TYPE: GPRMC", log_output) + self.assertIn("LAT: 48.117300", log_output) + self.assertIn("LON: 11.516667", log_output) + self.assertIn("SPEED_KNOTS: 22.4", log_output) + self.assertIn("COURSE: 84.4", log_output) + + def test_obp_data_packet_observer_labels_fbp_source(self): + config = minimal_config(("MASTER-A",)) + add_openbridge_system(config, "OBP-1", network_id=3001) + config["GLOBAL"]["DATA_PACKET_LOGGING"] = True + bridges = active_bridge( + "123", + 123, + (("OBP-1", 1), ("MASTER-A", 2)), + ) + logical_pdu = bytes.fromhex("8510123456789abcdef00123") + packet = PacketSpec( + dst_id=123, + slot=1, + call_type="vcsbk", + frame_type=HBPF_DATA_SYNC, + dtype_vseq=3, + payload=data_burst_payload(logical_pdu, data_type=3), + ) + + with DeterministicScenario(config=config, bridges=bridges) as scenario: + with self.assertLogs(scenario.bm.logger, level="DEBUG") as logs: + scenario.inject_obp("OBP-1", packet) + + log_output = "\n".join(logs.output) + self.assertIn("*DMR DATA OBSERVE*", log_output) + self.assertIn("TRANSPORT: FBP/OBP", log_output) + self.assertIn("CSBKO: 5", log_output) + self.assertIn("FID: 16", log_output) + + def test_data_packet_observer_never_logs_voice_frames(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + config["GLOBAL"]["DATA_PACKET_LOGGING"] = True + bridges = active_bridge( + "91", + 91, + (("MASTER-A", 2), ("MASTER-B", 2)), + ) + packet = PacketSpec( + dst_id=91, + slot=2, + call_type="group", + frame_type=HBPF_VOICE, + dtype_vseq=0, + ) + + with DeterministicScenario(config=config, bridges=bridges) as scenario: + with self.assertLogs(scenario.bm.logger, level="INFO") as logs: + scenario.inject_hbp("MASTER-A", packet) + + self.assertNotIn("*DMR DATA OBSERVE*", "\n".join(logs.output)) + def test_hbp_group_data_reports_as_data_without_voice_timeout_events(self): config = minimal_config(("MASTER-A", "MASTER-B")) config["REPORTS"]["REPORT"] = True @@ -2808,7 +3000,7 @@ NETWORK_DIRECT_DIAL_SLOT: {value} bridges = active_bridge("91", 91, (("MASTER-A", 2), ("MASTER-B", 2))) with DeterministicScenario(config=config, bridges=bridges) as scenario: - with self.assertLogs(scenario.bm.logger, level="INFO") as logs: + with self.assertLogs(scenario.bm.logger, level="DEBUG") as logs: for index, payload in enumerate(self.TA_EMB_LC_PAYLOADS, start=1): scenario.inject_hbp( "MASTER-A", @@ -2826,6 +3018,7 @@ NETWORK_DIRECT_DIAL_SLOT: {value} log_output = "\n".join(logs.output) self.assertIn("*IN-CALL TA*", log_output) self.assertIn("TEXT: 'CALL12'", log_output) + self.assertIn("*IN-CALL TA DETAIL*", log_output) self.assertIn("LC: 04004c43414c4c3132", log_output) def test_in_call_gps_embedded_lc_is_logged(self): @@ -2833,7 +3026,7 @@ NETWORK_DIRECT_DIAL_SLOT: {value} bridges = active_bridge("91", 91, (("MASTER-A", 2), ("MASTER-B", 2))) with DeterministicScenario(config=config, bridges=bridges) as scenario: - with self.assertLogs(scenario.bm.logger, level="INFO") as logs: + with self.assertLogs(scenario.bm.logger, level="DEBUG") as logs: for index, payload in enumerate(self.GPS_EMB_LC_PAYLOADS, start=1): scenario.inject_hbp( "MASTER-A", @@ -2852,6 +3045,7 @@ NETWORK_DIRECT_DIAL_SLOT: {value} self.assertIn("*IN-CALL GPS*", log_output) self.assertIn("LAT: 51.123451", log_output) self.assertIn("LON: -2.123451", log_output) + self.assertIn("*IN-CALL GPS DETAIL*", log_output) self.assertIn("LC: 080007fcfae048b57b", log_output) def test_hbp_vcsbk_data_reports_specific_rx_event_without_generic_duplicate(self): diff --git a/tests/test_freedmr_dmr_codec.py b/tests/test_freedmr_dmr_codec.py index 5010517..d3dd0d3 100644 --- a/tests/test_freedmr_dmr_codec.py +++ b/tests/test_freedmr_dmr_codec.py @@ -1,20 +1,26 @@ import unittest from freedmr_dmr_codec import ( + BS_DATA_SYNC, + DmrDataHumanCollector, + DmrDataError, EmbeddedLCError, FullLCError, LC_SERVICE_OPTIONS_HBLINK_LEGACY, LC_SERVICE_OPTIONS_NORMAL, build_group_voice_lc, SlotTypeError, + decode_bptc_19696, decode_embedded_lc, decode_full_lc, decode_slot_type, embedded_lc_fragment_from_payload, + encode_bptc_19696, encode_embedded_lc, encode_full_lc, encode_full_lc_parity, encode_slot_type, + inspect_data_burst, payload_with_embedded_lc_fragment, rs129_parity, voice, @@ -47,7 +53,120 @@ VOICE_BURST_PAYLOAD = bytes.fromhex( ) +def data_burst_payload(logical_pdu: bytes, data_type: int, color_code: int = 1) -> bytes: + info = encode_bptc_19696(logical_pdu) + slot_type = encode_slot_type(color_code, data_type) + return (info[:98] + slot_type[:10] + BS_DATA_SYNC + slot_type[10:] + info[98:]).tobytes() + + class FreeDmrDmrCodecTest(unittest.TestCase): + def test_human_collector_reassembles_utf16_sms_text_with_profile_hint(self): + collector = DmrDataHumanCollector() + key = ("HBP", "MASTER-A", 0x01020304, 3120001, 91, 2) + first = bytes.fromhex("1398139800480045004c004c") + second = bytes.fromhex("004f00200042004f00420000") + + first_result = collector.observe( + key, + inspect_data_burst(data_burst_payload(first, 0x7), 0x7), + ) + second_result = collector.observe( + key, + inspect_data_burst(data_burst_payload(second, 0x7), 0x7), + ) + + self.assertEqual(first_result.texts, ()) + message = next(item for item in second_result.texts if item.text == "HELLO BOB") + self.assertEqual(message.encoding, "UTF-16BE") + self.assertEqual(message.profile_hint, "ETSI/ANYTONE") + self.assertEqual(message.application, "SMS") + self.assertEqual([item.text for item in second_result.texts].count("HELLO BOB"), 1) + + def test_human_collector_decodes_complete_nmea_rmc_location(self): + collector = DmrDataHumanCollector() + key = ("HBP", "MASTER-A", 0x01020305, 3120001, 91, 2) + sentence = b"$GPRMC,123519,A,4807.038,N,01131.000,E,022.4,084.4,230394,,,A*00\r\n" + location = None + + for offset in range(0, len(sentence), 12): + block = sentence[offset:offset + 12].ljust(12, b"\x00") + result = collector.observe( + key, + inspect_data_burst(data_burst_payload(block, 0x7), 0x7), + ) + location = result.location or location + + self.assertIsNotNone(location) + self.assertEqual(location.sentence_type, "GPRMC") + self.assertAlmostEqual(location.latitude, 48.1173, places=4) + self.assertAlmostEqual(location.longitude, 11.5166667, places=4) + self.assertEqual(location.status, "A") + self.assertAlmostEqual(location.speed_knots, 22.4) + self.assertAlmostEqual(location.course, 84.4) + + def test_human_collector_bounds_stream_count_and_payload_bytes(self): + collector = DmrDataHumanCollector(max_streams=4, max_bytes=24) + observation = inspect_data_burst(data_burst_payload(b"A" * 12, 0x7), 0x7) + + for stream_id in range(8): + key = ("HBP", "MASTER-A", stream_id, 3120001, 91, 2) + collector.observe(key, observation) + collector.observe(key, observation) + collector.observe(key, observation) + + self.assertEqual(collector.stream_count, 4) + self.assertLessEqual(collector.largest_buffer, 24) + + def test_generic_bptc_19696_round_trip_preserves_all_twelve_bytes(self): + logical_pdu = bytes.fromhex("8304123456789abcdef00123") + + self.assertEqual(decode_bptc_19696(encode_bptc_19696(logical_pdu)), logical_pdu) + + def test_data_header_observer_exposes_bounded_raw_and_logical_metadata(self): + logical_pdu = bytes.fromhex("8304123456789abcdef00123") + payload = data_burst_payload(logical_pdu, data_type=0x6, color_code=7) + + observation = inspect_data_burst(payload, wrapper_data_type=0x6) + + self.assertEqual(observation.wrapper_name, "DATA_HEAD") + self.assertEqual(observation.color_code, 7) + self.assertEqual(observation.air_data_type, 0x6) + self.assertTrue(observation.type_matches) + self.assertEqual(observation.coding, "BPTC_196_96") + self.assertEqual(observation.logical_pdu, logical_pdu) + self.assertEqual(observation.data_packet_format, 3) + self.assertEqual(observation.data_packet_format_name, "CONFIRMED") + self.assertIsNone(observation.csbko) + self.assertEqual(observation.raw_payload, payload) + self.assertEqual(len(observation.sync), 6) + + def test_csbk_observer_exposes_opcode_and_feature_id(self): + logical_pdu = bytes.fromhex("8510123456789abcdef00123") + payload = data_burst_payload(logical_pdu, data_type=0x3) + + observation = inspect_data_burst(payload, wrapper_data_type=0x3) + + self.assertEqual(observation.csbko, 5) + self.assertEqual(observation.fid, 0x10) + self.assertIsNone(observation.data_packet_format) + + def test_trellis_data_observer_keeps_raw_payload_without_guessing_decode(self): + payload = bytes(range(33)) + + observation = inspect_data_burst(payload, wrapper_data_type=0x8) + + self.assertEqual(observation.coding, "TRELLIS_3_4") + self.assertIsNone(observation.logical_pdu) + self.assertEqual(observation.raw_payload, payload) + + def test_data_observer_rejects_voice_idle_and_wrong_payload_size(self): + with self.assertRaises(DmrDataError): + inspect_data_burst(b"\x00" * 33, wrapper_data_type=0x1) + with self.assertRaises(DmrDataError): + inspect_data_burst(b"\x00" * 33, wrapper_data_type=0x9) + with self.assertRaises(DmrDataError): + inspect_data_burst(b"\x00" * 32, wrapper_data_type=0x6) + def test_full_lc_header_encode_matches_current_codec_fixture(self): lc = GROUP_VOICE_LC diff --git a/tests/test_udp_blackbox_harness.py b/tests/test_udp_blackbox_harness.py index 01b65db..3d4c3ba 100644 --- a/tests/test_udp_blackbox_harness.py +++ b/tests/test_udp_blackbox_harness.py @@ -4,7 +4,13 @@ import time import unittest from pathlib import Path -from freedmr_dmr_codec import encode_embedded_lc, payload_with_embedded_lc_fragment +from freedmr_dmr_codec import ( + BS_DATA_SYNC, + encode_bptc_19696, + encode_embedded_lc, + encode_slot_type, + payload_with_embedded_lc_fragment, +) from tests.harness.deterministic import ( HBPF_DATA_SYNC, @@ -32,6 +38,12 @@ def embedded_lc_payloads(lc): ) +def data_burst_payload(logical_pdu, data_type, color_code=1): + info = encode_bptc_19696(logical_pdu) + slot_type = encode_slot_type(color_code, data_type) + return (info[:98] + slot_type[:10] + BS_DATA_SYNC + slot_type[10:] + info[98:]).tobytes() + + class UdpBlackBoxHarnessTest(unittest.TestCase): TA_EMB_LC_PAYLOADS = embedded_lc_payloads(bytes.fromhex("04004c43414c4c3132")) GPS_EMB_LC_PAYLOADS = embedded_lc_payloads(bytes.fromhex("080007fcfae048b57b")) @@ -79,6 +91,42 @@ class UdpBlackBoxHarnessTest(unittest.TestCase): self.assertEqual(captured.fields["slot"], 2) self.assertEqual(captured.fields["stream_id"], bytes_4(0x01020304)) + def test_hbp_data_observer_logs_live_human_readable_text(self): + require_udp_integration_enabled() + + blocks = ( + bytes.fromhex("1398139800480045004c004c"), + bytes.fromhex("004f00200042004f00420000"), + ) + with UdpBlackBoxScenario(data_packet_logging=True) as scenario: + master_a = scenario.repeater("MASTER-A", 1001) + try: + master_a.login() + for seq, block in enumerate(blocks, start=1): + master_a.send_dmr( + PacketSpec( + peer_id=1001, + rf_src=3120001, + dst_id=91, + slot=2, + stream_id=0x01020341, + seq=seq, + call_type="group", + frame_type=HBPF_DATA_SYNC, + dtype_vseq=7, + payload=data_burst_payload(block, data_type=7), + ) + ) + output = scenario.process.wait_for_log("TEXT: 'HELLO BOB'", timeout=2.0) + finally: + master_a.close() + + self.assertIn("TRANSPORT: HBP", output) + self.assertIn("*DMR DATA TEXT*", output) + self.assertIn("APPLICATION: SMS", output) + self.assertIn("PROFILE: ETSI/ANYTONE", output) + self.assertIn("ENCODING: UTF-16BE", output) + def test_hotspot_proxy_routes_hbp_repeater_to_master(self): require_udp_integration_enabled() @@ -282,7 +330,7 @@ class UdpBlackBoxHarnessTest(unittest.TestCase): master_b.close() self.assertIn("TEXT: 'CALL12'", output) - self.assertIn("LC: 04004c43414c4c3132", output) + self.assertNotIn("LC:", output) def test_hbp_in_call_gps_is_logged(self): require_udp_integration_enabled() @@ -315,7 +363,7 @@ class UdpBlackBoxHarnessTest(unittest.TestCase): self.assertIn("LAT: 51.123451", output) self.assertIn("LON: -2.123451", output) - self.assertIn("LC: 080007fcfae048b57b", output) + self.assertNotIn("LC:", output) def test_hbp_malformed_short_dmrd_is_ignored_and_later_packet_routes(self): require_udp_integration_enabled()