Add passive DMR data observability

master
Simon 3 months ago
parent 3a26097445
commit 91a24e91b2

@ -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]

@ -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]

@ -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),

@ -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]

@ -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.

@ -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]

@ -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)

@ -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},

@ -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,
}

@ -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):

@ -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

@ -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()

Loading…
Cancel
Save

Powered by TurnKey Linux.