DMR Voice Packet Decoding Completed!

This commit is contained in:
Cort Buffington 2016-10-20 20:56:19 -05:00
parent 6fb92d0808
commit 7a1aeeea43
6 changed files with 1690 additions and 795 deletions

42
bptc.py
View File

@ -120,16 +120,16 @@ def encode_terminator_lc(_lc):
# BPTC Embedded LC Decoding Routines
#------------------------------------------------------------------------------
def decode_emblc(_elc_b, _elc_c, _elc_d, _elc_e):
def decode_emblc(_elc):
_binlc = bitarray(endian='big')
_binlc.extend([_elc_b[0],_elc_b[8], _elc_b[16],_elc_b[24],_elc_c[0],_elc_c[8], _elc_c[16],_elc_c[24],_elc_d[0],_elc_d[8] ,_elc_d[16]])
_binlc.extend([_elc_b[1],_elc_b[9], _elc_b[17],_elc_b[25],_elc_c[1],_elc_c[9], _elc_c[17],_elc_c[25],_elc_d[1],_elc_d[9] ,_elc_d[17]])
_binlc.extend([_elc_b[2],_elc_b[10],_elc_b[18],_elc_b[26],_elc_c[2],_elc_c[10],_elc_c[18],_elc_c[26],_elc_d[2],_elc_d[10]])
_binlc.extend([_elc_b[3],_elc_b[11],_elc_b[19],_elc_b[27],_elc_c[3],_elc_c[11],_elc_c[19],_elc_c[27],_elc_d[3],_elc_d[11]])
_binlc.extend([_elc_b[4],_elc_b[12],_elc_b[20],_elc_b[28],_elc_c[4],_elc_c[12],_elc_c[20],_elc_c[28],_elc_d[4],_elc_d[12]])
_binlc.extend([_elc_b[5],_elc_b[13],_elc_b[21],_elc_b[29],_elc_c[5],_elc_c[13],_elc_c[21],_elc_c[29],_elc_d[5],_elc_d[13]])
_binlc.extend([_elc_b[6],_elc_b[14],_elc_b[22],_elc_b[30],_elc_c[6],_elc_c[14],_elc_c[22],_elc_c[30],_elc_d[6],_elc_d[14]])
_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())
@ -202,6 +202,10 @@ if __name__ == '__main__':
# Validation Example
voice_h = '\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 = '\x00\x10\x20\x00\x0c\x30\x2f\x9b\xe5' # \xda\xd4\x5a
@ -213,16 +217,17 @@ if __name__ == '__main__':
t0 = time()
full_lc_dec = decode_full_lc(full_lc_encode)
t1 = time()
lc_decode_time = t1-t0
decode_time = t1-t0
print('VALIDATION ROUTINES:')
print('Original Data: {}, {} bytes'.format(h(lc), len(lc)))
print('Orig Data: {}, {} bytes'.format(h(lc), len(lc)))
print('Orig Encoded: {}, {} bytes'.format(h(voice_hb), len(voice_hb.tobytes())))
print()
print('BPTC(196,96):')
print('Encoded data: {}, {} bytes'.format(h(full_lc_encode.tobytes()), len(full_lc_encode.tobytes())))
print('Encoding time: {} seconds'.format(encode_time))
print('Fast Decode: {}'.format(h(full_lc_dec)))
print('Fast Decode Time: {} seconds'.format(lc_decode_time))
print('Decoded data: {}'.format(h(full_lc_dec)))
print('Decode Time: {} seconds'.format(decode_time))
# Embedded LC
t0 = time()
@ -231,15 +236,12 @@ if __name__ == '__main__':
encode_time = t1 -t0
t0 = time()
decemblc = decode_emblc(emblc[0], emblc[1], emblc[2], emblc[3])
decemblc = decode_emblc(emblc[0] + emblc[1] + emblc[2] + emblc[3])
t1 = time()
decode_time = t1 -t0
print('\nEMBEDDED LC:')
print('Encoded Embedded LC: Burst B:{}, Burst C:{}, Burst D:{}, Burst E:{}'.format(h(emblc[0].tobytes()), h(emblc[1].tobytes()), h(emblc[2].tobytes()), h(emblc[3].tobytes())))
print('Endoder Time:', encode_time)
print('Decoded Embedded LC:', h(decemblc))
print('Decoder Time:', decode_time)
print('Encoded Data: Burst B:{} Burst C:{} Burst D:{} Burst E:{}'.format(h(emblc[0].tobytes()), h(emblc[1].tobytes()), h(emblc[2].tobytes()), h(emblc[3].tobytes())))
print('Endoding Time: {}'.format(encode_time))
print('Decoded data: {}'.format(h(decemblc)))
print('Decoding Time: {}'.format(decode_time))

View File

@ -10,7 +10,7 @@ from __future__ import print_function
from bitarray import bitarray
import bptc
import constants as const
#import constants as const
# Does anybody read this stuff? There's a PEP somewhere that says I should do this.
__author__ = 'Cortney T. Buffington, N0MJS'
@ -42,7 +42,8 @@ def voice_sync(_string):
ambe[0] = burst[0:72]
ambe[1] = burst[72:108] + burst[156:192]
ambe[2] = burst[192:264]
return (ambe)
sync = burst[108:156]
return (ambe, sync)
def voice(_string):
@ -52,11 +53,10 @@ def voice(_string):
ambe[1] = burst[72:108] + burst[156:192]
ambe[2] = burst[192:264]
emb = burst[108:116] + burst[148:156]
embeded = burst[116:148]
cc = (emb[0:4])
# pi = (emb[4:5])
lcss = (emb[5:7])
return (ambe, cc, lcss, embeded)
embedded = burst[116:148]
cc = (to_bytes(emb[0:4]))
lcss = (to_bytes(emb[5:7]))
return (ambe, cc, lcss, embedded)
def to_bytes(_bits):
@ -88,59 +88,67 @@ if __name__ == '__main__':
voice_f = '\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 = '\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'
embedded_lc = bitarray()
print('Header Validation:')
print('DMR PACKET DECODER VALIDATION\n')
print('Header:')
t0 = time()
lc = voice_head_term(data_head)
t1 = time()
print(h(lc[0]), h(lc[1]), h(lc[2]))
print(t1-t0, '\n')
print('LC: OPT-{} SRC-{} DST-{}, SLOT TYPE: CC-{} DTYPE-{}'.format(h(lc[0][0:3]),h(lc[0][3:6]),h(lc[0][6:9]),h(lc[1]),h(lc[2])))
print('Decode Time: {}\n'.format(t1-t0))
print('Voice Burst A Validation:')
print('Voice Burst A:')
t0 = time()
lc = voice_sync(voice_a)
t1 = time()
print(lc[0])
print('VOICE SYNC: {}'.format(h(lc[1])))
print(t1-t0, '\n')
print('Voice Burst B Validation:')
print('Voice Burst B:')
t0 = time()
lc = voice(voice_b)
embedded_lc += lc[3]
t1 = time()
print(lc[0], h(lc[1]), h(lc[2]), h(lc[3].tobytes()))
print('EMB: CC-{} LCSS-{}, EMBEDDED LC: {}'.format(h(lc[1]), h(lc[2]), h(lc[3].tobytes())))
print(t1-t0, '\n')
print('Voice Burst C Validation:')
print('Voice Burst C:')
t0 = time()
lc = voice(voice_c)
embedded_lc += lc[3]
t1 = time()
print(lc[0], h(lc[1]), h(lc[2]), h(lc[3].tobytes()))
print('EMB: CC-{} LCSS-{}, EMBEDDED LC: {}'.format(h(lc[1]), h(lc[2]), h(lc[3].tobytes())))
print(t1-t0, '\n')
print('Voice Burst D Validation:')
print('Voice Burst D:')
t0 = time()
lc = voice(voice_d)
embedded_lc += lc[3]
t1 = time()
print(lc[0], h(lc[1]), h(lc[2]), h(lc[3].tobytes()))
print('EMB: CC-{} LCSS-{}, EMBEDDED LC: {}'.format(h(lc[1]), h(lc[2]), h(lc[3].tobytes())))
print(t1-t0, '\n')
print('Voice Burst E Validation:')
print('Voice Burst E:')
t0 = time()
lc = voice(voice_e)
embedded_lc += lc[3]
embedded_lc = bptc.decode_emblc(embedded_lc)
t1 = time()
print(lc[0], h(lc[1]), h(lc[2]), h(lc[3].tobytes()))
print('EMB: CC-{} LCSS-{}, EMBEDDED LC: {}'.format(h(lc[1]), h(lc[2]), h(lc[3].tobytes())))
print('COMPLETE EMBEDDED LC: {}'.format(h(embedded_lc)))
print(t1-t0, '\n')
print('Voice Burst F Validation:')
print('Voice Burst F:')
t0 = time()
lc = voice(voice_f)
t1 = time()
print(lc[0], h(lc[1]), h(lc[2]), h(lc[3].tobytes()))
print('EMB: CC-{} LCSS-{}, EMBEDDED LC: {}'.format(h(lc[1]), h(lc[2]), h(lc[3].tobytes())))
print(t1-t0, '\n')
print('Terminator Validation:')
print('Terminator:')
t0 = time()
lc = voice_head_term(voice_term)
t1 = time()
print(h(lc[0]), h(lc[1]), h(lc[2]))
print(t1-t0)
print('LC: OPT-{} SRC-{} DST-{} SLOT TYPE: CC-{} DTYPE-{}'.format(h(lc[0][0:3]),h(lc[0][3:6]),h(lc[0][6:9]),h(lc[1]),h(lc[2])))
print('Decode Time: {}\n'.format(t1-t0))

View File

@ -31,40 +31,6 @@ def enc_15113(_data):
csum[3] = _data[0] ^ _data[1] ^ _data[2] ^ _data[4] ^ _data[6] ^ _data[7] ^ _data[10]
return csum
# DECODER - Returns a tuple of (decoded data, True if an error was corrected)
def dec_15113(_data):
chk0 = _data[0] ^ _data[1] ^ _data[2] ^ _data[3] ^ _data[5] ^ _data[7] ^ _data[8]
chk1 = _data[1] ^ _data[2] ^ _data[3] ^ _data[4] ^ _data[6] ^ _data[8] ^ _data[9]
chk2 = _data[2] ^ _data[3] ^ _data[4] ^ _data[5] ^ _data[7] ^ _data[9] ^ _data[10]
chk3 = _data[0] ^ _data[1] ^ _data[2] ^ _data[4] ^ _data[6] ^ _data[7] ^ _data[10]
n = 0
error = False
n |= 0x01 if chk0 != _data[11] else 0x00
n |= 0x02 if chk1 != _data[12] else 0x00
n |= 0x04 if chk2 != _data[13] else 0x00
n |= 0x08 if chk3 != _data[14] else 0x00
if n == 0x01: _data[11] = not _data[11]; return (_data, True)
if n == 0x02: _data[12] = not _data[12]; return (_data, True)
if n == 0x04: _data[13] = not _data[13]; return (_data, True)
if n == 0x08: _data[14] = not _data[14]; return (_data, True)
if n == 0x09: _data[0] = not _data[0]; return (_data, True)
if n == 0x0b: _data[1] = not _data[1]; return (_data, True)
if n == 0x0f: _data[2] = not _data[2]; return (_data, True)
if n == 0x07: _data[3] = not _data[3]; return (_data, True)
if n == 0x0e: _data[4] = not _data[4]; return (_data, True)
if n == 0x05: _data[5] = not _data[5]; return (_data, True)
if n == 0x0a: _data[6] = not _data[6]; return (_data, True)
if n == 0x0d: _data[7] = not _data[7]; return (_data, True)
if n == 0x03: _data[8] = not _data[8]; return (_data, True)
if n == 0x06: _data[9] = not _data[9]; return (_data, True)
if n == 0x0c: _data[10] = not _data[10]; return (_data, True)
return (_data, False)
#------------------------------------------------------------------------------
# Hamming 13,9,3 routines
@ -79,38 +45,7 @@ def enc_1393(_data):
csum[3] = _data[0] ^ _data[2] ^ _data[4] ^ _data[5] ^ _data[8]
return csum
# DECODER - Returns a tuple of (decoded data, True if an error was corrected)
def dec_1393(_data):
chk0 = _data[0] ^ _data[1] ^ _data[3] ^ _data[5] ^ _data[6]
chk1 = _data[0] ^ _data[1] ^ _data[2] ^ _data[4] ^ _data[6] ^ _data[7]
chk2 = _data[0] ^ _data[1] ^ _data[2] ^ _data[3] ^ _data[5] ^ _data[7] ^ _data[8]
chk3 = _data[0] ^ _data[2] ^ _data[4] ^ _data[5] ^ _data[8]
n = 0
error = False
n |= 0x01 if chk0 != _data[9] else 0x00
n |= 0x02 if chk1 != _data[10] else 0x00
n |= 0x04 if chk2 != _data[11] else 0x00
n |= 0x08 if chk3 != _data[12] else 0x00
if n == 0x01: _data[9] = not _data[9]; return (_data, True)
if n == 0x02: _data[10] = not _data[10]; return (_data, True)
if n == 0x04: _data[11] = not _data[11]; return (_data, True)
if n == 0x08: _data[12] = not _data[12]; return (_data, True)
if n == 0x0f: _data[0] = not _data[0]; return (_data, True)
if n == 0x07: _data[1] = not _data[1]; return (_data, True)
if n == 0x0e: _data[2] = not _data[2]; return (_data, True)
if n == 0x05: _data[3] = not _data[3]; return (_data, True)
if n == 0x0a: _data[4] = not _data[4]; return (_data, True)
if n == 0x0d: _data[5] = not _data[5]; return (_data, True)
if n == 0x03: _data[6] = not _data[6]; return (_data, True)
if n == 0x06: _data[7] = not _data[7]; return (_data, True)
if n == 0x0c: _data[8] = not _data[8]; return (_data, True)
return (_data, False)
#------------------------------------------------------------------------------
# Hamming 16,11,4 routines
#------------------------------------------------------------------------------
@ -124,69 +59,4 @@ def enc_16114(_data):
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
# DECODER - Returns a tuple of (decoded data, True if an error was corrected)
def dec_16114(_data):
chk0 = _data[0] ^ _data[1] ^ _data[2] ^ _data[3] ^ _data[5] ^ _data[7] ^ _data[8]
chk1 = _data[1] ^ _data[2] ^ _data[3] ^ _data[4] ^ _data[6] ^ _data[8] ^ _data[9]
chk2 = _data[2] ^ _data[3] ^ _data[4] ^ _data[5] ^ _data[7] ^ _data[9] ^ _data[10]
chk3 = _data[0] ^ _data[1] ^ _data[2] ^ _data[4] ^ _data[6] ^ _data[7] ^ _data[10]
chk4 = _data[0] ^ _data[2] ^ _data[5] ^ _data[6] ^ _data[8] ^ _data[9] ^ _data[10]
n = 0
error = False
n |= 0x01 if chk0 != _data[11] else 0x00
n |= 0x02 if chk1 != _data[12] else 0x00
n |= 0x04 if chk2 != _data[13] else 0x00
n |= 0x08 if chk3 != _data[14] else 0x00
n |= 0x10 if chk4 != _data[15] else 0x00
if n == 0x01: _data[11] = not _data[11]; return (_data, True)
if n == 0x02: _data[12] = not _data[12]; return (_data, True)
if n == 0x04: _data[13] = not _data[13]; return (_data, True)
if n == 0x08: _data[14] = not _data[14]; return (_data, True)
if n == 0x10: _data[15] = not _data[15]; return (_data, True)
if n == 0x19: _data[0] = not _data[0]; return (_data, True)
if n == 0x0b: _data[1] = not _data[1]; return (_data, True)
if n == 0x1f: _data[2] = not _data[2]; return (_data, True)
if n == 0x07: _data[3] = not _data[3]; return (_data, True)
if n == 0x0e: _data[4] = not _data[4]; return (_data, True)
if n == 0x15: _data[5] = not _data[5]; return (_data, True)
if n == 0x1a: _data[6] = not _data[6]; return (_data, True)
if n == 0x0d: _data[7] = not _data[7]; return (_data, True)
if n == 0x13: _data[8] = not _data[8]; return (_data, True)
if n == 0x16: _data[9] = not _data[9]; return (_data, True)
if n == 0x1c: _data[10] = not _data[10]; return (_data, True)
return (_data, False)
#------------------------------------------------------------------------------
# Used to execute the module directly to run built-in tests
#------------------------------------------------------------------------------
if __name__ == '__main__':
# Validation Example
good_data_15113 = bitarray('0000000000000000000100000011101000000000000000000000110001110011000000100100111110011010110111100101111001011010110101101100010110100110000110000111100111010011101101000010101001110000100101010100')
bad_data_15113 = bitarray('0000000000000000000110000011101000000000000000000000110001110011000000100100111110011010110111100101111001011010110101101100010110100110000110000111100111010011101101000010101001110000100101010100')
def check_15113(_data):
rows = (_data[1:16],_data[16:31],_data[31:46],_data[46:61],_data[61:76],_data[76:91],_data[91:106],_data[106:121],_data[121:136])
print('Processing New Integrity Check')
for row in rows:
print('original data:', row[0:11], 'original parity:', row[11:15])
hamming_dec = dec_15113(row[0:15])
code = hamming_dec[0]
error = hamming_dec[1]
print('\tDECODE: data: ', code[0:11], 'parity: ', code[11:15], 'error:', error)
print('\tENCODE: calculated parity:', enc_15113(row[0:11]))
print()
check_15113(good_data_15113)
check_15113(bad_data_15113)
return csum

View File

@ -72,7 +72,8 @@ class routerMASTER(HBMASTER):
def __init__(self, *args, **kwargs):
HBMASTER.__init__(self, *args, **kwargs)
self.lc_fragments = {'B': '', 'C': '', 'D': '', 'E': '', 'F': ''}
self.embeddec_lc_rx = {'B': '', 'C': '', 'D': '', 'E': '', 'F': ''}
self.embeddec_lc_tx = {'B': '', 'C': '', 'D': '', 'E': '', 'F': ''}
def dmrd_received(self, _radio_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data):
_bits = int_id(_data[15])
@ -114,7 +115,8 @@ class routerCLIENT(HBCLIENT):
def __init__(self, *args, **kwargs):
HBCLIENT.__init__(self, *args, **kwargs)
self.lc_fragments = {'B': '', 'C': '', 'D': '', 'E': '', 'F': ''}
self.embeddec_lc_rx = {'B': '', 'C': '', 'D': '', 'E': '', 'F': ''}
self.embeddec_lc_tx = {'B': '', 'C': '', 'D': '', 'E': '', 'F': ''}
def dmrd_received(self, _radio_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data):
_bits = int_id(_data[15])

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff