You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
385 lines
15 KiB
385 lines
15 KiB
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,
|
|
voice_head_term,
|
|
)
|
|
|
|
|
|
GROUP_VOICE_LC = bytes.fromhex("000000000c302f9be5")
|
|
HEADER_LC_BITS = (
|
|
"000010111100001000000100110101100001111000001100001010111001100000001000"
|
|
"010100000111010001100001000000000000001011110100100001110110011100000000"
|
|
"0000000000000010111000000000111111001000010110100111"
|
|
)
|
|
TERMINATOR_LC_BITS = (
|
|
"000010111010110100000100000000100001111010111100001010111110000000001000"
|
|
"001000000111010011000001000100010100001011000111000001110001000100000000"
|
|
"1100010000000010011000000000111001011000010110010100"
|
|
)
|
|
EMBEDDED_LC_BITS = (
|
|
"00001111000011110000011000000110",
|
|
"00010111000100010000000000000110",
|
|
"00001100000000110011010100011011",
|
|
"00010111001101100010001000100010",
|
|
)
|
|
VOICE_HEAD_TERM_PAYLOAD = bytes.fromhex(
|
|
"2b6004101f842dd00df07d41046dff57d75df5de30152e2070b20f803f88c695e2"
|
|
)
|
|
VOICE_BURST_PAYLOAD = bytes.fromhex(
|
|
"b9e881526173002a6bb9e881526134e0f060691173002a6bb9e881526173002a6a"
|
|
)
|
|
|
|
|
|
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
|
|
|
|
encoded = encode_full_lc(lc)
|
|
|
|
self.assertEqual(encoded.to01(), HEADER_LC_BITS)
|
|
|
|
def test_full_lc_terminator_encode_matches_current_codec_fixture(self):
|
|
lc = GROUP_VOICE_LC
|
|
|
|
encoded = encode_full_lc(lc, terminator=True)
|
|
|
|
self.assertEqual(encoded.to01(), TERMINATOR_LC_BITS)
|
|
|
|
def test_embedded_lc_group_voice_encode_matches_current_codec(self):
|
|
lc = GROUP_VOICE_LC
|
|
|
|
encoded = encode_embedded_lc(lc)
|
|
|
|
self.assertEqual(tuple(fragment.to01() for fragment in encoded), EMBEDDED_LC_BITS)
|
|
|
|
def test_current_runtime_compatibility_function_names(self):
|
|
import freedmr_dmr_codec as dmr_codec
|
|
|
|
lc = GROUP_VOICE_LC
|
|
|
|
self.assertEqual(dmr_codec.encode_header_lc(lc).to01(), HEADER_LC_BITS)
|
|
self.assertEqual(dmr_codec.encode_terminator_lc(lc).to01(), TERMINATOR_LC_BITS)
|
|
self.assertEqual(
|
|
tuple(dmr_codec.encode_emblc(lc)[index].to01() for index in (1, 2, 3, 4)),
|
|
EMBEDDED_LC_BITS,
|
|
)
|
|
|
|
def test_voice_head_term_decode_matches_current_codec(self):
|
|
decoded = voice_head_term(VOICE_HEAD_TERM_PAYLOAD)
|
|
|
|
self.assertEqual(decoded["LC"], bytes.fromhex("001020000c302f9be5"))
|
|
self.assertEqual(decoded["CC"], b"\x01")
|
|
self.assertEqual(decoded["DTYPE"], b"\x01")
|
|
self.assertEqual(decoded["SYNC"].to01(), "110111111111010101111101011101011101111101011101")
|
|
|
|
def test_voice_burst_decode_matches_current_codec(self):
|
|
decoded = voice(VOICE_BURST_PAYLOAD)
|
|
|
|
self.assertEqual(
|
|
[ambe.to01() for ambe in decoded["AMBE"]],
|
|
[
|
|
"101110011110100010000001010100100110000101110011000000000010101001101011",
|
|
"101110011110100010000001010100100110000101110011000000000010101001101011",
|
|
"101110011110100010000001010100100110000101110011000000000010101001101010",
|
|
],
|
|
)
|
|
self.assertEqual(decoded["CC"], b"\x01")
|
|
self.assertEqual(decoded["LCSS"], b"\x01")
|
|
self.assertEqual(decoded["EMBED"].to01(), "01001110000011110000011000000110")
|
|
|
|
def test_full_lc_decode_classifies_group_voice(self):
|
|
lc = GROUP_VOICE_LC
|
|
|
|
decoded = decode_full_lc(encode_full_lc(lc))
|
|
|
|
self.assertEqual(decoded.data, lc)
|
|
self.assertEqual(decoded.flco, 0x00)
|
|
self.assertTrue(decoded.is_group_call)
|
|
self.assertFalse(decoded.is_unit_call)
|
|
self.assertEqual(decoded.target_id, 3120)
|
|
self.assertEqual(decoded.source_id, 3120101)
|
|
|
|
def test_build_group_voice_lc_defaults_to_normal_service_options(self):
|
|
lc = build_group_voice_lc(bytes.fromhex("00005b"), bytes.fromhex("2f9be5"))
|
|
|
|
self.assertEqual(lc, bytes.fromhex("00000000005b2f9be5"))
|
|
self.assertEqual(decode_full_lc(encode_full_lc(lc)).service_options, LC_SERVICE_OPTIONS_NORMAL)
|
|
|
|
def test_build_group_voice_lc_can_represent_legacy_hblink_options_explicitly(self):
|
|
lc = build_group_voice_lc(
|
|
bytes.fromhex("00005b"),
|
|
bytes.fromhex("2f9be5"),
|
|
service_options=LC_SERVICE_OPTIONS_HBLINK_LEGACY,
|
|
)
|
|
|
|
self.assertEqual(lc, bytes.fromhex("00002000005b2f9be5"))
|
|
self.assertEqual(decode_full_lc(encode_full_lc(lc)).service_options, LC_SERVICE_OPTIONS_HBLINK_LEGACY)
|
|
|
|
def test_full_lc_decode_classifies_unit_voice(self):
|
|
lc = bytes.fromhex("030000000c302f9be5")
|
|
|
|
decoded = decode_full_lc(encode_full_lc(lc))
|
|
|
|
self.assertEqual(decoded.flco, 0x03)
|
|
self.assertFalse(decoded.is_group_call)
|
|
self.assertTrue(decoded.is_unit_call)
|
|
|
|
def test_full_lc_rs129_parity_matches_header_and_terminator_masks(self):
|
|
lc = bytes.fromhex("001020000c302f9be5")
|
|
|
|
self.assertEqual(rs129_parity(lc), bytes.fromhex("4c42cc"))
|
|
self.assertEqual(encode_full_lc_parity(lc), bytes.fromhex("dad45a"))
|
|
self.assertEqual(encode_full_lc_parity(lc, terminator=True), bytes.fromhex("d5db55"))
|
|
|
|
def test_full_lc_requires_nine_byte_lc(self):
|
|
with self.assertRaises(FullLCError):
|
|
encode_full_lc(b"\x00" * 8)
|
|
|
|
def test_slot_type_encode_matches_current_codec_fixtures(self):
|
|
self.assertEqual(encode_slot_type(1, 0x1).to01(), "00010001101110001100")
|
|
self.assertEqual(encode_slot_type(1, 0x2).to01(), "00010010101001011001")
|
|
self.assertEqual(encode_slot_type(1, 0x3).to01(), "00010011001010110010")
|
|
self.assertEqual(encode_slot_type(1, 0x6).to01(), "00010110000011001110")
|
|
|
|
def test_slot_type_decode_corrects_three_bits(self):
|
|
bits = encode_slot_type(1, 0x2)
|
|
bits[0] = not bits[0]
|
|
bits[7] = not bits[7]
|
|
bits[19] = not bits[19]
|
|
|
|
decoded = decode_slot_type(bits)
|
|
|
|
self.assertEqual(decoded.color_code, 1)
|
|
self.assertEqual(decoded.data_type, 0x2)
|
|
self.assertEqual(decoded.name, "VOICE_LC_TERM")
|
|
self.assertEqual(decoded.corrected, 3)
|
|
|
|
def test_slot_type_decode_rejects_uncorrectable_error(self):
|
|
bits = encode_slot_type(1, 0x2)
|
|
for index in (0, 1, 2, 3):
|
|
bits[index] = not bits[index]
|
|
|
|
with self.assertRaises(SlotTypeError):
|
|
decode_slot_type(bits)
|
|
|
|
def test_slot_type_rejects_values_outside_four_bits(self):
|
|
with self.assertRaises(SlotTypeError):
|
|
encode_slot_type(16, 0x1)
|
|
with self.assertRaises(SlotTypeError):
|
|
encode_slot_type(1, 16)
|
|
|
|
def test_embedded_lc_round_trips_talker_alias_header(self):
|
|
lc = bytes.fromhex("04004c43414c4c3132")
|
|
|
|
encoded = encode_embedded_lc(lc)
|
|
decoded = decode_embedded_lc(encoded)
|
|
|
|
self.assertEqual(decoded.data, lc)
|
|
self.assertEqual(decoded.flco, 0x04)
|
|
self.assertEqual(decoded.corrected, 0)
|
|
|
|
def test_embedded_lc_round_trips_gps_info(self):
|
|
lc = bytes.fromhex("080007fcfae048b57b")
|
|
|
|
encoded = encode_embedded_lc(lc)
|
|
decoded = decode_embedded_lc(encoded)
|
|
|
|
self.assertEqual(decoded.data, lc)
|
|
self.assertEqual(decoded.flco, 0x08)
|
|
|
|
def test_encoded_talker_alias_fragments_match_mmdvmhost_layout_fixture(self):
|
|
lc = bytes.fromhex("04004c43414c4c3132")
|
|
payload = b"\x55" * 33
|
|
|
|
encoded_payloads = [
|
|
payload_with_embedded_lc_fragment(payload, fragment).hex()
|
|
for fragment in encode_embedded_lc(lc)
|
|
]
|
|
|
|
self.assertEqual(
|
|
encoded_payloads,
|
|
[
|
|
"555555555555555555555555555550517092855555555555555555555555555555",
|
|
"555555555555555555555555555550382441d55555555555555555555555555555",
|
|
"5555555555555555555555555555522717b8e55555555555555555555555555555",
|
|
"5555555555555555555555555555522b73ae255555555555555555555555555555",
|
|
],
|
|
)
|
|
|
|
def test_payload_fragment_helpers_extract_and_apply_embedded_lc(self):
|
|
lc = bytes.fromhex("080007fcfae048b57b")
|
|
fragment = encode_embedded_lc(lc)[0]
|
|
original = b"\x55" * 33
|
|
|
|
updated = payload_with_embedded_lc_fragment(original, fragment)
|
|
extracted = embedded_lc_fragment_from_payload(updated)
|
|
|
|
self.assertEqual(extracted, fragment)
|
|
self.assertEqual(updated[:14], original[:14])
|
|
self.assertEqual(updated[19:], original[19:])
|
|
|
|
def test_embedded_lc_decode_corrects_single_bit_error(self):
|
|
lc = bytes.fromhex("04004c43414c4c3132")
|
|
fragments = list(encode_embedded_lc(lc))
|
|
fragments[0] = fragments[0].copy()
|
|
fragments[0][0] = not fragments[0][0]
|
|
|
|
decoded = decode_embedded_lc(fragments)
|
|
|
|
self.assertEqual(decoded.data, lc)
|
|
self.assertEqual(decoded.corrected, 1)
|
|
|
|
def test_embedded_lc_decode_rejects_uncorrectable_error(self):
|
|
lc = bytes.fromhex("04004c43414c4c3132")
|
|
fragments = list(encode_embedded_lc(lc))
|
|
fragments[0] = fragments[0].copy()
|
|
fragments[0][0] = not fragments[0][0]
|
|
fragments[0][8] = not fragments[0][8]
|
|
|
|
with self.assertRaises(EmbeddedLCError):
|
|
decode_embedded_lc(fragments)
|
|
|
|
def test_embedded_lc_requires_nine_byte_lc(self):
|
|
with self.assertRaises(EmbeddedLCError):
|
|
encode_embedded_lc(b"\x00" * 8)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|