feat(v2): vendor dmr_utils3 into domain/dmr (V2-P0-008)

Move BPTC/decode/const into the domain layer, merge encode_emblc bit-25 fix,
and drop the external dmr_utils3 pip dependency.
pull/1/head
Rodrigo Pérez 4 months ago
parent 80319f613e
commit d2f140cd04

@ -14,7 +14,6 @@ license = { text = "GPL-3.0-or-later" }
requires-python = ">=3.10" requires-python = ">=3.10"
dependencies = [ dependencies = [
"Twisted>=22.0", "Twisted>=22.0",
"dmr_utils3>=0.1.19",
"pyyaml>=6.0", "pyyaml>=6.0",
"bitarray>=2.0", "bitarray>=2.0",
"cryptography>=41.0", "cryptography>=41.0",

@ -2,7 +2,6 @@
# Python 3.10+ # Python 3.10+
Twisted>=22.0.0 Twisted>=22.0.0
dmr_utils3>=0.1.19
pyyaml>=6.0 pyyaml>=6.0
bitarray>=2.0.0 bitarray>=2.0.0
cryptography>=41.0.0 cryptography>=41.0.0

@ -35,8 +35,8 @@ from time import perf_counter
from typing import Any from typing import Any
from bitarray import bitarray from bitarray import bitarray
from dmr_utils3 import bptc, decode from ..domain.dmr import bptc, decode
from dmr_utils3.const import LC_OPT from ..domain.dmr.const import LC_OPT
from ..domain import HBPF_DATA_SYNC, HBPF_SLT_VHEAD, HBPF_SLT_VTERM, STREAM_TO, bytes_3, bytes_4, int_id from ..domain import HBPF_DATA_SYNC, HBPF_SLT_VHEAD, HBPF_SLT_VTERM, STREAM_TO, bytes_3, bytes_4, int_id
from ..domain.talker_alias import DMRA_BLOCK_COUNT from ..domain.talker_alias import DMRA_BLOCK_COUNT

@ -0,0 +1,10 @@
# ADN DMR Peer Server - DMR codec (domain)
# Copyright (C) 2026 Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
#
# Vendored from dmr_utils3 (HBlink / N0MJS, G4KLX). GPL v3 — see file headers in submodules.
"""Pure DMR BPTC/decode/const (no Twisted or I/O)."""
from . import bptc, const, crc, decode, hamming, rs129
__all__ = ["bptc", "const", "crc", "decode", "hamming", "rs129"]

@ -0,0 +1,263 @@
#!/usr/bin/env python
#
###############################################################################
# Copyright (C) 2016-2018 Cortney T. Buffington, N0MJS <n0mjs@me.com>
# Copyright (C) 2015 by Jonathan Naylor G4KLX
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
###############################################################################
from bitarray import bitarray
from . import crc, hamming, rs129
# Does anybody read this stuff? There's a PEP somewhere that says I should do this.
__author__ = 'Cortney T. Buffington, N0MJS'
__copyright__ = 'Copyright (c) 2016-2018 Cortney T. Buffington, N0MJS and the K0USY Group'
__credits__ = 'Jonathan Naylor, G4KLX; Ian Wraith'
__license__ = 'GNU GPLv3'
__maintainer__ = 'Cort Buffington, N0MJS'
__email__ = 'n0mjs@me.com'
#------------------------------------------------------------------------------
# Interleaver Index
#------------------------------------------------------------------------------
INDEX_181 = (
0, 181, 166, 151, 136, 121, 106, 91, 76, 61, 46, 31, 16, 1, 182, 167, 152, 137,
122, 107, 92, 77, 62, 47, 32, 17, 2, 183, 168, 153, 138, 123, 108, 93, 78, 63,
48, 33, 18, 3, 184, 169, 154, 139, 124, 109, 94, 79, 64, 49, 34, 19, 4, 185, 170,
155, 140, 125, 110, 95, 80, 65, 50, 35, 20, 5, 186, 171, 156, 141, 126, 111, 96,
81, 66, 51, 36, 21, 6, 187, 172, 157, 142, 127, 112, 97, 82, 67, 52, 37, 22, 7,
188, 173, 158, 143, 128, 113, 98, 83, 68, 53, 38, 23, 8, 189, 174, 159, 144, 129,
114, 99, 84, 69, 54, 39, 24, 9, 190, 175, 160, 145, 130, 115, 100, 85, 70, 55, 40,
25, 10, 191, 176, 161, 146, 131, 116, 101, 86, 71, 56, 41, 26, 11, 192, 177, 162,
147, 132, 117, 102, 87, 72, 57, 42, 27, 12, 193, 178, 163, 148, 133, 118, 103, 88,
73, 58, 43, 28, 13, 194, 179, 164, 149, 134, 119, 104, 89, 74, 59, 44, 29, 14,
195, 180, 165, 150, 135, 120, 105, 90, 75, 60, 45, 30, 15)
#------------------------------------------------------------------------------
# BPTC(196,96) Decoding Routings
#------------------------------------------------------------------------------
def decode_full_lc(_data):
binlc = bitarray(endian='big')
binlc.extend([_data[136],_data[121],_data[106],_data[91], _data[76], _data[61], _data[46], _data[31]])
binlc.extend([_data[152],_data[137],_data[122],_data[107],_data[92], _data[77], _data[62], _data[47], _data[32], _data[17], _data[2] ])
binlc.extend([_data[123],_data[108],_data[93], _data[78], _data[63], _data[48], _data[33], _data[18], _data[3], _data[184],_data[169]])
binlc.extend([_data[94], _data[79], _data[64], _data[49], _data[34], _data[19], _data[4], _data[185],_data[170],_data[155],_data[140]])
binlc.extend([_data[65], _data[50], _data[35], _data[20], _data[5], _data[186],_data[171],_data[156],_data[141],_data[126],_data[111]])
binlc.extend([_data[36], _data[21], _data[6], _data[187],_data[172],_data[157],_data[142],_data[127],_data[112],_data[97], _data[82] ])
binlc.extend([_data[7], _data[188],_data[173],_data[158],_data[143],_data[128],_data[113],_data[98], _data[83]])
'''
This is the rest of the Full LC data -- the RS1293 FEC that we don't need
_data[68],_data[53],_data[174],_data[159],_data[144],_data[129],_data[114],_data[99],_data[84],_data[69],_data[54],_data[39],
_data[24],_data[145],_data[130],_data[115],_data[100],_data[85],_data[70],_data[55],_data[40],_data[25],_data[10],_data[191]
'''
return binlc
#------------------------------------------------------------------------------
# BPTC(196,96) Encoding Routings
#------------------------------------------------------------------------------
def interleave_19696(_data):
inter = bitarray(196, endian='big')
for index in range(196):
inter[INDEX_181[index]] = _data[index] # the real math is slower: deint[index] = _data[(index * 181) % 196]
return inter
# Accepts 12 byte LC header + RS1293, converts to binary and pads for 196 bit
# encode hamming 15113 to rows and 1393 to columns
def encode_19696(_data):
# Create a bitarray from the 4 bytes of LC data (includes RS1293 ECC)
_bdata = bitarray(endian='big')
_bdata.frombytes(_data)
# Insert R0-R3 bits
for i in range(4):
_bdata.insert(0, 0)
# Get row hamming 15,11,3 and append. +1 is to account for R3 that makes an even 196bit string
for index in range(9):
spos = (index*15) + 1
epos= spos + 11
_rowp = hamming.enc_15113(_bdata[spos:epos])
for pbit in range(4):
_bdata.insert(epos+pbit,_rowp[pbit])
# Get column hamming 13,9,3 and append. +1 is to account for R3 that makes an even 196bit string
# Pad out the bitarray to a full 196 bits. Can't insert into 'columns'
for i in range(60):
_bdata.append(0)
column = bitarray(9, endian='big') # Temporary bitarray to hold column data
for col in range(15):
spos = col + 1
for index in range(9):
column[index] = _bdata[spos]
spos += 15
_colp = hamming.enc_1393(column)
# Insert bits into matrix...
cpar = 136 + col # Starting location in the matrix for column bits
for pbit in range(4):
_bdata[cpar] = _colp[pbit]
cpar += 15
return _bdata
def encode_header_lc(_lc):
full_lc = _lc + rs129.lc_header_encode(_lc)
full_lc = encode_19696(full_lc)
full_lc = interleave_19696(full_lc)
return full_lc
def encode_terminator_lc(_lc):
full_lc = _lc + rs129.lc_terminator_encode(_lc)
full_lc = encode_19696(full_lc)
full_lc = interleave_19696(full_lc)
return full_lc
#------------------------------------------------------------------------------
# BPTC Embedded LC Decoding Routines
#------------------------------------------------------------------------------
def decode_emblc(_elc):
_binlc = bitarray(endian='big')
_binlc.extend([_elc[0],_elc[8], _elc[16],_elc[24],_elc[32],_elc[40],_elc[48],_elc[56],_elc[64],_elc[72] ,_elc[80]])
_binlc.extend([_elc[1],_elc[9], _elc[17],_elc[25],_elc[33],_elc[41],_elc[49],_elc[57],_elc[65],_elc[73] ,_elc[81]])
_binlc.extend([_elc[2],_elc[10],_elc[18],_elc[26],_elc[34],_elc[42],_elc[50],_elc[58],_elc[66],_elc[74]])
_binlc.extend([_elc[3],_elc[11],_elc[19],_elc[27],_elc[35],_elc[43],_elc[51],_elc[59],_elc[67],_elc[75]])
_binlc.extend([_elc[4],_elc[12],_elc[20],_elc[28],_elc[36],_elc[44],_elc[52],_elc[60],_elc[68],_elc[76]])
_binlc.extend([_elc[5],_elc[13],_elc[21],_elc[29],_elc[37],_elc[45],_elc[53],_elc[61],_elc[69],_elc[77]])
_binlc.extend([_elc[6],_elc[14],_elc[22],_elc[30],_elc[38],_elc[46],_elc[54],_elc[62],_elc[70],_elc[78]])
return(_binlc.tobytes())
#------------------------------------------------------------------------------
# BPTC Embedded LC Encoding Routines
#------------------------------------------------------------------------------
# Accepts 12 byte LC header + 5-bit checksum, converts to binary and builts out the BPTC
# encoded result with hamming(16,11,4) and parity.
def encode_emblc(_lc):
# Get the 5-bit checksum for the Embedded LC
_csum = crc.csum5(_lc)
# Create a bitarray from the 4 bytes of LC data (includes 5-bit checksum).
_binlc = bitarray(endian='big')
_binlc.frombytes(_lc)
# Insert the checksum bits at the right location in the matrix (this is actually faster than with a for loop)
_binlc.insert(32,_csum[0])
_binlc.insert(43,_csum[1])
_binlc.insert(54,_csum[2])
_binlc.insert(65,_csum[3])
_binlc.insert(76,_csum[4])
# Insert the hamming bits at the right location in the matrix
for index in range(0,112,16):
for hindex,hbit in zip(range(index+11,index+16), hamming.enc_16114(_binlc[index:index+11])):
_binlc.insert(hindex,hbit)
# Insert the column parity bits at the right location in the matrix
for index in range(0,16):
_binlc.insert(index+112, _binlc[index+0] ^ _binlc[index+16] ^ _binlc[index+32] ^ _binlc[index+48] ^ _binlc[index+64] ^ _binlc[index+80] ^ _binlc[index+96])
# Create Embedded LC segments in 48 bit blocks
emblc_b = bitarray(endian='big')
emblc_b.extend([_binlc[0], _binlc[16],_binlc[32],_binlc[48],_binlc[64],_binlc[80],_binlc[96], _binlc[112]])
emblc_b.extend([_binlc[1], _binlc[17],_binlc[33],_binlc[49],_binlc[65],_binlc[81],_binlc[97], _binlc[113]])
emblc_b.extend([_binlc[2], _binlc[18],_binlc[34],_binlc[50],_binlc[66],_binlc[82],_binlc[98], _binlc[114]])
emblc_b.extend([_binlc[3], _binlc[19],_binlc[35],_binlc[51],_binlc[67],_binlc[83],_binlc[99], _binlc[115]])
emblc_c = bitarray(endian='big')
emblc_c.extend([_binlc[4], _binlc[20],_binlc[36],_binlc[52],_binlc[68],_binlc[84],_binlc[100],_binlc[116]])
emblc_c.extend([_binlc[5], _binlc[21],_binlc[37],_binlc[53],_binlc[69],_binlc[85],_binlc[101],_binlc[117]])
emblc_c.extend([_binlc[6], _binlc[22],_binlc[38],_binlc[54],_binlc[70],_binlc[86],_binlc[102],_binlc[118]])
emblc_c.extend([_binlc[7], _binlc[23],_binlc[39],_binlc[55],_binlc[71],_binlc[87],_binlc[103],_binlc[119]])
emblc_d = bitarray(endian='big')
emblc_d.extend([_binlc[8], _binlc[24],_binlc[40],_binlc[56],_binlc[72],_binlc[88],_binlc[104],_binlc[120]])
emblc_d.extend([_binlc[9], _binlc[25],_binlc[41],_binlc[57],_binlc[73],_binlc[89],_binlc[105],_binlc[121]]) # bit 25 (upstream wrongly used 24)
emblc_d.extend([_binlc[10],_binlc[26],_binlc[42],_binlc[58],_binlc[74],_binlc[90],_binlc[106],_binlc[122]])
emblc_d.extend([_binlc[11],_binlc[27],_binlc[43],_binlc[59],_binlc[75],_binlc[91],_binlc[107],_binlc[123]])
emblc_e = bitarray(endian='big')
emblc_e.extend([_binlc[12],_binlc[28],_binlc[44],_binlc[60],_binlc[76],_binlc[92],_binlc[108],_binlc[124]])
emblc_e.extend([_binlc[13],_binlc[29],_binlc[45],_binlc[61],_binlc[77],_binlc[93],_binlc[109],_binlc[125]])
emblc_e.extend([_binlc[14],_binlc[30],_binlc[46],_binlc[62],_binlc[78],_binlc[94],_binlc[110],_binlc[126]])
emblc_e.extend([_binlc[15],_binlc[31],_binlc[47],_binlc[63],_binlc[79],_binlc[95],_binlc[111],_binlc[127]])
return({1: emblc_b, 2: emblc_c, 3: emblc_d, 4: emblc_e})
#------------------------------------------------------------------------------
# Used to execute the module directly to run built-in tests
#------------------------------------------------------------------------------
if __name__ == '__main__':
from binascii import b2a_hex as ahex
from time import time
# Validation Example
voice_h = b'\x2b\x60\x04\x10\x1f\x84\x2d\xd0\x0d\xf0\x7d\x41\x04\x6d\xff\x57\xd7\x5d\xf5\xde\x30\x15\x2e\x20\x70\xb2\x0f\x80\x3f\x88\xc6\x95\xe2'
voice_hb = bitarray(endian='big')
voice_hb.frombytes(voice_h)
voice_hb = voice_hb[0:98] + voice_hb[166:264]
# Header LC -- Terminator similar
lc = b'\x00\x10\x20\x00\x0c\x30\x2f\x9b\xe5' # \xda\xd4\x5a
t0 = time()
full_lc_encode = encode_header_lc(lc)
t1 = time()
encode_time = t1-t0
t0 = time()
full_lc_dec = decode_full_lc(full_lc_encode)
t1 = time()
decode_time = t1-t0
print('VALIDATION ROUTINES:')
print('Orig Data: {}, {} bytes'.format(ahex(lc), len(lc)))
print('Orig Encoded: {}, {} bytes'.format(ahex(voice_hb.tobytes()), len(voice_hb.tobytes())))
print()
print('BPTC(196,96):')
print('Encoded data: {}, {} bytes'.format(ahex(full_lc_encode.tobytes()), len(full_lc_encode.tobytes())))
print('Encoding time: {} seconds'.format(encode_time))
print('Decoded data: {}'.format(ahex(full_lc_dec.tobytes())))
print('Decode Time: {} seconds'.format(decode_time))
# Embedded LC
t0 = time()
emblc = encode_emblc(lc)
t1 = time()
encode_time = t1 -t0
t0 = time()
decemblc = decode_emblc(emblc[1] + emblc[2] + emblc[3] + emblc[4])
t1 = time()
decode_time = t1 -t0
print('\nEMBEDDED LC:')
print('Encoded Data: Burst B:{} Burst C:{} Burst D:{} Burst E:{}'.format(ahex(emblc[1].tobytes()), ahex(emblc[2].tobytes()), ahex(emblc[3].tobytes()), ahex(emblc[4].tobytes())))
print('Endoding Time: {}'.format(encode_time))
print('Decoded data: {}'.format(ahex(decemblc)))
print('Decoding Time: {}'.format(decode_time))

@ -0,0 +1,132 @@
#!/usr/bin/env python
#
###############################################################################
# Copyright (C) 2016-2018 Cortney T. Buffington, N0MJS <n0mjs@me.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
###############################################################################
from bitarray import bitarray
__author__ = 'Cortney T. Buffington, N0MJS'
__copyright__ = 'Copyright (c) 2016-2018 Cortney T. Buffington, N0MJS and the K0USY Group'
__credits__ = ''
__license__ = 'GNU GPLv3'
__maintainer__ = 'Cort Buffington, N0MJS'
__email__ = 'n0mjs@me.com'
# Slot Type Data types
DMR_SLT_VHEAD = b'\x01'
DMR_SLT_VTERM = b'\x02'
# Sync patterns used for LC and Voice Burst A packets
BS_VOICE_SYNC = bitarray()
BS_DATA_SYNC = bitarray()
BS_VOICE_SYNC.frombytes(b'\x75\x5F\xD7\xDF\x75\xF7')
BS_DATA_SYNC.frombytes(b'\xDF\xF5\x7D\x75\xDF\x5D')
SYNC = {
'BS_VOICE': BS_VOICE_SYNC,
'BS_DATA': BS_DATA_SYNC
}
# LC Options - G for Group, U for unit, LC_OPT is legacy for group.
LC_OPT_G = b'\x00\x00\x00'
LC_OPT_U = b'\x03\x00\x00'
LC_OPT = LC_OPT_G
# Precomputed EMB values, where CC always = 1, and PI always = 0
EMB = {
'BURST_B': bitarray('0001001110010001'),
'BURST_C': bitarray('0001011101110100'),
'BURST_D': bitarray('0001011101110100'),
'BURST_E': bitarray('0001010100000111'),
'BURST_F': bitarray('0001000111100010')
}
# Precomputed Slot Type values where CC always = 1
SLOT_TYPE = {
'PI_HEAD': bitarray('00010000001101100111'),
'VOICE_LC_HEAD': bitarray('00010001101110001100'),
'VOICE_LC_TERM': bitarray('00010010101001011001'),
'CSBK': bitarray('00010011001010110010'),
'MBC_HEAD': bitarray('00010100100111110000'),
'MBC_CONT': bitarray('00010101000100011011'),
'DATA_HEAD': bitarray('00010110000011001110'),
'1/2_DATA': bitarray('00010111100000100101'),
'3/4_DATA': bitarray('00011000111010100001'),
'IDLE': bitarray('00011001011001001010'),
'1/1_DATA': bitarray('00011010011110011111'),
'RES_1': bitarray('00011011111101110100'),
'RES_2': bitarray('00011100010000110110'),
'RES_3': bitarray('00011101110011011101'),
'RES_4': bitarray('00011110110100001000'),
'RES_5': bitarray('00011111010111100011')
}
# LC infor for first 3 Bytes:
# Byte 1: PF (1),Res(1),FLCO(6) -- Byte 2: FID(8) -- Byte 3: Service Options(8)
LC_VOICE = {
'FLCO-GRP': bitarray('00000000'),
'FLCO-USR': bitarray('00000011'),
'FID-GENC': bitarray('00000000'),
'FID-MOTO': bitarray('00010000'),
'SVC-OVCM': bitarray('00100000'),
'SVC-NONE': bitarray('00000000')
}
# Data Type as defined for use in the Slot Type Message
DATA_TYPE = {
'PI_HEAD': bitarray('0000'),
'VOICE_LC_HEAD': bitarray('0001'),
'VOICE_LC_TERM': bitarray('0010'),
'CSBK': bitarray('0011'),
'MBC_HEAD': bitarray('0100'),
'MBC_CONT': bitarray('0101'),
'DATA_HEAD': bitarray('0110'),
'1/2_RATE': bitarray('0111'),
'3/4_RATE': bitarray('1000'),
'IDLE': bitarray('1001'),
'1/1_RATE': bitarray('1010'),
'RES_1': bitarray('1011'),
'RES_2': bitarray('1100'),
'RES_3': bitarray('1101'),
'RES_4': bitarray('1110'),
'RES_5': bitarray('1111')
}
'''
EMB: CC(4b), PI(1b), LCSS(2b), EMB Parity(9b - QR 16,7,5)
Slot Type: CC(4b), DataType(4), Slot Type Parity(12b - )
'''
#------------------------------------------------------------------------------
# Used to execute the module directly to run built-in tests
#------------------------------------------------------------------------------
if __name__ == '__main__':
from pprint import pprint
pprint(SYNC)
pprint(EMB)
pprint(SLOT_TYPE)
print(LC_OPT)

@ -0,0 +1,52 @@
#!/usr/bin/env python
#
###############################################################################
# Copyright (C) 2016-2018 Cortney T. Buffington, N0MJS <n0mjs@me.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
###############################################################################
from bitarray import bitarray
# Does anybody read this stuff? There's a PEP somewhere that says I should do this.
__author__ = 'Cortney T. Buffington, N0MJS'
__copyright__ = 'Copyright (c) 2016-2018 Cortney T. Buffington, N0MJS and the K0USY Group'
__credits__ = 'Jonathan Naylor, G4KLX'
__license__ = 'GNU GPLv3'
__maintainer__ = 'Cort Buffington, N0MJS'
__email__ = 'n0mjs@me.com'
def csum5(_data):
accum = 0
assert len(_data) == 9, 'csum5 expected 9 bytes of data and got something else'
for i in range(9):
accum += _data[i]
accum = bytes([accum % 31])
csum = bitarray()
csum.frombytes(accum)
del csum[0:3]
return csum
if __name__ == '__main__':
message = b'\x00\x10\x20\x00\x0c\x30\x2f\x9b\xe5'
result = csum5(message)
print(result, type(result))

@ -0,0 +1,184 @@
#!/usr/bin/env python
#
###############################################################################
# Copyright (C) 2016-2018 Cortney T. Buffington, N0MJS <n0mjs@me.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
###############################################################################
from bitarray import bitarray
from . import bptc
# Does anybody read this stuff? There's a PEP somewhere that says I should do this.
__author__ = 'Cortney T. Buffington, N0MJS'
__copyright__ = 'Copyright (c) 2016-2018 Cortney T. Buffington, N0MJS and the K0USY Group'
__credits__ = 'Jonathan Naylor, G4KLX; Ian Wraith'
__license__ = 'GNU GPLv3'
__maintainer__ = 'Cort Buffington, N0MJS'
__email__ = 'n0mjs@me.com'
def to_bits(_string):
_bits = bitarray(endian='big')
_bits.frombytes(_string)
return _bits
def voice_head_term(_string):
burst = to_bits(_string)
info = burst[0:98] + burst[166:264]
slot_type = burst[98:108] + burst[156:166]
sync = burst[108:156]
lc = bptc.decode_full_lc(info).tobytes()
cc = to_bytes(slot_type[0:4])
dtype = to_bytes(slot_type[4:8])
return {'LC': lc, 'CC': cc, 'DTYPE': dtype, 'SYNC': sync}
def voice_sync(_string):
burst = to_bits(_string)
ambe = [0,0,0]
ambe[0] = burst[0:72]
ambe[1] = burst[72:108] + burst[156:192]
ambe[2] = burst[192:264]
sync = burst[108:156]
return {'AMBE': ambe, 'SYNC': sync}
def voice(_string):
burst = to_bits(_string)
ambe = [0,0,0]
ambe[0] = burst[0:72]
ambe[1] = burst[72:108] + burst[156:192]
ambe[2] = burst[192:264]
emb = burst[108:116] + burst[148:156]
embed = burst[116:148]
cc = (to_bytes(emb[0:4]))
lcss = (to_bytes(emb[5:7]))
return {'AMBE': ambe, 'CC': cc, 'LCSS': lcss, 'EMBED': embed}
def to_bytes(_bits):
add_bits = 8 - (len(_bits) % 8)
if add_bits < 8:
for bit in range(add_bits):
_bits.insert(0,0)
_string = _bits.tobytes()
return _string
#------------------------------------------------------------------------------
# Used to execute the module directly to run built-in tests
#------------------------------------------------------------------------------
if __name__ == '__main__':
from binascii import b2a_hex as ahex
from time import time
# SAMPLE, KNOWN GOOD DMR BURSTS
data_head = b'\x2b\x60\x04\x10\x1f\x84\x2d\xd0\x0d\xf0\x7d\x41\x04\x6d\xff\x57\xd7\x5d\xf5\xde\x30\x15\x2e\x20\x70\xb2\x0f\x80\x3f\x88\xc6\x95\xe2'
voice_a = b'\xb9\xe8\x81\x52\x61\x73\x00\x2a\x6b\xb9\xe8\x81\x52\x67\x55\xfd\x7d\xf7\x5f\x71\x73\x00\x2a\x6b\xb9\xe8\x81\x52\x61\x73\x00\x2a\x6a'
voice_b = b'\xb9\xe8\x81\x52\x61\x73\x00\x2a\x6b\xb9\xe8\x81\x52\x61\x34\xe0\xf0\x60\x69\x11\x73\x00\x2a\x6b\xb9\xe8\x81\x52\x61\x73\x00\x2a\x6a'
voice_c = b'\xb9\xe8\x81\x52\x61\x73\x00\x2a\x6b\xb9\xe8\x81\x52\x61\x71\x71\x10\x04\x77\x41\x73\x00\x2a\x6b\xb9\xe8\x81\x52\x61\x73\x00\x2a\x6a'
voice_d = b'\xb9\xe8\x81\x52\x61\x73\x00\x2a\x6b\x95\x4b\xe6\x50\x01\x70\xc0\x31\x81\xb7\x43\x10\xb0\x07\x77\xa6\xc6\xcb\x53\x73\x27\x89\x48\x3a'
voice_e = b'\x86\x5a\xe7\x61\x75\x55\xb5\x06\x01\xb7\x58\xe6\x65\x11\x51\x75\xa0\xf4\xe0\x71\x24\x81\x50\x01\xff\xf5\xa3\x37\x70\x61\x28\xa7\xca'
voice_f = b'\xee\xe7\x81\x75\x74\x61\x4d\xf2\xff\xcc\xf4\xa0\x55\x11\x10\x00\x00\x00\x0e\x24\x30\x59\xe7\xf9\xe9\x08\xa0\x75\x62\x02\xcc\xd6\x22'
voice_term = b'\x2b\x0f\x04\xc4\x1f\x34\x2d\xa8\x0d\x80\x7d\xe1\x04\xad\xff\x57\xd7\x5d\xf5\xd9\x65\x01\x2d\x18\x77\xd2\x03\xc0\x37\x88\xdf\x95\xd1'
embed_lc = bitarray()
print('DMR PACKET DECODER VALIDATION\n')
print('Header:')
t0 = time()
lc = voice_head_term(data_head)
t1 = time()
print('LC: OPT-{} SRC-{} DST-{}, SLOT TYPE: CC-{} DTYPE-{}'.format(ahex(lc['LC'][0:3]),ahex(lc['LC'][3:6]),ahex(lc['LC'][6:9]),ahex(lc['CC']),ahex(lc['DTYPE'])))
print('Decode Time: {}\n'.format(t1-t0))
print('Voice Burst A:')
t0 = time()
pkt = voice_sync(voice_a)
t1 = time()
print('VOICE SYNC: {}'.format(ahex(lc['SYNC'].tobytes())))
print('AMBE 0: {}, {}'.format(pkt['AMBE'][0], len(pkt['AMBE'][0])))
print('AMBE 1: {}, {}'.format(pkt['AMBE'][1], len(pkt['AMBE'][1])))
print('AMBE 2: {}, {}'.format(pkt['AMBE'][1], len(pkt['AMBE'][2])))
print(t1-t0, '\n')
print('Voice Burst B:')
t0 = time()
pkt = voice(voice_b)
embed_lc += pkt['EMBED']
t1 = time()
print('EMB: CC-{} LCSS-{}, EMBEDDED LC: {}'.format(ahex(pkt['CC']), ahex(pkt['LCSS']), ahex(pkt['EMBED'].tobytes())))
print('AMBE 0: {}, {}'.format(pkt['AMBE'][0], len(pkt['AMBE'][0])))
print('AMBE 1: {}, {}'.format(pkt['AMBE'][1], len(pkt['AMBE'][1])))
print('AMBE 2: {}, {}'.format(pkt['AMBE'][1], len(pkt['AMBE'][2])))
print(t1-t0, '\n')
print('Voice Burst C:')
t0 = time()
pkt = voice(voice_c)
embed_lc += pkt['EMBED']
t1 = time()
print('EMB: CC-{} LCSS-{}, EMBEDDED LC: {}'.format(ahex(pkt['CC']), ahex(pkt['LCSS']), ahex(pkt['EMBED'].tobytes())))
print('AMBE 0: {}, {}'.format(pkt['AMBE'][0], len(pkt['AMBE'][0])))
print('AMBE 1: {}, {}'.format(pkt['AMBE'][1], len(pkt['AMBE'][1])))
print('AMBE 2: {}, {}'.format(pkt['AMBE'][1], len(pkt['AMBE'][2])))
print(t1-t0, '\n')
print('Voice Burst D:')
t0 = time()
pkt = voice(voice_d)
embed_lc += pkt['EMBED']
t1 = time()
print('EMB: CC-{} LCSS-{}, EMBEDDED LC: {}'.format(ahex(pkt['CC']), ahex(pkt['LCSS']), ahex(pkt['EMBED'].tobytes())))
print('AMBE 0: {}, {}'.format(pkt['AMBE'][0], len(pkt['AMBE'][0])))
print('AMBE 1: {}, {}'.format(pkt['AMBE'][1], len(pkt['AMBE'][1])))
print('AMBE 2: {}, {}'.format(pkt['AMBE'][1], len(pkt['AMBE'][2])))
print(t1-t0, '\n')
print('Voice Burst E:')
t0 = time()
pkt = voice(voice_e)
embed_lc += pkt['EMBED']
embed_lc = bptc.decode_emblc(embed_lc)
t1 = time()
print('EMB: CC-{} LCSS-{}, EMBEDDED LC: {}'.format(ahex(pkt['CC']), ahex(pkt['LCSS']), ahex(pkt['EMBED'].tobytes())))
print('COMPLETE EMBEDDED LC: {}'.format(ahex(embed_lc)))
print('AMBE 0: {}, {}'.format(pkt['AMBE'][0], len(pkt['AMBE'][0])))
print('AMBE 1: {}, {}'.format(pkt['AMBE'][1], len(pkt['AMBE'][1])))
print('AMBE 2: {}, {}'.format(pkt['AMBE'][1], len(pkt['AMBE'][2])))
print(t1-t0, '\n')
print('Voice Burst F:')
t0 = time()
pkt = voice(voice_f)
t1 = time()
print('EMB: CC-{} LCSS-{}, EMBEDDED LC: {}'.format(ahex(pkt['CC']), ahex(pkt['LCSS']), ahex(pkt['EMBED'].tobytes())))
print('AMBE 0: {}, {}'.format(pkt['AMBE'][0], len(pkt['AMBE'][0])))
print('AMBE 1: {}, {}'.format(pkt['AMBE'][1], len(pkt['AMBE'][1])))
print('AMBE 2: {}, {}'.format(pkt['AMBE'][1], len(pkt['AMBE'][2])))
print(t1-t0, '\n')
print('Terminator:')
t0 = time()
lc = voice_head_term(voice_term)
t1 = time()
print('LC: OPT-{} SRC-{} DST-{} SLOT TYPE: CC-{} DTYPE-{}'.format(ahex(lc['LC'][0:3]),ahex(lc['LC'][3:6]),ahex(lc['LC'][6:9]),ahex(lc['CC']),ahex(lc['DTYPE'])))
print('Decode Time: {}\n'.format(t1-t0))

@ -0,0 +1,75 @@
#!/usr/bin/env python
#
###############################################################################
# Copyright (C) 2016-2018 Cortney T. Buffington, N0MJS <n0mjs@me.com>
# Copyright (C) 2015 by Jonathan Naylor G4KLX
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
###############################################################################
from bitarray import bitarray
# Does anybody read this stuff? There's a PEP somewhere that says I should do this.
__author__ = 'Cortney T. Buffington, N0MJS'
__copyright__ = 'Copyright (c) 2016-2018 Cortney T. Buffington, N0MJS and the K0USY Group'
__credits__ = 'Jonathan Naylor, G4KLX'
__license__ = 'GNU GPLv3'
__maintainer__ = 'Cort Buffington, N0MJS'
__email__ = 'n0mjs@me.com'
#------------------------------------------------------------------------------
# Hamming 15,11,3 routines
#------------------------------------------------------------------------------
# ENCODER- returns a bitarray object containing the hamming checksums
def enc_15113(_data):
csum = bitarray(4)
csum[0] = _data[0] ^ _data[1] ^ _data[2] ^ _data[3] ^ _data[5] ^ _data[7] ^ _data[8]
csum[1] = _data[1] ^ _data[2] ^ _data[3] ^ _data[4] ^ _data[6] ^ _data[8] ^ _data[9]
csum[2] = _data[2] ^ _data[3] ^ _data[4] ^ _data[5] ^ _data[7] ^ _data[9] ^ _data[10]
csum[3] = _data[0] ^ _data[1] ^ _data[2] ^ _data[4] ^ _data[6] ^ _data[7] ^ _data[10]
return csum
#------------------------------------------------------------------------------
# Hamming 13,9,3 routines
#------------------------------------------------------------------------------
# ENCODER - returns a bitarray object containing the hamming checksums
def enc_1393(_data):
csum = bitarray(4)
csum[0] = _data[0] ^ _data[1] ^ _data[3] ^ _data[5] ^ _data[6]
csum[1] = _data[0] ^ _data[1] ^ _data[2] ^ _data[4] ^ _data[6] ^ _data[7]
csum[2] = _data[0] ^ _data[1] ^ _data[2] ^ _data[3] ^ _data[5] ^ _data[7] ^ _data[8]
csum[3] = _data[0] ^ _data[2] ^ _data[4] ^ _data[5] ^ _data[8]
return csum
#------------------------------------------------------------------------------
# Hamming 16,11,4 routines
#------------------------------------------------------------------------------
# ENCODER - returns a bitarray object containing the hamming checksums
def enc_16114(_data):
assert len(_data) == 11, 'Hamming Encoder 16,11,4: Data not 11 bits long'
csum = bitarray(5)
csum[0] = _data[0] ^ _data[1] ^ _data[2] ^ _data[3] ^ _data[5] ^ _data[7] ^ _data[8]
csum[1] = _data[1] ^ _data[2] ^ _data[3] ^ _data[4] ^ _data[6] ^ _data[8] ^ _data[9]
csum[2] = _data[2] ^ _data[3] ^ _data[4] ^ _data[5] ^ _data[7] ^ _data[9] ^ _data[10]
csum[3] = _data[0] ^ _data[1] ^ _data[2] ^ _data[4] ^ _data[6] ^ _data[7] ^ _data[10]
csum[4] = _data[0] ^ _data[2] ^ _data[5] ^ _data[6] ^ _data[8] ^ _data[9] ^ _data[10]
return csum

@ -0,0 +1,151 @@
#!/usr/bin/env python
#
###############################################################################
# Copyright (C) 2016 Cortney T. Buffington, N0MJS <n0mjs@me.com>
# Copyright (C) 2015 by Jonathan Naylor G4KLX
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
###############################################################################
START_MASK = [0x96, 0x96, 0x96]
END_MASK = [0x99, 0x99, 0x99]
NUM_BYTES = 9
NPAR = 3
POLY= [64, 56, 14, 1, 0, 0, 0, 0, 0, 0, 0, 0]
EXP_TABLE = (
0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80, 0x1D, 0x3A, 0x74, 0xE8, 0xCD, 0x87, 0x13, 0x26,
0x4C, 0x98, 0x2D, 0x5A, 0xB4, 0x75, 0xEA, 0xC9, 0x8F, 0x03, 0x06, 0x0C, 0x18, 0x30, 0x60, 0xC0,
0x9D, 0x27, 0x4E, 0x9C, 0x25, 0x4A, 0x94, 0x35, 0x6A, 0xD4, 0xB5, 0x77, 0xEE, 0xC1, 0x9F, 0x23,
0x46, 0x8C, 0x05, 0x0A, 0x14, 0x28, 0x50, 0xA0, 0x5D, 0xBA, 0x69, 0xD2, 0xB9, 0x6F, 0xDE, 0xA1,
0x5F, 0xBE, 0x61, 0xC2, 0x99, 0x2F, 0x5E, 0xBC, 0x65, 0xCA, 0x89, 0x0F, 0x1E, 0x3C, 0x78, 0xF0,
0xFD, 0xE7, 0xD3, 0xBB, 0x6B, 0xD6, 0xB1, 0x7F, 0xFE, 0xE1, 0xDF, 0xA3, 0x5B, 0xB6, 0x71, 0xE2,
0xD9, 0xAF, 0x43, 0x86, 0x11, 0x22, 0x44, 0x88, 0x0D, 0x1A, 0x34, 0x68, 0xD0, 0xBD, 0x67, 0xCE,
0x81, 0x1F, 0x3E, 0x7C, 0xF8, 0xED, 0xC7, 0x93, 0x3B, 0x76, 0xEC, 0xC5, 0x97, 0x33, 0x66, 0xCC,
0x85, 0x17, 0x2E, 0x5C, 0xB8, 0x6D, 0xDA, 0xA9, 0x4F, 0x9E, 0x21, 0x42, 0x84, 0x15, 0x2A, 0x54,
0xA8, 0x4D, 0x9A, 0x29, 0x52, 0xA4, 0x55, 0xAA, 0x49, 0x92, 0x39, 0x72, 0xE4, 0xD5, 0xB7, 0x73,
0xE6, 0xD1, 0xBF, 0x63, 0xC6, 0x91, 0x3F, 0x7E, 0xFC, 0xE5, 0xD7, 0xB3, 0x7B, 0xF6, 0xF1, 0xFF,
0xE3, 0xDB, 0xAB, 0x4B, 0x96, 0x31, 0x62, 0xC4, 0x95, 0x37, 0x6E, 0xDC, 0xA5, 0x57, 0xAE, 0x41,
0x82, 0x19, 0x32, 0x64, 0xC8, 0x8D, 0x07, 0x0E, 0x1C, 0x38, 0x70, 0xE0, 0xDD, 0xA7, 0x53, 0xA6,
0x51, 0xA2, 0x59, 0xB2, 0x79, 0xF2, 0xF9, 0xEF, 0xC3, 0x9B, 0x2B, 0x56, 0xAC, 0x45, 0x8A, 0x09,
0x12, 0x24, 0x48, 0x90, 0x3D, 0x7A, 0xF4, 0xF5, 0xF7, 0xF3, 0xFB, 0xEB, 0xCB, 0x8B, 0x0B, 0x16,
0x2C, 0x58, 0xB0, 0x7D, 0xFA, 0xE9, 0xCF, 0x83, 0x1B, 0x36, 0x6C, 0xD8, 0xAD, 0x47, 0x8E, 0x01,
0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80, 0x1D, 0x3A, 0x74, 0xE8, 0xCD, 0x87, 0x13, 0x26, 0x4C,
0x98, 0x2D, 0x5A, 0xB4, 0x75, 0xEA, 0xC9, 0x8F, 0x03, 0x06, 0x0C, 0x18, 0x30, 0x60, 0xC0, 0x9D,
0x27, 0x4E, 0x9C, 0x25, 0x4A, 0x94, 0x35, 0x6A, 0xD4, 0xB5, 0x77, 0xEE, 0xC1, 0x9F, 0x23, 0x46,
0x8C, 0x05, 0x0A, 0x14, 0x28, 0x50, 0xA0, 0x5D, 0xBA, 0x69, 0xD2, 0xB9, 0x6F, 0xDE, 0xA1, 0x5F,
0xBE, 0x61, 0xC2, 0x99, 0x2F, 0x5E, 0xBC, 0x65, 0xCA, 0x89, 0x0F, 0x1E, 0x3C, 0x78, 0xF0, 0xFD,
0xE7, 0xD3, 0xBB, 0x6B, 0xD6, 0xB1, 0x7F, 0xFE, 0xE1, 0xDF, 0xA3, 0x5B, 0xB6, 0x71, 0xE2, 0xD9,
0xAF, 0x43, 0x86, 0x11, 0x22, 0x44, 0x88, 0x0D, 0x1A, 0x34, 0x68, 0xD0, 0xBD, 0x67, 0xCE, 0x81,
0x1F, 0x3E, 0x7C, 0xF8, 0xED, 0xC7, 0x93, 0x3B, 0x76, 0xEC, 0xC5, 0x97, 0x33, 0x66, 0xCC, 0x85,
0x17, 0x2E, 0x5C, 0xB8, 0x6D, 0xDA, 0xA9, 0x4F, 0x9E, 0x21, 0x42, 0x84, 0x15, 0x2A, 0x54, 0xA8,
0x4D, 0x9A, 0x29, 0x52, 0xA4, 0x55, 0xAA, 0x49, 0x92, 0x39, 0x72, 0xE4, 0xD5, 0xB7, 0x73, 0xE6,
0xD1, 0xBF, 0x63, 0xC6, 0x91, 0x3F, 0x7E, 0xFC, 0xE5, 0xD7, 0xB3, 0x7B, 0xF6, 0xF1, 0xFF, 0xE3,
0xDB, 0xAB, 0x4B, 0x96, 0x31, 0x62, 0xC4, 0x95, 0x37, 0x6E, 0xDC, 0xA5, 0x57, 0xAE, 0x41, 0x82,
0x19, 0x32, 0x64, 0xC8, 0x8D, 0x07, 0x0E, 0x1C, 0x38, 0x70, 0xE0, 0xDD, 0xA7, 0x53, 0xA6, 0x51,
0xA2, 0x59, 0xB2, 0x79, 0xF2, 0xF9, 0xEF, 0xC3, 0x9B, 0x2B, 0x56, 0xAC, 0x45, 0x8A, 0x09, 0x12,
0x24, 0x48, 0x90, 0x3D, 0x7A, 0xF4, 0xF5, 0xF7, 0xF3, 0xFB, 0xEB, 0xCB, 0x8B, 0x0B, 0x16, 0x2C,
0x58, 0xB0, 0x7D, 0xFA, 0xE9, 0xCF, 0x83, 0x1B, 0x36, 0x6C, 0xD8, 0xAD, 0x47, 0x8E, 0x01, 0x00
)
LOG_TABLE = (
0x00, 0x00, 0x01, 0x19, 0x02, 0x32, 0x1A, 0xC6, 0x03, 0xDF, 0x33, 0xEE, 0x1B, 0x68, 0xC7, 0x4B,
0x04, 0x64, 0xE0, 0x0E, 0x34, 0x8D, 0xEF, 0x81, 0x1C, 0xC1, 0x69, 0xF8, 0xC8, 0x08, 0x4C, 0x71,
0x05, 0x8A, 0x65, 0x2F, 0xE1, 0x24, 0x0F, 0x21, 0x35, 0x93, 0x8E, 0xDA, 0xF0, 0x12, 0x82, 0x45,
0x1D, 0xB5, 0xC2, 0x7D, 0x6A, 0x27, 0xF9, 0xB9, 0xC9, 0x9A, 0x09, 0x78, 0x4D, 0xE4, 0x72, 0xA6,
0x06, 0xBF, 0x8B, 0x62, 0x66, 0xDD, 0x30, 0xFD, 0xE2, 0x98, 0x25, 0xB3, 0x10, 0x91, 0x22, 0x88,
0x36, 0xD0, 0x94, 0xCE, 0x8F, 0x96, 0xDB, 0xBD, 0xF1, 0xD2, 0x13, 0x5C, 0x83, 0x38, 0x46, 0x40,
0x1E, 0x42, 0xB6, 0xA3, 0xC3, 0x48, 0x7E, 0x6E, 0x6B, 0x3A, 0x28, 0x54, 0xFA, 0x85, 0xBA, 0x3D,
0xCA, 0x5E, 0x9B, 0x9F, 0x0A, 0x15, 0x79, 0x2B, 0x4E, 0xD4, 0xE5, 0xAC, 0x73, 0xF3, 0xA7, 0x57,
0x07, 0x70, 0xC0, 0xF7, 0x8C, 0x80, 0x63, 0x0D, 0x67, 0x4A, 0xDE, 0xED, 0x31, 0xC5, 0xFE, 0x18,
0xE3, 0xA5, 0x99, 0x77, 0x26, 0xB8, 0xB4, 0x7C, 0x11, 0x44, 0x92, 0xD9, 0x23, 0x20, 0x89, 0x2E,
0x37, 0x3F, 0xD1, 0x5B, 0x95, 0xBC, 0xCF, 0xCD, 0x90, 0x87, 0x97, 0xB2, 0xDC, 0xFC, 0xBE, 0x61,
0xF2, 0x56, 0xD3, 0xAB, 0x14, 0x2A, 0x5D, 0x9E, 0x84, 0x3C, 0x39, 0x53, 0x47, 0x6D, 0x41, 0xA2,
0x1F, 0x2D, 0x43, 0xD8, 0xB7, 0x7B, 0xA4, 0x76, 0xC4, 0x17, 0x49, 0xEC, 0x7F, 0x0C, 0x6F, 0xF6,
0x6C, 0xA1, 0x3B, 0x52, 0x29, 0x9D, 0x55, 0xAA, 0xFB, 0x60, 0x86, 0xB1, 0xBB, 0xCC, 0x3E, 0x5A,
0xCB, 0x59, 0x5F, 0xB0, 0x9C, 0xA9, 0xA0, 0x51, 0x0B, 0xF5, 0x16, 0xEB, 0x7A, 0x75, 0x2C, 0xD7,
0x4F, 0xAE, 0xD5, 0xE9, 0xE6, 0xE7, 0xAD, 0xE8, 0x74, 0xD6, 0xF4, 0xEA, 0xA8, 0x50, 0x58, 0xAF
)
# multiplication using logarithms
def log_mult(a, b):
if a == 0 or b == 0:
return 0
x = LOG_TABLE[a]
y = LOG_TABLE[b]
z = EXP_TABLE[x + y]
return z
# Reed-Solomon (12,9) encoder
def encode(_msg):
assert len(_msg) == 9, 'RS129_encode error: Message not 9 bytes: %s' % print_hex(_msg)
parity = [0x00, 0x00, 0x00]
for i in range(NUM_BYTES):
dbyte = _msg[i] ^ parity[NPAR - 1]
for j in range(NPAR - 1, 0, -1):
parity[j] = parity[j - 1] ^ log_mult(POLY[j], dbyte)
parity[0] = log_mult(POLY[0], dbyte)
return [parity[2], parity[1], parity[0]]
# Apply DMR XOR LC Header MASK
def lc_header_mask(_parity):
xor = [0,0,0]
for i in range(len(_parity)):
xor[i] = _parity[i] ^ START_MASK[i]
return xor
# Apply DMR XOR LC Terminator MASK
def lc_terminator_mask(_parity):
xor = [0,0,0]
for i in range(len(_parity)):
xor[i] = _parity[i] ^ END_MASK[i]
return xor
# All Inclusive function to take an LC string and provide the RS129 string to append
def lc_header_encode(_message):
bin_message = bytearray(_message)
parity = encode(bin_message)
masked_parity = lc_header_mask(parity)
return bytes([masked_parity[0]]) + bytes([masked_parity[1]]) + bytes([masked_parity[2]])
# All Inclusive function to take an LC string and provide the RS129 string to append
def lc_terminator_encode(_message):
bin_message = bytearray(_message)
parity = encode(bin_message)
masked_parity = lc_terminator_mask(parity)
return bytes([masked_parity[0]]) + bytes([masked_parity[1]]) + bytes([masked_parity[2]])
if __name__ == '__main__':
from binascii import b2a_hex as ahex
# For testing the code
def print_hex(_list):
print('[{}]'.format(', '.join(hex(x) for x in _list)))
# Validation Example
message = b'\x00\x10\x20\x00\x0c\x30\x2f\x9b\xe5'
parity_should_be = b'\xda\x4d\x5a'
print('Original Message: {}'.format(ahex(message)))
print('Masked Parity Should be: {}'.format(ahex(parity_should_be)))
parity = lc_header_encode(message)
print('Calculated Masked Parity is: {}'.format(ahex(parity)))

@ -270,11 +270,12 @@ def try_buffer_ta_from_voice_fragments(
) -> bool: ) -> bool:
"""Reassemble embedded LC across voice bursts B–E (vseq 1–4) and store if it is a TA. """Reassemble embedded LC across voice bursts B–E (vseq 1–4) and store if it is a TA.
``dmr_utils3.bptc.decode_emblc`` is correct on properly FEC-encoded input (the encode ``adn_server.domain.dmr.bptc.decode_emblc`` is correct on properly FEC-encoded input (the encode
helper is the buggy one), so a TA sent by a real radio decodes losslessly here. helper had an upstream bit-25 bug, fixed in our vendored copy), so a TA sent by a real radio
decodes losslessly here.
Returns True when a TA block was decoded and stored. Returns True when a TA block was decoded and stored.
""" """
from dmr_utils3 import bptc, decode from .dmr import bptc, decode
if vseq not in (1, 2, 3, 4) or len(dmrpkt) < 33: if vseq not in (1, 2, 3, 4) or len(dmrpkt) < 33:
return False return False

@ -1,72 +0,0 @@
# ADN DMR Peer Server - DMR BPTC embedded LC (dmr_utils3 adapter)
# Copyright (C) 2026 Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
"""Corrected embedded-LC BPTC encoder (drop-in for ``dmr_utils3.bptc.encode_emblc``)."""
from __future__ import annotations
from bitarray import bitarray
from dmr_utils3 import crc, hamming
def encode_emblc(lc: bytes) -> dict[int, bitarray]:
"""Encode a 9-byte embedded LC into burst B–E fragment dicts {1..4: bitarray}.
Upstream ``dmr_utils3.bptc.encode_emblc`` uses ``_binlc[24]`` twice in segment D row 2
instead of ``_binlc[25]``, which can corrupt one bit (e.g. Talker Alias ``Rodrigo`` →
``Rodrigg``). This copy matches upstream except for that index, so decode on radios and
``dmr_utils3.bptc.decode_emblc`` stay aligned.
Use this everywhere ADN generates embedded LC (bridge group LC, TA overlay, parrot).
"""
_csum = crc.csum5(lc)
_binlc = bitarray(endian="big")
_binlc.frombytes(lc)
_binlc.insert(32, _csum[0])
_binlc.insert(43, _csum[1])
_binlc.insert(54, _csum[2])
_binlc.insert(65, _csum[3])
_binlc.insert(76, _csum[4])
for index in range(0, 112, 16):
for hindex, hbit in zip(
range(index + 11, index + 16), hamming.enc_16114(_binlc[index:index + 11])
):
_binlc.insert(hindex, hbit)
for index in range(0, 16):
_binlc.insert(
index + 112,
_binlc[index] ^ _binlc[index + 16] ^ _binlc[index + 32] ^ _binlc[index + 48]
^ _binlc[index + 64] ^ _binlc[index + 80] ^ _binlc[index + 96],
)
def _seg(rows: tuple[tuple[int, ...], ...]) -> bitarray:
out = bitarray(endian="big")
for row in rows:
out.extend([_binlc[i] for i in row])
return out
emblc_b = _seg((
(0, 16, 32, 48, 64, 80, 96, 112),
(1, 17, 33, 49, 65, 81, 97, 113),
(2, 18, 34, 50, 66, 82, 98, 114),
(3, 19, 35, 51, 67, 83, 99, 115),
))
emblc_c = _seg((
(4, 20, 36, 52, 68, 84, 100, 116),
(5, 21, 37, 53, 69, 85, 101, 117),
(6, 22, 38, 54, 70, 86, 102, 118),
(7, 23, 39, 55, 71, 87, 103, 119),
))
emblc_d = _seg((
(8, 24, 40, 56, 72, 88, 104, 120),
(9, 25, 41, 57, 73, 89, 105, 121), # fixed: bit 25 (dmr_utils3 wrongly uses 24)
(10, 26, 42, 58, 74, 90, 106, 122),
(11, 27, 43, 59, 75, 91, 107, 123),
))
emblc_e = _seg((
(12, 28, 44, 60, 76, 92, 108, 124),
(13, 29, 45, 61, 77, 93, 109, 125),
(14, 30, 46, 62, 78, 94, 110, 126),
(15, 31, 47, 63, 79, 95, 111, 127),
))
return {1: emblc_b, 2: emblc_c, 3: emblc_d, 4: emblc_e}

@ -1,7 +1,7 @@
# ADN DMR Peer Server - Talker Alias embedded LC wire encoding # ADN DMR Peer Server - Talker Alias embedded LC wire encoding
# Copyright (C) 2026 Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl> # Copyright (C) 2026 Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
"""Build embedded-LC burst dicts for Talker Alias (uses domain TA rules + ``dmr_bptc``).""" """Build embedded-LC burst dicts for Talker Alias (uses domain TA rules + domain DMR BPTC)."""
from __future__ import annotations from __future__ import annotations
@ -18,7 +18,7 @@ from ..domain.talker_alias import (
talker_alias_decode_complete, talker_alias_decode_complete,
talker_alias_lc_bytes, talker_alias_lc_bytes,
) )
from .dmr_bptc import encode_emblc from ..domain.dmr.bptc import encode_emblc
if TYPE_CHECKING: if TYPE_CHECKING:
from bitarray import bitarray from bitarray import bitarray

@ -37,8 +37,8 @@ from hmac import new as hmac_new
from random import randint from random import randint
from typing import Any, Callable from typing import Any, Callable
from dmr_utils3 import decode from ...domain.dmr import decode
from dmr_utils3.const import LC_OPT from ...domain.dmr.const import LC_OPT
from twisted.internet import reactor, task from twisted.internet import reactor, task
from twisted.internet.protocol import DatagramProtocol from twisted.internet.protocol import DatagramProtocol

@ -29,11 +29,10 @@ from random import randint
from typing import Any, Iterator from typing import Any, Iterator
from bitarray import bitarray from bitarray import bitarray
from dmr_utils3 import bptc from ...domain.dmr import bptc
from dmr_utils3.const import BS_DATA_SYNC, BS_VOICE_SYNC, EMB, LC_OPT, SLOT_TYPE from ...domain.dmr.bptc import encode_emblc
from dmr_utils3.utils import bytes_4 from ...domain.dmr.const import BS_DATA_SYNC, BS_VOICE_SYNC, EMB, LC_OPT, SLOT_TYPE
from ...domain import bytes_4
from ..dmr_bptc import encode_emblc
# Precalculated DMRD byte 15 (slot << 7 | this) # Precalculated DMRD byte 15 (slot << 7 | this)
HEADBITS = 0b00100001 HEADBITS = 0b00100001

@ -74,7 +74,7 @@ from .infrastructure.config_normalizer import (
normalize_peer_config as _normalize_peer_config, normalize_peer_config as _normalize_peer_config,
) )
from .infrastructure.config_reload import BindSpec, reload_server_config from .infrastructure.config_reload import BindSpec, reload_server_config
from .infrastructure.dmr_bptc import encode_emblc from .domain.dmr.bptc import encode_emblc
from .infrastructure.persistence import PickleSubMapStore from .infrastructure.persistence import PickleSubMapStore
from .infrastructure.persistence.alias_loader import DefaultAliasLoader from .infrastructure.persistence.alias_loader import DefaultAliasLoader
from .infrastructure.persistence.keys_store import JsonKeysStore from .infrastructure.persistence.keys_store import JsonKeysStore

Loading…
Cancel
Save

Powered by TurnKey Linux.