diff --git a/API.py b/API.py index 45a5d7b..1649916 100644 --- a/API.py +++ b/API.py @@ -22,7 +22,7 @@ import logging from twisted.web.resource import Resource -from dmr_utils3.utils import bytes_4 +from utils import bytes_4 logger = logging.getLogger(__name__) diff --git a/FreeDMR-MINIMAL.cfg b/FreeDMR-MINIMAL.cfg index 383fc08..68c86a3 100755 --- a/FreeDMR-MINIMAL.cfg +++ b/FreeDMR-MINIMAL.cfg @@ -32,12 +32,15 @@ TGID_TS2_ACL: PERMIT:ALL DEFAULT_UA_TIMER: 60 TS1_STATIC: TS2_STATIC: +DEFAULT_DIAL_TS1: 0 +DEFAULT_DIAL_TS2: 0 DEFAULT_REFLECTOR: 0 SINGLE_MODE: True +DIAL_A_TG: True +DYNAMIC_TG_ROUTING: True VOICE_IDENT: True ANNOUNCEMENT_LANGUAGE: en_GB GENERATOR: 100 ALLOW_UNREG_ID: False PROXY_CONTROL: True OVERRIDE_IDENT_TG: - diff --git a/FreeDMR-SAMPLE-commented.cfg b/FreeDMR-SAMPLE-commented.cfg index 979a8f3..436d290 100755 --- a/FreeDMR-SAMPLE-commented.cfg +++ b/FreeDMR-SAMPLE-commented.cfg @@ -201,9 +201,13 @@ TGID_TS2_ACL: PERMIT:ALL DEFAULT_UA_TIMER: 10 SINGLE_MODE: True VOICE_IDENT: True +DIAL_A_TG: True +DYNAMIC_TG_ROUTING: True #the next three lines no longer have any effect TS1_STATIC: TS2_STATIC: +DEFAULT_DIAL_TS1: 0 +DEFAULT_DIAL_TS2: 0 DEFAULT_REFLECTOR: 0 ANNOUNCEMENT_LANGUAGE: en_GB GENERATOR: 1 diff --git a/FreeDMR-SAMPLE.cfg b/FreeDMR-SAMPLE.cfg index fbeb09e..d0d87b4 100755 --- a/FreeDMR-SAMPLE.cfg +++ b/FreeDMR-SAMPLE.cfg @@ -88,8 +88,12 @@ TGID_TS2_ACL: PERMIT:ALL DEFAULT_UA_TIMER: 60 SINGLE_MODE: True VOICE_IDENT: True +DIAL_A_TG: True +DYNAMIC_TG_ROUTING: True TS1_STATIC: TS2_STATIC: +DEFAULT_DIAL_TS1: 0 +DEFAULT_DIAL_TS2: 0 DEFAULT_REFLECTOR: 0 ANNOUNCEMENT_LANGUAGE: en_GB GENERATOR: 100 diff --git a/bridge.py b/bridge.py index 4ffab35..70bc448 100755 --- a/bridge.py +++ b/bridge.py @@ -43,11 +43,11 @@ from twisted.internet import reactor, task # Things we import from the main hblink module from hblink import HBSYSTEM, OPENBRIDGE, systems, hblink_handler, reportFactory, REPORT_OPCODES, mk_aliases -from dmr_utils3.utils import bytes_3, int_id, get_alias -from dmr_utils3 import decode, bptc, const +import freedmr_dmr_codec as dmr_codec import config import log from const import * +from utils import bytes_3, get_alias, int_id # Stuff for socket reporting import pickle @@ -275,13 +275,13 @@ class routerOBP(OPENBRIDGE): # If we can, use the LC from the voice header as to keep all options intact if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD: - decoded = decode.voice_head_term(dmrpkt) + decoded = dmr_codec.voice_head_term(dmrpkt) self.STATUS[_stream_id]['LC'] = decoded['LC'] # If we don't have a voice header then don't wait to decode the Embedded LC # just make a new one from the HBP header. This is good enough, and it saves lots of time else: - self.STATUS[_stream_id]['LC'] = LC_OPT + _dst_id + _rf_src + self.STATUS[_stream_id]['LC'] = dmr_codec.build_group_voice_lc(_dst_id, _rf_src) logger.info('(%s) *CALL START* STREAM ID: %s SUB: %s (%s) PEER: %s (%s) TGID %s (%s), TS %s', \ @@ -370,9 +370,9 @@ class routerOBP(OPENBRIDGE): } # Generate LCs (full and EMB) for the TX stream dst_lc = b''.join([self.STATUS[_stream_id]['LC'][0:3], _target['TGID'], _rf_src]) - _target_status[_stream_id]['H_LC'] = bptc.encode_header_lc(dst_lc) - _target_status[_stream_id]['T_LC'] = bptc.encode_terminator_lc(dst_lc) - _target_status[_stream_id]['EMB_LC'] = bptc.encode_emblc(dst_lc) + _target_status[_stream_id]['H_LC'] = dmr_codec.encode_header_lc(dst_lc) + _target_status[_stream_id]['T_LC'] = dmr_codec.encode_terminator_lc(dst_lc) + _target_status[_stream_id]['EMB_LC'] = dmr_codec.encode_emblc(dst_lc) logger.info('(%s) Conference Bridge: %s, Call Bridged to OBP System: %s TS: %s, TGID: %s', self._system, _bridge, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) if CONFIG['REPORTS']['REPORT']: @@ -447,9 +447,9 @@ class routerOBP(OPENBRIDGE): _target_status[_target['TS']]['TX_PEER'] = _peer_id # Generate LCs (full and EMB) for the TX stream dst_lc = b''.join([self.STATUS[_stream_id]['LC'][0:3], _target['TGID'], _rf_src]) - _target_status[_target['TS']]['TX_H_LC'] = bptc.encode_header_lc(dst_lc) - _target_status[_target['TS']]['TX_T_LC'] = bptc.encode_terminator_lc(dst_lc) - _target_status[_target['TS']]['TX_EMB_LC'] = bptc.encode_emblc(dst_lc) + _target_status[_target['TS']]['TX_H_LC'] = dmr_codec.encode_header_lc(dst_lc) + _target_status[_target['TS']]['TX_T_LC'] = dmr_codec.encode_terminator_lc(dst_lc) + _target_status[_target['TS']]['TX_EMB_LC'] = dmr_codec.encode_emblc(dst_lc) logger.debug('(%s) Generating TX FULL and EMB LCs for HomeBrew destination: System: %s, TS: %s, TGID: %s', self._system, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) logger.info('(%s) Conference Bridge: %s, Call Bridged to HBP System: %s TS: %s, TGID: %s', self._system, _bridge, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) if CONFIG['REPORTS']['REPORT']: @@ -621,13 +621,13 @@ class routerHBP(HBSYSTEM): # If we can, use the LC from the voice header as to keep all options intact if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD: - decoded = decode.voice_head_term(dmrpkt) + decoded = dmr_codec.voice_head_term(dmrpkt) self.STATUS[_slot]['RX_LC'] = decoded['LC'] # If we don't have a voice header then don't wait to decode it from the Embedded LC # just make a new one from the HBP header. This is good enough, and it saves lots of time else: - self.STATUS[_slot]['RX_LC'] = LC_OPT + _dst_id + _rf_src + self.STATUS[_slot]['RX_LC'] = dmr_codec.build_group_voice_lc(_dst_id, _rf_src) #LoopControl# for system in systems: @@ -702,9 +702,9 @@ class routerHBP(HBSYSTEM): } # Generate LCs (full and EMB) for the TX stream dst_lc = b''.join([self.STATUS[_slot]['RX_LC'][0:3], _target['TGID'], _rf_src]) - _target_status[_stream_id]['H_LC'] = bptc.encode_header_lc(dst_lc) - _target_status[_stream_id]['T_LC'] = bptc.encode_terminator_lc(dst_lc) - _target_status[_stream_id]['EMB_LC'] = bptc.encode_emblc(dst_lc) + _target_status[_stream_id]['H_LC'] = dmr_codec.encode_header_lc(dst_lc) + _target_status[_stream_id]['T_LC'] = dmr_codec.encode_terminator_lc(dst_lc) + _target_status[_stream_id]['EMB_LC'] = dmr_codec.encode_emblc(dst_lc) logger.info('(%s) Conference Bridge: %s, Call Bridged to OBP System: %s TS: %s, TGID: %s', self._system, _bridge, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) if CONFIG['REPORTS']['REPORT']: @@ -775,9 +775,9 @@ class routerHBP(HBSYSTEM): _target_status[_target['TS']]['TX_PEER'] = _peer_id # Generate LCs (full and EMB) for the TX stream dst_lc = self.STATUS[_slot]['RX_LC'][0:3] + _target['TGID'] + _rf_src - _target_status[_target['TS']]['TX_H_LC'] = bptc.encode_header_lc(dst_lc) - _target_status[_target['TS']]['TX_T_LC'] = bptc.encode_terminator_lc(dst_lc) - _target_status[_target['TS']]['TX_EMB_LC'] = bptc.encode_emblc(dst_lc) + _target_status[_target['TS']]['TX_H_LC'] = dmr_codec.encode_header_lc(dst_lc) + _target_status[_target['TS']]['TX_T_LC'] = dmr_codec.encode_terminator_lc(dst_lc) + _target_status[_target['TS']]['TX_EMB_LC'] = dmr_codec.encode_emblc(dst_lc) logger.debug('(%s) Generating TX FULL and EMB LCs for HomeBrew destination: System: %s, TS: %s, TGID: %s', self._system, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) logger.info('(%s) Conference Bridge: %s, Call Bridged to HBP System: %s TS: %s, TGID: %s', self._system, _bridge, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) if CONFIG['REPORTS']['REPORT']: diff --git a/bridge_master.py b/bridge_master.py index 2cd0487..ba3844c 100644 --- a/bridge_master.py +++ b/bridge_master.py @@ -57,14 +57,14 @@ from twisted.web.server import Site # Things we import from the main hblink module from hblink import HBSYSTEM, OPENBRIDGE, systems, hblink_handler, reportFactory, REPORT_OPCODES, mk_aliases, acl_check -from dmr_utils3.utils import bytes_3, int_id, get_alias, bytes_4 -from dmr_utils3 import decode, bptc, const +import freedmr_dmr_codec as dmr_codec +from freedmr_dmr_codec import EmbeddedLCError, decode_embedded_lc as codec_decode_embedded_lc import config from config import acl_build import log from const import * from mk_voice import pkt_gen -from utils import load_json, save_json +from utils import bytes_3, bytes_4, get_alias, int_id, load_json, save_json #from voice_lib import words #Read voices @@ -104,6 +104,15 @@ DIAL_A_TG_PROHIBITED_DEFAULTS = [ ] DIAL_A_TG_MAX = 999999 DMR_ID_MAX = 16777215 +EMB_LC_TA_HEADER = 0x04 +EMB_LC_TA_BLOCKS = {0x05: 1, 0x06: 2, 0x07: 3} +EMB_LC_GPS_INFO = 0x08 +TA_FORMAT_NAMES = { + 0: '7-bit ASCII', + 1: 'ISO 8-bit', + 2: 'UTF-8', + 3: 'UTF-16BE', +} def group_data_event_name(_call_type, _dtype_vseq): if _call_type == 'group' and _dtype_vseq == 6: @@ -181,6 +190,139 @@ def valid_ident_override_tg(tg): return None return parsed +def target_requires_emb_lc_rewrite(_dst_id, _target_tgid): + return _dst_id != _target_tgid + + +# In-call embedded LC observation. This is log-only and must not affect routing +# or packet mutation. +def _signed_value(_value, _bits): + if _value & (1 << (_bits - 1)): + return _value - (1 << _bits) + return _value + +def _bits_to_int(_bits): + value = 0 + for bit in _bits: + value = (value << 1) | int(bit) + return value + +def _decode_ta_7bit(_bits): + chars = [] + for offset in range(0, len(_bits) - 6, 7): + char = _bits_to_int(_bits[offset:offset + 7]) + if char: + chars.append(chr(char)) + return ''.join(chars) + +def _decode_ta_text(_ta_state): + if not _ta_state or 'FORMAT' not in _ta_state: + return '' + + data_format = _ta_state['FORMAT'] + length = _ta_state.get('LENGTH', 0) + chunks = [_ta_state.get('HEADER', b'')] + chunks.extend(_ta_state.get('BLOCKS', {}).get(block, b'') for block in (1, 2, 3)) + payload = b''.join(chunk for chunk in chunks if chunk) + if not payload: + return '' + + if data_format == 0: + text = _decode_ta_7bit(_ta_state.get('BITS', bitarray(endian='big'))) + for block_bits in _ta_state.get('BLOCK_BITS', {}).values(): + text += _decode_ta_7bit(block_bits) + return text.rstrip('\x00 ') + if data_format == 1: + return payload.decode('latin-1', errors='replace').rstrip('\x00 ') + if data_format == 2: + return payload.decode('utf-8', errors='replace').rstrip('\x00 ') + if data_format == 3: + return payload.decode('utf-16-be', errors='replace').rstrip('\x00 ') + return '' + +def _log_talker_alias(_system, _status, _rf_src, _dst_id, _slot, _stream_id, _lc, _flco): + ta_state = _status.setdefault('TA', {'BLOCKS': {}, 'BLOCK_BITS': {}}) + lc_bits = bitarray(endian='big') + lc_bits.frombytes(_lc) + + if _flco == EMB_LC_TA_HEADER: + data_format = (int(_lc[2]) & 0xC0) >> 6 + length = (int(_lc[2]) & 0x3E) >> 1 + ta_state.clear() + ta_state.update({ + 'FORMAT': data_format, + 'LENGTH': length, + 'HEADER': _lc[3:9], + 'BITS': lc_bits[23:72], + 'BLOCKS': {}, + 'BLOCK_BITS': {}, + }) + block_name = 'HEADER' + else: + block_number = EMB_LC_TA_BLOCKS[_flco] + ta_state.setdefault('BLOCKS', {})[block_number] = _lc[2:9] + ta_state.setdefault('BLOCK_BITS', {})[block_number] = lc_bits[16:72] + block_name = 'BLOCK%s' % block_number + + 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', + _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() + ) + +def _log_gps_info(_system, _rf_src, _dst_id, _slot, _stream_id, _lc): + lc_bits = bitarray(endian='big') + lc_bits.frombytes(_lc) + position_error = _bits_to_int(lc_bits[20:23]) + longitude_raw = _signed_value(_bits_to_int(lc_bits[23:48]), 25) + 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.info( + '(%s) *IN-CALL GPS* STREAM ID: %s SUB: %s TGID: %s TS: %s LAT: %.6f LON: %.6f POS_ERR: %s LC: %s', + _system, int_id(_stream_id), int_id(_rf_src), int_id(_dst_id), _slot, + latitude, longitude, position_error, ahex(_lc).decode() + ) + +def log_embedded_lc_observations( + _system, _status, _rf_src, _dst_id, _slot, _stream_id, _frame_type, _dtype_vseq, _dmrpkt +): + if _frame_type != HBPF_VOICE or _dtype_vseq not in (1, 2, 3, 4): + return + + decoded = dmr_codec.voice(_dmrpkt) + fragments = _status.setdefault('EMB_LC_RX', {}) + if _dtype_vseq == 1: + fragments.clear() + fragments[_dtype_vseq] = decoded['EMBED'] + if not all(fragment in fragments for fragment in (1, 2, 3, 4)): + return + + embedded_lc = fragments[1] + fragments[2] + fragments[3] + fragments[4] + fragments.clear() + try: + lc_info = codec_decode_embedded_lc(( + embedded_lc[0:32], + embedded_lc[32:64], + embedded_lc[64:96], + embedded_lc[96:128], + )) + except EmbeddedLCError as error: + logger.debug( + '(%s) Embedded LC decode failed for stream %s: %s', + _system, int_id(_stream_id), error + ) + return + lc = lc_info.data + flco = int(lc[0]) & 0x3F + if flco == EMB_LC_TA_HEADER or flco in EMB_LC_TA_BLOCKS: + _log_talker_alias(_system, _status, _rf_src, _dst_id, _slot, _stream_id, lc, flco) + elif flco == EMB_LC_GPS_INFO: + _log_gps_info(_system, _rf_src, _dst_id, _slot, _stream_id, lc) + #Set header bits #used for slot rewrite and type rewrite def header(slot,call_type,bits): @@ -279,6 +421,8 @@ def make_bridges(_rules): _rules[_bridge].append({'SYSTEM': _confsystem, 'TS': 2, 'TGID': bytes_3(int(_bridge)),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(int(_bridge)),],'RESET': [], 'TIMER': time()}) else: _tmout = CONFIG['SYSTEMS'][_confsystem]['DEFAULT_UA_TIMER'] + if ts1 == False: + _rules[_bridge].append({'SYSTEM': _confsystem, 'TS': 1, 'TGID': bytes_3(9),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [bytes_3(4000)],'ON': [],'RESET': [], 'TIMER': time()}) if ts2 == False: _rules[_bridge].append({'SYSTEM': _confsystem, 'TS': 2, 'TGID': bytes_3(9),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [bytes_3(4000)],'ON': [],'RESET': [], 'TIMER': time()}) @@ -324,16 +468,16 @@ def make_stat_bridge(_tgid): BRIDGES[_tgid_s].append({'SYSTEM': _system, 'TS': 1, 'TGID': _tgid,'ACTIVE': True,'TIMEOUT': '','TO_TYPE': 'STAT','OFF': [],'ON': [],'RESET': [], 'TIMER': time()}) -def make_default_reflector(reflector,_tmout,system): +def make_default_reflector(reflector,_tmout,system,slot=2): bridge = ''.join(['#',str(reflector)]) #_tmout = CONFIG['SYSTEMS'][system]['DEFAULT_UA_TIMER'] if bridge not in BRIDGES: BRIDGES[bridge] = [] - make_single_reflector(bytes_3(reflector),_tmout, system) + make_single_reflector(bytes_3(reflector),_tmout, system, slot) bridgetemp = [] for bridgesystem in BRIDGES[bridge]: - if bridgesystem['SYSTEM'] == system and bridgesystem['TS'] == 2: - bridgetemp.append({'SYSTEM': system, 'TS': 2, 'TGID': bytes_3(9),'ACTIVE': True,'TIMEOUT': _tmout * 60,'TO_TYPE': 'OFF','OFF': [],'ON': [bytes_3(reflector),],'RESET': [], 'TIMER': time() + (_tmout * 60)}) + if bridgesystem['SYSTEM'] == system and bridgesystem['TS'] == slot: + bridgetemp.append({'SYSTEM': system, 'TS': slot, 'TGID': bytes_3(9),'ACTIVE': True,'TIMEOUT': _tmout * 60,'TO_TYPE': 'OFF','OFF': [],'ON': [bytes_3(reflector),],'RESET': [], 'TIMER': time() + (_tmout * 60)}) else: bridgetemp.append(bridgesystem) @@ -344,12 +488,14 @@ def make_default_reflectors(): for system in CONFIG['SYSTEMS']: if CONFIG['SYSTEMS'][system]['MODE'] != 'MASTER': continue - _default_reflector = CONFIG['SYSTEMS'][system]['DEFAULT_REFLECTOR'] - if valid_dial_a_tg_reflector(_default_reflector): - make_default_reflector(_default_reflector,CONFIG['SYSTEMS'][system]['DEFAULT_UA_TIMER'],system) - elif int(_default_reflector) > 0: - logger.warning('(ROUTER) %s default dial-a-tg %s is invalid or prohibited, ignoring', system, _default_reflector) - CONFIG['SYSTEMS'][system]['DEFAULT_REFLECTOR'] = 0 + for slot, key in ((1, 'DEFAULT_DIAL_TS1'), (2, 'DEFAULT_DIAL_TS2')): + _default_reflector = CONFIG['SYSTEMS'][system].get(key, 0) + if valid_dial_a_tg_reflector(_default_reflector): + make_default_reflector(_default_reflector,CONFIG['SYSTEMS'][system]['DEFAULT_UA_TIMER'],system,slot) + elif int(_default_reflector) > 0: + logger.warning('(ROUTER) %s %s %s is invalid or prohibited, ignoring', system, key, _default_reflector) + CONFIG['SYSTEMS'][system][key] = 0 + CONFIG['SYSTEMS'][system]['DEFAULT_REFLECTOR'] = CONFIG['SYSTEMS'][system].get('DEFAULT_DIAL_TS2', 0) def make_static_tgs(): logger.debug('(ROUTER) setting static TGs') @@ -410,7 +556,7 @@ def reset_static_tg(tg,ts,_tmout,system): #BRIDGES[bridge] = bridgetemp #print(BRIDGES[bridge]) -def reset_all_reflector_system(_tmout,resetSystem): +def reset_all_reflector_system(_tmout,resetSystem,resetSlot=None): logger.trace('RST: In reset_all_reflector_system - timeout: %s, resetSystem: %s',_tmout,resetSystem) bt = {} for system in CONFIG['SYSTEMS']: @@ -421,9 +567,9 @@ def reset_all_reflector_system(_tmout,resetSystem): bridgetemp = [] for bridgesystem in BRIDGES[bridge]: logger.trace('RST: for %s in BRIDGES[%s]',bridgesystem,bridge) - if bridgesystem['SYSTEM'] == resetSystem and bridgesystem['TS'] == 2: + if bridgesystem['SYSTEM'] == resetSystem and (resetSlot is None or bridgesystem['TS'] == resetSlot): logger.trace('RST: MATCH: setting inactive for %s',bridgesystem['SYSTEM']) - bridgetemp.append({'SYSTEM': resetSystem, 'TS': 2, 'TGID': bytes_3(9),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(int(bridge[1:])),],'RESET': [], 'TIMER': time() + (_tmout * 60)}) + bridgetemp.append({'SYSTEM': resetSystem, 'TS': bridgesystem['TS'], 'TGID': bytes_3(9),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(int(bridge[1:])),],'RESET': [], 'TIMER': time() + (_tmout * 60)}) else: logger.trace('RST: NO MATCH: using existing: %s',bridgesystem) bridgetemp.append(bridgesystem) @@ -434,7 +580,7 @@ def reset_all_reflector_system(_tmout,resetSystem): BRIDGES[bridge] = bt[bridge] -def make_single_reflector(_tgid,_tmout,_sourcesystem): +def make_single_reflector(_tgid,_tmout,_sourcesystem,_slot=2): _tgid_s = str(int_id(_tgid)) _bridge = ''.join(['#',_tgid_s]) #1 min timeout for echo @@ -445,10 +591,10 @@ def make_single_reflector(_tgid,_tmout,_sourcesystem): #if _system[0:3] != 'OBP': if CONFIG['SYSTEMS'][_system]['MODE'] == 'MASTER': #_tmout = CONFIG['SYSTEMS'][_system]['DEFAULT_UA_TIMER'] - if _system == _sourcesystem: - BRIDGES[_bridge].append({'SYSTEM': _system, 'TS': 2, 'TGID': bytes_3(9),'ACTIVE': True,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [_tgid,],'RESET': [], 'TIMER': time() + (_tmout * 60)}) - else: - BRIDGES[_bridge].append({'SYSTEM': _system, 'TS': 2, 'TGID': bytes_3(9),'ACTIVE': False,'TIMEOUT': CONFIG['SYSTEMS'][_system]['DEFAULT_UA_TIMER'] * 60,'TO_TYPE': 'ON','OFF': [],'ON': [_tgid,],'RESET': [], 'TIMER': time()}) + for _ts in (1, 2): + _active = _system == _sourcesystem and _ts == _slot + _entry_timeout = _tmout if _system == _sourcesystem else CONFIG['SYSTEMS'][_system]['DEFAULT_UA_TIMER'] + BRIDGES[_bridge].append({'SYSTEM': _system, 'TS': _ts, 'TGID': bytes_3(9),'ACTIVE': _active,'TIMEOUT': _entry_timeout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [_tgid,],'RESET': [], 'TIMER': time() + (_entry_timeout * 60) if _active else time()}) if _system[0:3] == 'OBP' and (int_id(_tgid) >= 79 and (int_id(_tgid) < 9990 or int_id(_tgid) > 9999)): BRIDGES[_bridge].append({'SYSTEM': _system, 'TS': 1, 'TGID': _tgid,'ACTIVE': True,'TIMEOUT': '','TO_TYPE': 'NONE','OFF': [],'ON': [],'RESET': [], 'TIMER': time()}) @@ -599,7 +745,7 @@ def bridgeDebug(): for system in CONFIG['SYSTEMS']: bridgeroll = 0 - dialroll = 0 + dialroll = {1: 0, 2: 0} activeroll = 0 for _bridge in BRIDGES: for enabled_system in BRIDGES[_bridge]: @@ -607,7 +753,7 @@ def bridgeDebug(): bridgeroll += 1 if enabled_system['ACTIVE']: if _bridge and _bridge[0:1] == '#': - dialroll += 1 + dialroll[enabled_system['TS']] += 1 activeroll += 1 else: activeroll += 1 @@ -617,31 +763,33 @@ def bridgeDebug(): if bridgeroll: logger.debug('(BRIDGEDEBUG) system %s has %s bridges of which %s are in an ACTIVE state', system, bridgeroll, activeroll) - if dialroll > 1 and CONFIG['SYSTEMS'][system]['MODE'] == 'MASTER': - logger.warning('(BRIDGEDEBUG) system %s has more than one active dial bridge (%s) - fixing',system, dialroll) + for dialslot in (1, 2): + if dialroll[dialslot] <= 1 or CONFIG['SYSTEMS'][system]['MODE'] != 'MASTER': + continue + _tmout = CONFIG['SYSTEMS'][system]['DEFAULT_UA_TIMER'] + logger.warning('(BRIDGEDEBUG) system %s has more than one active dial bridge on TS%s (%s) - fixing',system, dialslot, dialroll[dialslot]) times = {} for _bridge in BRIDGES: for enabled_system in BRIDGES[_bridge]: - if enabled_system['ACTIVE'] and _bridge and _bridge[0:1] == '#': + if enabled_system['ACTIVE'] and _bridge and _bridge[0:1] == '#' and enabled_system['SYSTEM'] == system and enabled_system['TS'] == dialslot: times[enabled_system['TIMER']] = _bridge ordered = sorted(times.keys()) #bridgetmout = ordered.pop() #_setbridge = str(times[bridgetmout]) - if CONFIG['SYSTEMS'][system]['MODE'] == 'MASTER': - for _bridge in set(times.values()): - logger.warning('(BRIDGEDEBUG) deactivating system: %s for bridge: %s',system,_bridge) - bridgetemp = [] - for bridgesystem in BRIDGES[_bridge]: - if _bridge[0:1] == '#': - _setbridge = int(_bridge[1:]) - else: - _setbridge = int(_bridge) + for _bridge in set(times.values()): + logger.warning('(BRIDGEDEBUG) deactivating system: %s TS%s for bridge: %s',system,dialslot,_bridge) + bridgetemp = [] + for bridgesystem in BRIDGES[_bridge]: + if _bridge[0:1] == '#': + _setbridge = int(_bridge[1:]) + else: + _setbridge = int(_bridge) - if bridgesystem['SYSTEM'] == system and bridgesystem['TS'] == 2: - bridgetemp.append({'SYSTEM': system, 'TS': 2, 'TGID': bytes_3(9),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(_setbridge),],'RESET': [], 'TIMER': _rst_time + (_tmout * 60)}) - else: - bridgetemp.append(bridgesystem) - BRIDGES[_bridge] = bridgetemp + if bridgesystem['SYSTEM'] == system and bridgesystem['TS'] == dialslot: + bridgetemp.append({'SYSTEM': system, 'TS': dialslot, 'TGID': bytes_3(9),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(_setbridge),],'RESET': [], 'TIMER': _rst_time + (_tmout * 60)}) + else: + bridgetemp.append(bridgesystem) + BRIDGES[_bridge] = bridgetemp logger.info('(BRIDGEDEBUG) The server currently has %s STATic bridges',statroll) @@ -871,10 +1019,11 @@ def disconnectedVoice(system): _nine = bytes_3(9) _source_id = bytes_3(5000) _lang = CONFIG['SYSTEMS'][system]['ANNOUNCEMENT_LANGUAGE'] + _default_reflector = CONFIG['SYSTEMS'][system].get('DEFAULT_DIAL_TS2', CONFIG['SYSTEMS'][system].get('DEFAULT_REFLECTOR', 0)) logger.debug('(%s) Sending disconnected voice',system) _say = [words[_lang]['silence']] _say.append(words[_lang]['silence']) - if CONFIG['SYSTEMS'][system]['DEFAULT_REFLECTOR'] > 0: + if _default_reflector > 0: _say.append(words[_lang]['silence']) _say.append(words[_lang]['linkedto']) _say.append(words[_lang]['silence']) @@ -882,7 +1031,7 @@ def disconnectedVoice(system): _say.append(words[_lang]['silence']) _say.append(words[_lang]['silence']) - for number in str(CONFIG['SYSTEMS'][system]['DEFAULT_REFLECTOR']): + for number in str(_default_reflector): _say.append(words[_lang][number]) _say.append(words[_lang]['silence']) else: @@ -1105,8 +1254,6 @@ def options_config(): CONFIG['SYSTEMS'][_system]['_opt_key'] = False - if 'DIAL' in _options: - _options['DEFAULT_REFLECTOR'] = _options.pop('DIAL') if 'TIMER' in _options: _options['DEFAULT_UA_TIMER'] = _options.pop('TIMER') if 'TS1' in _options: @@ -1119,10 +1266,28 @@ def options_config(): _options['OVERRIDE_IDENT_TG'] = _options.pop('VOICETG') if 'IDENT' in _options: _options['VOICE'] = _options.pop('IDENT') + if 'DIALTG' in _options: + _options['DIAL_A_TG'] = _options.pop('DIALTG') + if 'DYNAMIC' in _options: + _options['DYNAMIC_TG_ROUTING'] = _options.pop('DYNAMIC') + if 'default_dial_ts1' in _options: + _options['DEFAULT_DIAL_TS1'] = _options.pop('default_dial_ts1') + if 'default_dial_ts2' in _options: + _options['DEFAULT_DIAL_TS2'] = _options.pop('default_dial_ts2') + if 'DIAL' in _options: + dial_value = _options.pop('DIAL') + if 'DEFAULT_DIAL_TS2' not in _options: + _options['DEFAULT_DIAL_TS2'] = dial_value + if 'DEFAULT_REFLECTOR' in _options: + default_reflector_value = _options.pop('DEFAULT_REFLECTOR') + if 'DEFAULT_DIAL_TS2' not in _options: + _options['DEFAULT_DIAL_TS2'] = default_reflector_value #DMR+ style options if 'StartRef' in _options: - _options['DEFAULT_REFLECTOR'] = _options.pop('StartRef') + start_ref_value = _options.pop('StartRef') + if 'DEFAULT_DIAL_TS2' not in _options: + _options['DEFAULT_DIAL_TS2'] = start_ref_value if 'RelinkTime' in _options: _options['DEFAULT_UA_TIMER'] = _options.pop('RelinkTime') if 'TS1_1' in _options: @@ -1171,19 +1336,30 @@ def options_config(): if 'TS2_STATIC' not in _options: _options['TS2_STATIC'] = False - if 'DEFAULT_REFLECTOR' not in _options: - _options['DEFAULT_REFLECTOR'] = 0 + if 'DEFAULT_DIAL_TS1' not in _options: + _options['DEFAULT_DIAL_TS1'] = CONFIG['SYSTEMS'][_system].get('DEFAULT_DIAL_TS1', 0) + if 'DEFAULT_DIAL_TS2' not in _options: + _options['DEFAULT_DIAL_TS2'] = CONFIG['SYSTEMS'][_system].get('DEFAULT_DIAL_TS2', CONFIG['SYSTEMS'][_system].get('DEFAULT_REFLECTOR', 0)) if 'OVERRIDE_IDENT_TG' not in _options: _options['OVERRIDE_IDENT_TG'] = False if 'DEFAULT_UA_TIMER' not in _options: _options['DEFAULT_UA_TIMER'] = CONFIG['SYSTEMS'][_system]['DEFAULT_UA_TIMER'] + if 'DIAL_A_TG' not in _options: + _options['DIAL_A_TG'] = int(CONFIG['SYSTEMS'][_system].get('DIAL_A_TG', True)) + if 'DYNAMIC_TG_ROUTING' not in _options: + _options['DYNAMIC_TG_ROUTING'] = int(CONFIG['SYSTEMS'][_system].get('DYNAMIC_TG_ROUTING', True)) - _default_reflector = parse_default_reflector_option(_options['DEFAULT_REFLECTOR'], default=None) - if _default_reflector is None: - logger.debug('(OPTIONS) %s - DEFAULT_REFLECTOR is not an integer, ignoring',_system) - _default_reflector = CONFIG['SYSTEMS'][_system]['DEFAULT_REFLECTOR'] + _default_dial_ts1 = parse_default_reflector_option(_options['DEFAULT_DIAL_TS1'], default=None) + if _default_dial_ts1 is None: + logger.debug('(OPTIONS) %s - DEFAULT_DIAL_TS1 is not an integer, ignoring',_system) + _default_dial_ts1 = CONFIG['SYSTEMS'][_system].get('DEFAULT_DIAL_TS1', 0) + + _default_dial_ts2 = parse_default_reflector_option(_options['DEFAULT_DIAL_TS2'], default=None) + if _default_dial_ts2 is None: + logger.debug('(OPTIONS) %s - DEFAULT_DIAL_TS2 is not an integer, ignoring',_system) + _default_dial_ts2 = CONFIG['SYSTEMS'][_system].get('DEFAULT_DIAL_TS2', CONFIG['SYSTEMS'][_system].get('DEFAULT_REFLECTOR', 0)) _default_ua_timer = parse_int_option(_options['DEFAULT_UA_TIMER'], default=None) if _default_ua_timer is None: @@ -1218,8 +1394,18 @@ def options_config(): elif CONFIG['SYSTEMS'][_system]['SINGLE_MODE'] != _single_mode: CONFIG['SYSTEMS'][_system]['SINGLE_MODE'] = _single_mode logger.debug("(OPTIONS) %s - Setting SINGLE_MODE to %s",_system,CONFIG['SYSTEMS'][_system]['SINGLE_MODE']) + + _dial_a_tg = parse_bool_option(_options['DIAL_A_TG'], default=None) + if _dial_a_tg is None: + logger.debug('(OPTIONS) %s - DIAL_A_TG is not 0 or 1, ignoring',_system) + _dial_a_tg = CONFIG['SYSTEMS'][_system].get('DIAL_A_TG', True) + + _dynamic_tg_routing = parse_bool_option(_options['DYNAMIC_TG_ROUTING'], default=None) + if _dynamic_tg_routing is None: + logger.debug('(OPTIONS) %s - DYNAMIC_TG_ROUTING is not 0 or 1, ignoring',_system) + _dynamic_tg_routing = CONFIG['SYSTEMS'][_system].get('DYNAMIC_TG_ROUTING', True) - if 'TS1_STATIC' not in _options or 'TS2_STATIC' not in _options or 'DEFAULT_REFLECTOR' not in _options or 'DEFAULT_UA_TIMER' not in _options: + if 'TS1_STATIC' not in _options or 'TS2_STATIC' not in _options or 'DEFAULT_DIAL_TS1' not in _options or 'DEFAULT_DIAL_TS2' not in _options or 'DEFAULT_UA_TIMER' not in _options: logger.debug('(OPTIONS) %s - Required field missing, ignoring',_system) continue @@ -1240,19 +1426,34 @@ def options_config(): logger.debug('(OPTIONS) %s Updating DEFAULT_UA_TIMER for existing bridges.',_system) update_timeout(_system,_tmout) - if _default_reflector != CONFIG['SYSTEMS'][_system]['DEFAULT_REFLECTOR']: + for _default_slot, _default_key, _default_value in ( + (1, 'DEFAULT_DIAL_TS1', _default_dial_ts1), + (2, 'DEFAULT_DIAL_TS2', _default_dial_ts2), + ): + if _default_value != CONFIG['SYSTEMS'][_system].get(_default_key, 0): + if _default_value > 0 and not valid_dial_a_tg_reflector(_default_value): + logger.debug('(OPTIONS) %s %s is in prohibited list, ignoring change',_system,_default_key) + reset_all_reflector_system(_tmout,_system,_default_slot) + _default_value = 0 + elif _default_value > 0: + logger.debug('(OPTIONS) %s %s changed, updating',_system,_default_key) + reset_all_reflector_system(_tmout,_system,_default_slot) + make_default_reflector(_default_value,_tmout,_system,_default_slot) + else: + logger.debug('(OPTIONS) %s %s disabled, updating',_system,_default_key) + reset_all_reflector_system(_tmout,_system,_default_slot) + if _default_slot == 1: + _default_dial_ts1 = _default_value + else: + _default_dial_ts2 = _default_value - if _default_reflector > 0 and not valid_dial_a_tg_reflector(_default_reflector): - logger.debug('(OPTIONS) %s default dial-a-tg is in prohibited list, ignoring change',_system) - reset_all_reflector_system(_tmout,_system) - _default_reflector = 0 - elif _default_reflector > 0: - logger.debug('(OPTIONS) %s default dial-a-tg changed, updating',_system) - reset_all_reflector_system(_tmout,_system) - make_default_reflector(_default_reflector,_tmout,_system) - else: - logger.debug('(OPTIONS) %s default dial-a-tg disabled, updating',_system) + if _dial_a_tg != CONFIG['SYSTEMS'][_system].get('DIAL_A_TG', True): + logger.debug('(OPTIONS) %s DIAL_A_TG changed to %s',_system,_dial_a_tg) + if not _dial_a_tg: reset_all_reflector_system(_tmout,_system) + + if _dynamic_tg_routing != CONFIG['SYSTEMS'][_system].get('DYNAMIC_TG_ROUTING', True): + logger.debug('(OPTIONS) %s DYNAMIC_TG_ROUTING changed to %s',_system,_dynamic_tg_routing) ts1 = [] if ('_reloadoptions' in CONFIG['SYSTEMS'][_system] and CONFIG['SYSTEMS'][_system]['_reloadoptions']) or (_options['TS1_STATIC'] != CONFIG['SYSTEMS'][_system]['TS1_STATIC']): @@ -1273,8 +1474,12 @@ def options_config(): CONFIG['SYSTEMS'][_system]['TS1_STATIC'] = _options['TS1_STATIC'] CONFIG['SYSTEMS'][_system]['TS2_STATIC'] = _options['TS2_STATIC'] - CONFIG['SYSTEMS'][_system]['DEFAULT_REFLECTOR'] = _default_reflector + CONFIG['SYSTEMS'][_system]['DEFAULT_DIAL_TS1'] = _default_dial_ts1 + CONFIG['SYSTEMS'][_system]['DEFAULT_DIAL_TS2'] = _default_dial_ts2 + CONFIG['SYSTEMS'][_system]['DEFAULT_REFLECTOR'] = _default_dial_ts2 CONFIG['SYSTEMS'][_system]['DEFAULT_UA_TIMER'] = _default_ua_timer + CONFIG['SYSTEMS'][_system]['DIAL_A_TG'] = _dial_a_tg + CONFIG['SYSTEMS'][_system]['DYNAMIC_TG_ROUTING'] = _dynamic_tg_routing if '_reloadoptions' in CONFIG['SYSTEMS'][_system] and CONFIG['SYSTEMS'][_system]['_reloadoptions']: CONFIG['SYSTEMS'][_system]['_reloadoptions'] = False @@ -1357,9 +1562,9 @@ class routerOBP(OPENBRIDGE): logger.exception('(to_target) caught exception') _target_status[_stream_id]['LAST'] = pkt_time return - _target_status[_stream_id]['H_LC'] = bptc.encode_header_lc(dst_lc) - _target_status[_stream_id]['T_LC'] = bptc.encode_terminator_lc(dst_lc) - _target_status[_stream_id]['EMB_LC'] = bptc.encode_emblc(dst_lc) + _target_status[_stream_id]['H_LC'] = dmr_codec.encode_header_lc(dst_lc) + _target_status[_stream_id]['T_LC'] = dmr_codec.encode_terminator_lc(dst_lc) + _target_status[_stream_id]['EMB_LC'] = dmr_codec.encode_emblc(dst_lc) logger.debug('(%s) Conference Bridge: %s, Call Bridged to OBP System: %s TS: %s, TGID: %s', self._system, _bridge, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) if CONFIG['REPORTS']['REPORT']: @@ -1396,7 +1601,7 @@ class routerOBP(OPENBRIDGE): if not _target_status[_stream_id].get('DATA_STREAM'): _target_status[_stream_id]['_fin'] = True # Create a Burst B-E packet (Embedded LC) - elif _frame_type == HBPF_VOICE and _dtype_vseq in [1,2,3,4]: + elif target_requires_emb_lc_rewrite(_dst_id, _target['TGID']) and _frame_type == HBPF_VOICE and _dtype_vseq in [1,2,3,4]: try: dmrbits = dmrbits[0:116] + _target_status[_stream_id]['EMB_LC'][_dtype_vseq] + dmrbits[148:264] except KeyError: @@ -1449,9 +1654,9 @@ class routerOBP(OPENBRIDGE): _target_status[_target['TS']]['TX_DATA_STREAM'] = _data_control # Generate LCs (full and EMB) for the TX stream dst_lc = b''.join([self.STATUS[_stream_id]['LC'][0:3], _target['TGID'], _rf_src]) - _target_status[_target['TS']]['TX_H_LC'] = bptc.encode_header_lc(dst_lc) - _target_status[_target['TS']]['TX_T_LC'] = bptc.encode_terminator_lc(dst_lc) - _target_status[_target['TS']]['TX_EMB_LC'] = bptc.encode_emblc(dst_lc) + _target_status[_target['TS']]['TX_H_LC'] = dmr_codec.encode_header_lc(dst_lc) + _target_status[_target['TS']]['TX_T_LC'] = dmr_codec.encode_terminator_lc(dst_lc) + _target_status[_target['TS']]['TX_EMB_LC'] = dmr_codec.encode_emblc(dst_lc) logger.debug('(%s) Generating TX FULL and EMB LCs for HomeBrew destination: System: %s, TS: %s, TGID: %s', self._system, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) logger.debug('(%s) Conference Bridge: %s, Call Bridged to HBP System: %s TS: %s, TGID: %s', self._system, _bridge, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) if CONFIG['REPORTS']['REPORT']: @@ -1488,7 +1693,7 @@ class routerOBP(OPENBRIDGE): call_duration = pkt_time - _target_status[_target['TS']]['TX_START'] systems[_target['SYSTEM']]._report.send_bridgeEvent('GROUP VOICE,END,TX,{},{},{},{},{},{},{:.2f}'.format(_target['SYSTEM'], int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _target['TS'], int_id(_target['TGID']), call_duration).encode(encoding='utf-8', errors='ignore')) # Create a Burst B-E packet (Embedded LC) - elif _frame_type == HBPF_VOICE and _dtype_vseq in [1,2,3,4]: + elif target_requires_emb_lc_rewrite(_dst_id, _target['TGID']) and _frame_type == HBPF_VOICE and _dtype_vseq in [1,2,3,4]: dmrbits = dmrbits[0:116] + _target_status[_target['TS']]['TX_EMB_LC'][_dtype_vseq] + dmrbits[148:264] dmrpkt = dmrbits.tobytes() #_tmp_data = b''.join([_tmp_data, dmrpkt, b'\x00\x00']) # Add two bytes of nothing since OBP doesn't include BER & RSSI bytes #_data[53:55] @@ -1744,13 +1949,13 @@ class routerOBP(OPENBRIDGE): # If we can, use the LC from the voice header as to keep all options intact if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD: - decoded = decode.voice_head_term(dmrpkt) + decoded = dmr_codec.voice_head_term(dmrpkt) self.STATUS[_stream_id]['LC'] = decoded['LC'] # If we don't have a voice header then don't wait to decode the Embedded LC # just make a new one from the HBP header. This is good enough, and it saves lots of time else: - self.STATUS[_stream_id]['LC'] = b''.join([LC_OPT,_dst_id,_rf_src]) + self.STATUS[_stream_id]['LC'] = dmr_codec.build_group_voice_lc(_dst_id, _rf_src) _inthops = 0 if _hops: @@ -1870,6 +2075,7 @@ class routerOBP(OPENBRIDGE): self.STATUS[_stream_id]['crcs'].add(_pkt_crc) + log_embedded_lc_observations(self._system, self.STATUS[_stream_id], _rf_src, _dst_id, _slot, _stream_id, _frame_type, _dtype_vseq, dmrpkt) self.STATUS[_stream_id]['LAST'] = pkt_time @@ -2052,9 +2258,9 @@ class routerHBP(HBSYSTEM): } # Generate LCs (full and EMB) for the TX stream dst_lc = b''.join([self.STATUS[_slot]['RX_LC'][0:3], _target['TGID'], _rf_src]) - _target_status[_stream_id]['H_LC'] = bptc.encode_header_lc(dst_lc) - _target_status[_stream_id]['T_LC'] = bptc.encode_terminator_lc(dst_lc) - _target_status[_stream_id]['EMB_LC'] = bptc.encode_emblc(dst_lc) + _target_status[_stream_id]['H_LC'] = dmr_codec.encode_header_lc(dst_lc) + _target_status[_stream_id]['T_LC'] = dmr_codec.encode_terminator_lc(dst_lc) + _target_status[_stream_id]['EMB_LC'] = dmr_codec.encode_emblc(dst_lc) logger.debug('(%s) Conference Bridge: %s, Call Bridged to OBP System: %s TS: %s, TGID: %s', self._system, _bridge, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) if CONFIG['REPORTS']['REPORT']: @@ -2088,7 +2294,7 @@ class routerHBP(HBSYSTEM): if not _target_status[_stream_id].get('DATA_STREAM'): _target_status[_stream_id]['_fin'] = True # Create a Burst B-E packet (Embedded LC) - elif _frame_type == HBPF_VOICE and _dtype_vseq in [1,2,3,4]: + elif target_requires_emb_lc_rewrite(_dst_id, _target['TGID']) and _frame_type == HBPF_VOICE and _dtype_vseq in [1,2,3,4]: dmrbits = dmrbits[0:116] + _target_status[_stream_id]['EMB_LC'][_dtype_vseq] + dmrbits[148:264] dmrpkt = dmrbits.tobytes() _tmp_data = b''.join([_tmp_data, dmrpkt]) @@ -2133,9 +2339,9 @@ class routerHBP(HBSYSTEM): _target_status[_target['TS']]['TX_DATA_STREAM'] = _data_control # Generate LCs (full and EMB) for the TX stream dst_lc = b''.join([self.STATUS[_slot]['RX_LC'][0:3],_target['TGID'],_rf_src]) - _target_status[_target['TS']]['TX_H_LC'] = bptc.encode_header_lc(dst_lc) - _target_status[_target['TS']]['TX_T_LC'] = bptc.encode_terminator_lc(dst_lc) - _target_status[_target['TS']]['TX_EMB_LC'] = bptc.encode_emblc(dst_lc) + _target_status[_target['TS']]['TX_H_LC'] = dmr_codec.encode_header_lc(dst_lc) + _target_status[_target['TS']]['TX_T_LC'] = dmr_codec.encode_terminator_lc(dst_lc) + _target_status[_target['TS']]['TX_EMB_LC'] = dmr_codec.encode_emblc(dst_lc) logger.debug('(%s) Generating TX FULL and EMB LCs for HomeBrew destination: System: %s, TS: %s, TGID: %s', self._system, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) logger.debug('(%s) Conference Bridge: %s, Call Bridged to HBP System: %s TS: %s, TGID: %s', self._system, _bridge, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) if CONFIG['REPORTS']['REPORT']: @@ -2172,7 +2378,7 @@ class routerHBP(HBSYSTEM): call_duration = pkt_time - _target_status[_target['TS']]['TX_START'] systems[_target['SYSTEM']]._report.send_bridgeEvent('GROUP VOICE,END,TX,{},{},{},{},{},{},{:.2f}'.format(_target['SYSTEM'], int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _target['TS'], int_id(_target['TGID']), call_duration).encode(encoding='utf-8', errors='ignore')) # Create a Burst B-E packet (Embedded LC) - elif _frame_type == HBPF_VOICE and _dtype_vseq in [1,2,3,4]: + elif target_requires_emb_lc_rewrite(_dst_id, _target['TGID']) and _frame_type == HBPF_VOICE and _dtype_vseq in [1,2,3,4]: dmrbits = dmrbits[0:116] + _target_status[_target['TS']]['TX_EMB_LC'][_dtype_vseq] + dmrbits[148:264] try: dmrpkt = dmrbits.tobytes() @@ -2438,8 +2644,9 @@ class routerHBP(HBSYSTEM): #Handle private voice calls (for dial-a-tg) elif _call_type == 'unit' and not _data_call and not self.STATUS[_slot]['_allStarMode']: - # Dial-A-TG private calls on either RF slot control the TS2 reflector state. - _dial_tg_slot = 2 + # Dial-A-TG private calls control the reflector state for the RF slot they arrive on. + _dial_tg_slot = _slot + _dial_tg_enabled = CONFIG['SYSTEMS'][self._system].get('DIAL_A_TG', True) _dial_tg_max = DIAL_A_TG_MAX _dial_tg_in_range = _int_dst_id >= 5 and _int_dst_id <= _dial_tg_max _dial_tg_out_of_range = _int_dst_id > _dial_tg_max @@ -2459,11 +2666,11 @@ class routerHBP(HBSYSTEM): self.STATUS[_slot]['_stopTgAnnounce'] = False logger.info('(%s) Dial-A-TG: Private call from %s to %s',self._system, int_id(_rf_src), _int_dst_id) - if _dial_tg_in_range and _int_dst_id != 8 and _int_dst_id != 9: + if _dial_tg_enabled and _dial_tg_in_range and _int_dst_id != 8 and _int_dst_id != 9: _bridgename = ''.join(['#',str(_int_dst_id)]) if _dial_tg_can_link and _bridgename not in BRIDGES: logger.info('(%s) [A] Dial-A-TG for TG %s does not exist. Creating as User Activated. Timeout: %s',self._system, _int_dst_id,CONFIG['SYSTEMS'][self._system]['DEFAULT_UA_TIMER']) - make_single_reflector(_dst_id,CONFIG['SYSTEMS'][self._system]['DEFAULT_UA_TIMER'],self._system) + make_single_reflector(_dst_id,CONFIG['SYSTEMS'][self._system]['DEFAULT_UA_TIMER'],self._system,_dial_tg_slot) if _dial_tg_can_link or _dial_tg_disconnect: for _bridge in BRIDGES: @@ -2523,14 +2730,20 @@ class routerHBP(HBSYSTEM): _say = [words[_lang]['silence']] - if _int_dst_id < 8 or _int_dst_id == 9 : + if not _dial_tg_enabled: + logger.info('(%s) Dial-A-TG: voice called - dial-a-tg disabled - "busy"', self._system) + _say.append(words[_lang]['busy']) + _say.append(words[_lang]['silence']) + self.STATUS[_slot]['_stopTgAnnounce'] = True + + elif _int_dst_id < 8 or _int_dst_id == 9 : logger.info('(%s) Dial-A-TG: voice called - TG < 8 or 9 - "busy""', self._system) _say.append(words[_lang]['busy']) _say.append(words[_lang]['silence']) self.STATUS[_slot]['_stopTgAnnounce'] = True #Allstar mode switch - if CONFIG['ALLSTAR']['ENABLED'] and _int_dst_id == 8: + elif CONFIG['ALLSTAR']['ENABLED'] and _int_dst_id == 8: logger.info('(%s) Dial-A-TG: voice called - TG 8 AllStar"', self._system) _say.append(words[_lang]['all-star-link-mode']) _say.append(words[_lang]['silence']) @@ -2546,7 +2759,7 @@ class routerHBP(HBSYSTEM): #If disconnection called - if _int_dst_id == 4000: + elif _int_dst_id == 4000: logger.info('(%s) Dial-A-TG: voice called - 4000 "not linked"', self._system) _say.append(words[_lang]['notlinked']) _say.append(words[_lang]['silence']) @@ -2688,16 +2901,16 @@ class routerHBP(HBSYSTEM): # If we can, use the LC from the voice header as to keep all options intact if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD: - decoded = decode.voice_head_term(dmrpkt) + decoded = dmr_codec.voice_head_term(dmrpkt) self.STATUS[_slot]['RX_LC'] = decoded['LC'] # If we don't have a voice header then don't wait to decode it from the Embedded LC # just make a new one from the HBP header. This is good enough, and it saves lots of time else: - self.STATUS[_slot]['RX_LC'] = b''.join([LC_OPT,_dst_id,_rf_src]) + self.STATUS[_slot]['RX_LC'] = dmr_codec.build_group_voice_lc(_dst_id, _rf_src) #Create default bridge for unknown TG - if int_id(_dst_id) >= 5 and int_id(_dst_id) != 9 and int_id(_dst_id) != 4000 and int_id(_dst_id) != 5000 and (str(int_id(_dst_id)) not in BRIDGES): + if CONFIG['SYSTEMS'][self._system].get('DYNAMIC_TG_ROUTING', True) and int_id(_dst_id) >= 5 and int_id(_dst_id) != 9 and int_id(_dst_id) != 4000 and int_id(_dst_id) != 5000 and (str(int_id(_dst_id)) not in BRIDGES): logger.info('(%s) Bridge for TG %s does not exist. Creating as User Activated. Timeout %s',self._system, int_id(_dst_id),CONFIG['SYSTEMS'][self._system]['DEFAULT_UA_TIMER']) make_single_bridge(_dst_id,self._system,_slot,CONFIG['SYSTEMS'][self._system]['DEFAULT_UA_TIMER']) @@ -2773,6 +2986,7 @@ class routerHBP(HBSYSTEM): self.STATUS[_slot]['lastSeq'] = _seq #Save this packet self.STATUS[_slot]['lastData'] = _data + log_embedded_lc_observations(self._system, self.STATUS[_slot], _rf_src, _dst_id, _slot, _stream_id, _frame_type, _dtype_vseq, dmrpkt) _sysIgnore = deque() for _bridge in BRIDGES: diff --git a/config.py b/config.py index 892b277..552ef22 100755 --- a/config.py +++ b/config.py @@ -322,9 +322,13 @@ def build_config(_config_file): 'DEFAULT_UA_TIMER': config.getint(section, 'DEFAULT_UA_TIMER', fallback=10), 'SINGLE_MODE': config.getboolean(section, 'SINGLE_MODE', fallback=True), 'VOICE_IDENT': config.getboolean(section, 'VOICE_IDENT', fallback=True), + 'DIAL_A_TG': config.getboolean(section, 'DIAL_A_TG', fallback=True), + 'DYNAMIC_TG_ROUTING': config.getboolean(section, 'DYNAMIC_TG_ROUTING', fallback=True), 'TS1_STATIC': config.get(section,'TS1_STATIC', fallback=''), 'TS2_STATIC': config.get(section,'TS2_STATIC', fallback=''), - 'DEFAULT_REFLECTOR': config.getint(section, 'DEFAULT_REFLECTOR'), + 'DEFAULT_DIAL_TS1': config.getint(section, 'DEFAULT_DIAL_TS1', fallback=0), + 'DEFAULT_DIAL_TS2': config.getint(section, 'DEFAULT_DIAL_TS2', fallback=config.getint(section, 'DEFAULT_REFLECTOR', fallback=0)), + 'DEFAULT_REFLECTOR': config.getint(section, 'DEFAULT_DIAL_TS2', fallback=config.getint(section, 'DEFAULT_REFLECTOR', fallback=0)), 'GENERATOR': config.getint(section, 'GENERATOR', fallback=100), 'ANNOUNCEMENT_LANGUAGE': config.get(section, 'ANNOUNCEMENT_LANGUAGE', fallback='en_GB'), 'ALLOW_UNREG_ID': config.getboolean(section,'ALLOW_UNREG_ID', fallback=False), @@ -400,7 +404,7 @@ if __name__ == '__main__': import os import argparse from pprint import pprint - from dmr_utils3.utils import int_id + from utils import int_id # Change the current directory to the location of the application os.chdir(os.path.dirname(os.path.realpath(sys.argv[0]))) diff --git a/const.py b/const.py index 630b76f..2a84705 100755 --- a/const.py +++ b/const.py @@ -39,8 +39,11 @@ ID_MAX = 16776415 # Timers STREAM_TO = .360 -# Options from the LC - used for late entry -LC_OPT = b'\x00\x00\x20' +# Legacy HBLink synthetic group voice LC prefix. Active FreeDMR late-entry LC +# generation uses freedmr_dmr_codec.build_group_voice_lc() with normal service +# options; keep this alias for older/lab code that still imports const.*. +HBLINK_LEGACY_GROUP_VOICE_LC_OPT = b'\x00\x00\x20' +LC_OPT = HBLINK_LEGACY_GROUP_VOICE_LC_OPT # HomeBrew Protocol Frame Types HBPF_VOICE = 0x0 diff --git a/docker-configs/docker-compose_install.sh b/docker-configs/docker-compose_install.sh index 8d22a75..7b74b39 100644 --- a/docker-configs/docker-compose_install.sh +++ b/docker-configs/docker-compose_install.sh @@ -134,6 +134,8 @@ SINGLE_MODE: True VOICE_IDENT: True TS1_STATIC: TS2_STATIC: +DEFAULT_DIAL_TS1: 0 +DEFAULT_DIAL_TS2: 0 DEFAULT_REFLECTOR: 0 ANNOUNCEMENT_LANGUAGE: en_GB GENERATOR: 100 diff --git a/docker-configs/docker_install.sh b/docker-configs/docker_install.sh index 99c245c..eecaa21 100755 --- a/docker-configs/docker_install.sh +++ b/docker-configs/docker_install.sh @@ -119,6 +119,8 @@ SINGLE_MODE: True VOICE_IDENT: True TS1_STATIC: TS2_STATIC: +DEFAULT_DIAL_TS1: 0 +DEFAULT_DIAL_TS2: 0 DEFAULT_REFLECTOR: 0 ANNOUNCEMENT_LANGUAGE: en_GB_2 GENERATOR: 100 @@ -183,4 +185,3 @@ chmod 700 ./update_freedmr.sh echo FreeDMR setup complete! - diff --git a/docker-configs/freedmr.cfg b/docker-configs/freedmr.cfg index 50d2194..4b92631 100644 --- a/docker-configs/freedmr.cfg +++ b/docker-configs/freedmr.cfg @@ -60,8 +60,12 @@ TGID_TS2_ACL: PERMIT:ALL DEFAULT_UA_TIMER: 10 SINGLE_MODE: True VOICE_IDENT: True +DIAL_A_TG: True +DYNAMIC_TG_ROUTING: True TS1_STATIC: TS2_STATIC: +DEFAULT_DIAL_TS1: 0 +DEFAULT_DIAL_TS2: 0 DEFAULT_REFLECTOR: 0 ANNOUNCEMENT_LANGUAGE: en_GB GENERATOR: 100 diff --git a/freedmr.cfg b/freedmr.cfg index 79f0702..792bc58 100755 --- a/freedmr.cfg +++ b/freedmr.cfg @@ -34,8 +34,12 @@ TGID_TS2_ACL: PERMIT:ALL DEFAULT_UA_TIMER: 60 SINGLE_MODE: True VOICE_IDENT: True +DIAL_A_TG: True +DYNAMIC_TG_ROUTING: True TS1_STATIC: TS2_STATIC: +DEFAULT_DIAL_TS1: 0 +DEFAULT_DIAL_TS2: 0 DEFAULT_REFLECTOR: 0 ANNOUNCEMENT_LANGUAGE: en_GB GENERATOR: 0 @@ -75,4 +79,3 @@ SUB_ACL: DENY:1 TGID_TS1_ACL: PERMIT:ALL TGID_TS2_ACL: PERMIT:ALL ANNOUNCEMENT_LANGUAGE: en_GB - diff --git a/freedmr_dmr_codec.py b/freedmr_dmr_codec.py new file mode 100644 index 0000000..e1265e2 --- /dev/null +++ b/freedmr_dmr_codec.py @@ -0,0 +1,648 @@ +#!/usr/bin/env python +# +# Embedded LC codec helpers for FreeDMR. +# +# The embedded LC BPTC layout follows MMDVMHost CDMREmbeddedData, +# CHamming::encode16114/decode16114, and CCRC::encodeFiveBit. +# MMDVMHost is Copyright (C) Jonathan Naylor G4KLX and licensed GPLv2+ +# (or later). FreeDMR is GPLv3+, so this port is used under GPLv3+. + +from __future__ import annotations + +from dataclasses import dataclass + +from bitarray import bitarray + + +FULL_LC_BITS = 196 +FULL_LC_BYTES = 9 +FULL_LC_PARITY_BYTES = 3 +FULL_LC_CODE_BYTES = 12 +FULL_LC_INFO_BITS = 196 +EMBEDDED_LC_FRAGMENT_BITS = 32 +EMBEDDED_LC_RAW_BITS = 128 +EMBEDDED_LC_PAYLOAD_BITS = 72 +EMBEDDED_LC_PAYLOAD_BYTES = 9 +SLOT_TYPE_BITS = 20 +SLOT_TYPE_DATA_BITS = 8 +LC_FLCO_GROUP_VOICE = 0x00 +LC_FLCO_UNIT_VOICE = 0x03 +LC_FID_ETSI = 0x00 +LC_SERVICE_OPTIONS_NORMAL = 0x00 +LC_SERVICE_OPTIONS_OVCM = 0x04 +LC_SERVICE_OPTIONS_HBLINK_LEGACY = 0x20 +GROUP_VOICE_LC_OPT = b'\x00\x00\x00' +UNIT_VOICE_LC_OPT = b'\x03\x00\x00' +BS_VOICE_SYNC = bitarray('011101010101111111010111110111110111010111110111') +BS_DATA_SYNC = bitarray('110111111111010101111101011101011101111101011101') +EMB = { + 'BURST_B': bitarray('0001001110010001'), + 'BURST_C': bitarray('0001011101110100'), + 'BURST_D': bitarray('0001011101110100'), + 'BURST_E': bitarray('0001010100000111'), + 'BURST_F': bitarray('0001000111100010'), +} +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'), +} + +FULL_LC_INTERLEAVE_19696 = ( + 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, +) + +FULL_LC_PAYLOAD_INDEXES = ( + 136, 121, 106, 91, 76, 61, 46, 31, + 152, 137, 122, 107, 92, 77, 62, 47, 32, 17, 2, + 123, 108, 93, 78, 63, 48, 33, 18, 3, 184, 169, + 94, 79, 64, 49, 34, 19, 4, 185, 170, 155, 140, + 65, 50, 35, 20, 5, 186, 171, 156, 141, 126, 111, + 36, 21, 6, 187, 172, 157, 142, 127, 112, 97, 82, + 7, 188, 173, 158, 143, 128, 113, 98, 83, +) + +RS129_HEADER_MASK = (0x96, 0x96, 0x96) +RS129_TERMINATOR_MASK = (0x99, 0x99, 0x99) +RS129_POLY = (64, 56, 14, 1) + +GOLAY_2087_PARITY = ( + 0x0000, 0xB08E, 0xE093, 0x501D, 0x70A9, 0xC027, 0x903A, 0x20B4, + 0x60DC, 0xD052, 0x804F, 0x30C1, 0x1075, 0xA0FB, 0xF0E6, 0x4068, + 0x7036, 0xC0B8, 0x90A5, 0x202B, 0x009F, 0xB011, 0xE00C, 0x5082, + 0x10EA, 0xA064, 0xF079, 0x40F7, 0x6043, 0xD0CD, 0x80D0, 0x305E, + 0xD06C, 0x60E2, 0x30FF, 0x8071, 0xA0C5, 0x104B, 0x4056, 0xF0D8, + 0xB0B0, 0x003E, 0x5023, 0xE0AD, 0xC019, 0x7097, 0x208A, 0x9004, + 0xA05A, 0x10D4, 0x40C9, 0xF047, 0xD0F3, 0x607D, 0x3060, 0x80EE, + 0xC086, 0x7008, 0x2015, 0x909B, 0xB02F, 0x00A1, 0x50BC, 0xE032, + 0x90D9, 0x2057, 0x704A, 0xC0C4, 0xE070, 0x50FE, 0x00E3, 0xB06D, + 0xF005, 0x408B, 0x1096, 0xA018, 0x80AC, 0x3022, 0x603F, 0xD0B1, + 0xE0EF, 0x5061, 0x007C, 0xB0F2, 0x9046, 0x20C8, 0x70D5, 0xC05B, + 0x8033, 0x30BD, 0x60A0, 0xD02E, 0xF09A, 0x4014, 0x1009, 0xA087, + 0x40B5, 0xF03B, 0xA026, 0x10A8, 0x301C, 0x8092, 0xD08F, 0x6001, + 0x2069, 0x90E7, 0xC0FA, 0x7074, 0x50C0, 0xE04E, 0xB053, 0x00DD, + 0x3083, 0x800D, 0xD010, 0x609E, 0x402A, 0xF0A4, 0xA0B9, 0x1037, + 0x505F, 0xE0D1, 0xB0CC, 0x0042, 0x20F6, 0x9078, 0xC065, 0x70EB, + 0xA03D, 0x10B3, 0x40AE, 0xF020, 0xD094, 0x601A, 0x3007, 0x8089, + 0xC0E1, 0x706F, 0x2072, 0x90FC, 0xB048, 0x00C6, 0x50DB, 0xE055, + 0xD00B, 0x6085, 0x3098, 0x8016, 0xA0A2, 0x102C, 0x4031, 0xF0BF, + 0xB0D7, 0x0059, 0x5044, 0xE0CA, 0xC07E, 0x70F0, 0x20ED, 0x9063, + 0x7051, 0xC0DF, 0x90C2, 0x204C, 0x00F8, 0xB076, 0xE06B, 0x50E5, + 0x108D, 0xA003, 0xF01E, 0x4090, 0x6024, 0xD0AA, 0x80B7, 0x3039, + 0x0067, 0xB0E9, 0xE0F4, 0x507A, 0x70CE, 0xC040, 0x905D, 0x20D3, + 0x60BB, 0xD035, 0x8028, 0x30A6, 0x1012, 0xA09C, 0xF081, 0x400F, + 0x30E4, 0x806A, 0xD077, 0x60F9, 0x404D, 0xF0C3, 0xA0DE, 0x1050, + 0x5038, 0xE0B6, 0xB0AB, 0x0025, 0x2091, 0x901F, 0xC002, 0x708C, + 0x40D2, 0xF05C, 0xA041, 0x10CF, 0x307B, 0x80F5, 0xD0E8, 0x6066, + 0x200E, 0x9080, 0xC09D, 0x7013, 0x50A7, 0xE029, 0xB034, 0x00BA, + 0xE088, 0x5006, 0x001B, 0xB095, 0x9021, 0x20AF, 0x70B2, 0xC03C, + 0x8054, 0x30DA, 0x60C7, 0xD049, 0xF0FD, 0x4073, 0x106E, 0xA0E0, + 0x90BE, 0x2030, 0x702D, 0xC0A3, 0xE017, 0x5099, 0x0084, 0xB00A, + 0xF062, 0x40EC, 0x10F1, 0xA07F, 0x80CB, 0x3045, 0x6058, 0xD0D6, +) + +SLOT_TYPE_NAMES = { + 0x0: "PI_HEAD", + 0x1: "VOICE_LC_HEAD", + 0x2: "VOICE_LC_TERM", + 0x3: "CSBK", + 0x4: "MBC_HEAD", + 0x5: "MBC_CONT", + 0x6: "DATA_HEAD", + 0x7: "1/2_RATE", + 0x8: "3/4_RATE", + 0x9: "IDLE", + 0xA: "1/1_RATE", + 0xB: "RES_1", + 0xC: "RES_2", + 0xD: "RES_3", + 0xE: "RES_4", + 0xF: "RES_5", +} +EMBEDDED_LC_PAYLOAD_RANGES = ( + (0, 11), + (16, 27), + (32, 42), + (48, 58), + (64, 74), + (80, 90), + (96, 106), +) +EMBEDDED_LC_CRC_POSITIONS = (42, 58, 74, 90, 106) + + +@dataclass(frozen=True) +class EmbeddedLC: + data: bytes + flco: int + raw: bitarray + corrected: int = 0 + + +@dataclass(frozen=True) +class FullLC: + data: bytes + flco: int + source_id: int + target_id: int + service_options: int + is_group_call: bool + is_unit_call: bool + + +def build_group_voice_lc( + dst_id: bytes, + src_id: bytes, + service_options: int = LC_SERVICE_OPTIONS_NORMAL, +) -> bytes: + if len(dst_id) != 3 or len(src_id) != 3: + raise FullLCError("DMR LC target and source IDs must be three bytes") + if service_options < 0 or service_options > 0xFF: + raise FullLCError("DMR LC service options must fit in one byte") + + return bytes([LC_FLCO_GROUP_VOICE, LC_FID_ETSI, service_options]) + dst_id + src_id + + +@dataclass(frozen=True) +class SlotType: + color_code: int + data_type: int + name: str + corrected: int = 0 + + +class EmbeddedLCError(ValueError): + pass + + +class FullLCError(ValueError): + pass + + +class SlotTypeError(ValueError): + pass + + +def bytes_to_bits(data: bytes) -> bitarray: + bits = bitarray(endian="big") + bits.frombytes(data) + return bits + + +def bits_to_int(bits: bitarray) -> int: + value = 0 + for bit in bits: + value = (value << 1) | int(bit) + return value + + +def _bits_from_int(value: int, length: int) -> bitarray: + return bitarray((bool(value & (1 << bit)) for bit in range(length - 1, -1, -1)), endian="big") + + +def _hamming_distance(a: int, b: int) -> int: + return (a ^ b).bit_count() + + +def _gf256_mul(left: int, right: int) -> int: + result = 0 + while right: + if right & 0x01: + result ^= left + right >>= 1 + left <<= 1 + if left & 0x100: + left ^= 0x11D + return result & 0xFF + + +def encode_hamming_15113(data: bitarray) -> bitarray: + if len(data) != 11: + raise FullLCError("Hamming(15,11,3) input must be 11 bits") + return bitarray(( + data[0] ^ data[1] ^ data[2] ^ data[3] ^ data[5] ^ data[7] ^ data[8], + data[1] ^ data[2] ^ data[3] ^ data[4] ^ data[6] ^ data[8] ^ data[9], + data[2] ^ data[3] ^ data[4] ^ data[5] ^ data[7] ^ data[9] ^ data[10], + data[0] ^ data[1] ^ data[2] ^ data[4] ^ data[6] ^ data[7] ^ data[10], + ), endian="big") + + +def encode_hamming_1393(data: bitarray) -> bitarray: + if len(data) != 9: + raise FullLCError("Hamming(13,9,3) input must be 9 bits") + return bitarray(( + data[0] ^ data[1] ^ data[3] ^ data[5] ^ data[6], + data[0] ^ data[1] ^ data[2] ^ data[4] ^ data[6] ^ data[7], + data[0] ^ data[1] ^ data[2] ^ data[3] ^ data[5] ^ data[7] ^ data[8], + data[0] ^ data[2] ^ data[4] ^ data[5] ^ data[8], + ), endian="big") + + +def rs129_parity(data: bytes) -> bytes: + if len(data) != FULL_LC_BYTES: + raise FullLCError("RS(12,9) LC payload must be 9 bytes") + + parity = [0x00, 0x00, 0x00] + for byte in data: + dbyte = byte ^ parity[2] + parity[2] = parity[1] ^ _gf256_mul(RS129_POLY[2], dbyte) + parity[1] = parity[0] ^ _gf256_mul(RS129_POLY[1], dbyte) + parity[0] = _gf256_mul(RS129_POLY[0], dbyte) + return bytes((parity[2], parity[1], parity[0])) + + +def encode_full_lc_parity(data: bytes, terminator: bool = False) -> bytes: + mask = RS129_TERMINATOR_MASK if terminator else RS129_HEADER_MASK + return bytes(value ^ mask[index] for index, value in enumerate(rs129_parity(data))) + + +def _encode_19696(data: bytes) -> bitarray: + if len(data) != FULL_LC_CODE_BYTES: + raise FullLCError("BPTC(196,96) input must be 12 bytes") + + bits = bytes_to_bits(data) + for _index in range(4): + bits.insert(0, 0) + + for index in range(9): + start = (index * 15) + 1 + end = start + 11 + parity = encode_hamming_15113(bits[start:end]) + for pbit in range(4): + bits.insert(end + pbit, parity[pbit]) + + for _index in range(60): + bits.append(0) + + column = bitarray(9, endian="big") + for col in range(15): + start = col + 1 + for index in range(9): + column[index] = bits[start] + start += 15 + parity = encode_hamming_1393(column) + + target = 136 + col + for pbit in range(4): + bits[target] = parity[pbit] + target += 15 + + return bits + + +def interleave_19696(data: bitarray) -> bitarray: + if len(data) != FULL_LC_BITS: + raise FullLCError("BPTC(196,96) data must be 196 bits") + interleaved = bitarray(FULL_LC_BITS, endian="big") + for index in range(FULL_LC_BITS): + interleaved[FULL_LC_INTERLEAVE_19696[index]] = data[index] + return interleaved + + +def encode_full_lc(data: bytes, terminator: bool = False) -> bitarray: + if len(data) != FULL_LC_BYTES: + raise FullLCError("full LC payload must be 9 bytes") + code = data + encode_full_lc_parity(data, terminator=terminator) + return interleave_19696(_encode_19696(code)) + + +def encode_header_lc(data: bytes) -> bitarray: + return encode_full_lc(data, terminator=False) + + +def encode_terminator_lc(data: bytes) -> bitarray: + return encode_full_lc(data, terminator=True) + + +def decode_full_lc(bits: bitarray) -> FullLC: + if len(bits) != FULL_LC_INFO_BITS: + raise FullLCError("full LC data must be 196 bits") + payload = bitarray((bits[index] for index in FULL_LC_PAYLOAD_INDEXES), endian="big") + data = payload.tobytes() + flco = data[0] & 0x3F + return FullLC( + data=data, + flco=flco, + source_id=int.from_bytes(data[6:9], "big"), + target_id=int.from_bytes(data[3:6], "big"), + service_options=data[2], + is_group_call=flco == LC_FLCO_GROUP_VOICE, + is_unit_call=flco == LC_FLCO_UNIT_VOICE, + ) + + +def _padded_bits_to_bytes(bits: bitarray) -> bytes: + padded = bits.copy() + add_bits = 8 - (len(padded) % 8) + if add_bits < 8: + for _bit in range(add_bits): + padded.insert(0, 0) + return padded.tobytes() + + +def voice_head_term(payload: bytes) -> dict[str, object]: + bits = bytes_to_bits(payload) + info = bits[0:98] + bits[166:264] + slot_type = bits[98:108] + bits[156:166] + sync = bits[108:156] + lc = decode_full_lc(info).data + return { + "LC": lc, + "CC": _padded_bits_to_bytes(slot_type[0:4]), + "DTYPE": _padded_bits_to_bytes(slot_type[4:8]), + "SYNC": sync, + } + + +def voice_sync(payload: bytes) -> dict[str, object]: + bits = bytes_to_bits(payload) + return { + "AMBE": [ + bits[0:72], + bits[72:108] + bits[156:192], + bits[192:264], + ], + "SYNC": bits[108:156], + } + + +def voice(payload: bytes) -> dict[str, object]: + bits = bytes_to_bits(payload) + emb = bits[108:116] + bits[148:156] + return { + "AMBE": [ + bits[0:72], + bits[72:108] + bits[156:192], + bits[192:264], + ], + "CC": _padded_bits_to_bytes(emb[0:4]), + "LCSS": _padded_bits_to_bytes(emb[5:7]), + "EMBED": bits[116:148], + } + + +def encode_slot_type(color_code: int, data_type: int) -> bitarray: + if not 0 <= color_code <= 0x0F: + raise SlotTypeError("slot color code must fit in four bits") + if not 0 <= data_type <= 0x0F: + raise SlotTypeError("slot data type must fit in four bits") + + value = (color_code << 4) | data_type + checksum = GOLAY_2087_PARITY[value] + code = (value << 12) | ((checksum & 0xFF) << 4) | (checksum >> 12) + return _bits_from_int(code, SLOT_TYPE_BITS) + + +def decode_slot_type(bits: bitarray) -> SlotType: + if len(bits) != SLOT_TYPE_BITS: + raise SlotTypeError("slot type must be 20 bits") + + observed = bits_to_int(bits) + candidates = [] + for value in range(256): + encoded = bits_to_int(encode_slot_type(value >> 4, value & 0x0F)) + candidates.append((_hamming_distance(observed, encoded), value)) + candidates.sort() + + distance, value = candidates[0] + if len(candidates) > 1 and candidates[1][0] == distance: + raise SlotTypeError("ambiguous Golay(20,8,7) slot type") + if distance > 3: + raise SlotTypeError("slot type Golay(20,8,7) check failed") + + data_type = value & 0x0F + return SlotType( + color_code=value >> 4, + data_type=data_type, + name=SLOT_TYPE_NAMES[data_type], + corrected=distance, + ) + + +def crc5(data: bytes | bitarray) -> int: + bits = bytes_to_bits(data) if isinstance(data, bytes) else data + if len(bits) != EMBEDDED_LC_PAYLOAD_BITS: + raise EmbeddedLCError("embedded LC payload must be 72 bits") + + total = 0 + for offset in range(0, EMBEDDED_LC_PAYLOAD_BITS, 8): + total += bits_to_int(bits[offset:offset + 8]) + return total % 31 + + +def encode_hamming_16114(row: bitarray) -> None: + if len(row) != 16: + raise EmbeddedLCError("Hamming row must be 16 bits") + + row[11] = row[0] ^ row[1] ^ row[2] ^ row[3] ^ row[5] ^ row[7] ^ row[8] + row[12] = row[1] ^ row[2] ^ row[3] ^ row[4] ^ row[6] ^ row[8] ^ row[9] + row[13] = row[2] ^ row[3] ^ row[4] ^ row[5] ^ row[7] ^ row[9] ^ row[10] + row[14] = row[0] ^ row[1] ^ row[2] ^ row[4] ^ row[6] ^ row[7] ^ row[10] + row[15] = row[0] ^ row[2] ^ row[5] ^ row[6] ^ row[8] ^ row[9] ^ row[10] + + +def decode_hamming_16114(row: bitarray) -> bool: + if len(row) != 16: + raise EmbeddedLCError("Hamming row must be 16 bits") + + c0 = row[0] ^ row[1] ^ row[2] ^ row[3] ^ row[5] ^ row[7] ^ row[8] + c1 = row[1] ^ row[2] ^ row[3] ^ row[4] ^ row[6] ^ row[8] ^ row[9] + c2 = row[2] ^ row[3] ^ row[4] ^ row[5] ^ row[7] ^ row[9] ^ row[10] + c3 = row[0] ^ row[1] ^ row[2] ^ row[4] ^ row[6] ^ row[7] ^ row[10] + c4 = row[0] ^ row[2] ^ row[5] ^ row[6] ^ row[8] ^ row[9] ^ row[10] + + syndrome = 0 + syndrome |= 0x01 if c0 != row[11] else 0 + syndrome |= 0x02 if c1 != row[12] else 0 + syndrome |= 0x04 if c2 != row[13] else 0 + syndrome |= 0x08 if c3 != row[14] else 0 + syndrome |= 0x10 if c4 != row[15] else 0 + + corrections = { + 0x01: 11, + 0x02: 12, + 0x04: 13, + 0x08: 14, + 0x10: 15, + 0x19: 0, + 0x0B: 1, + 0x1F: 2, + 0x07: 3, + 0x0E: 4, + 0x15: 5, + 0x1A: 6, + 0x0D: 7, + 0x13: 8, + 0x16: 9, + 0x1C: 10, + } + + if syndrome == 0: + return True + if syndrome not in corrections: + return False + row[corrections[syndrome]] = not row[corrections[syndrome]] + return True + + +def encode_embedded_lc(lc: bytes | bitarray) -> tuple[bitarray, bitarray, bitarray, bitarray]: + payload = bytes_to_bits(lc) if isinstance(lc, bytes) else lc.copy() + if len(payload) != EMBEDDED_LC_PAYLOAD_BITS: + raise EmbeddedLCError("embedded LC must be 9 bytes / 72 bits") + + data = bitarray([0] * EMBEDDED_LC_RAW_BITS, endian="big") + checksum = crc5(payload) + data[106] = bool(checksum & 0x01) + data[90] = bool(checksum & 0x02) + data[74] = bool(checksum & 0x04) + data[58] = bool(checksum & 0x08) + data[42] = bool(checksum & 0x10) + + position = 0 + for start, end in EMBEDDED_LC_PAYLOAD_RANGES: + data[start:end] = payload[position:position + (end - start)] + position += end - start + + for row_start in range(0, 112, 16): + row = data[row_start:row_start + 16] + encode_hamming_16114(row) + data[row_start:row_start + 16] = row + + for column in range(16): + data[column + 112] = ( + data[column + 0] + ^ data[column + 16] + ^ data[column + 32] + ^ data[column + 48] + ^ data[column + 64] + ^ data[column + 80] + ^ data[column + 96] + ) + + raw = bitarray([0] * EMBEDDED_LC_RAW_BITS, endian="big") + position = 0 + for index in range(EMBEDDED_LC_RAW_BITS): + raw[index] = data[position] + position += 16 + if position > 127: + position -= 127 + + return tuple( + raw[index:index + EMBEDDED_LC_FRAGMENT_BITS] + for index in range(0, EMBEDDED_LC_RAW_BITS, EMBEDDED_LC_FRAGMENT_BITS) + ) + + +def encode_emblc(lc: bytes | bitarray) -> dict[int, bitarray]: + fragments = encode_embedded_lc(lc) + return { + 1: fragments[0], + 2: fragments[1], + 3: fragments[2], + 4: fragments[3], + } + + +def decode_embedded_lc(fragments: tuple[bitarray, bitarray, bitarray, bitarray] | list[bitarray]) -> EmbeddedLC: + if len(fragments) != 4: + raise EmbeddedLCError("embedded LC requires four fragments") + + raw = bitarray(endian="big") + for fragment in fragments: + if len(fragment) != EMBEDDED_LC_FRAGMENT_BITS: + raise EmbeddedLCError("embedded LC fragments must be 32 bits") + raw += fragment + + data = bitarray([0] * EMBEDDED_LC_RAW_BITS, endian="big") + position = 0 + for index in range(EMBEDDED_LC_RAW_BITS): + data[position] = raw[index] + position += 16 + if position > 127: + position -= 127 + + corrected = 0 + for row_start in range(0, 112, 16): + row = data[row_start:row_start + 16] + before = row.copy() + if not decode_hamming_16114(row): + raise EmbeddedLCError("embedded LC Hamming check failed") + if row != before: + corrected += 1 + data[row_start:row_start + 16] = row + + for column in range(16): + parity = ( + data[column + 0] + ^ data[column + 16] + ^ data[column + 32] + ^ data[column + 48] + ^ data[column + 64] + ^ data[column + 80] + ^ data[column + 96] + ^ data[column + 112] + ) + if parity: + raise EmbeddedLCError("embedded LC column parity check failed") + + payload = bitarray(endian="big") + for start, end in EMBEDDED_LC_PAYLOAD_RANGES: + payload += data[start:end] + + checksum = 0 + if data[42]: + checksum += 16 + if data[58]: + checksum += 8 + if data[74]: + checksum += 4 + if data[90]: + checksum += 2 + if data[106]: + checksum += 1 + + if crc5(payload) != checksum: + raise EmbeddedLCError("embedded LC 5-bit checksum failed") + + decoded = payload.tobytes() + return EmbeddedLC(data=decoded, flco=decoded[0] & 0x3F, raw=raw, corrected=corrected) + + +def embedded_lc_fragment_from_payload(payload: bytes) -> bitarray: + if len(payload) != 33: + raise EmbeddedLCError("DMR payload must be 33 bytes") + bits = bytes_to_bits(payload) + return bits[116:148] + + +def payload_with_embedded_lc_fragment(payload: bytes, fragment: bitarray) -> bytes: + if len(payload) != 33: + raise EmbeddedLCError("DMR payload must be 33 bytes") + if len(fragment) != EMBEDDED_LC_FRAGMENT_BITS: + raise EmbeddedLCError("embedded LC fragment must be 32 bits") + bits = bytes_to_bits(payload) + bits = bits[:116] + fragment + bits[148:264] + return bits.tobytes() diff --git a/hblink.py b/hblink.py index 3b3a51f..a4a4e8b 100755 --- a/hblink.py +++ b/hblink.py @@ -45,8 +45,7 @@ from twisted.internet import reactor, task import log import config from const import * -from utils import mk_id_dict, try_download,load_json,blake2bsum -from dmr_utils3.utils import int_id, bytes_4 +from utils import bytes_4, int_id, mk_id_dict, try_download,load_json,blake2bsum # Imports for the reporting server import pickle diff --git a/hdstack/hotspot_proxy_v2.py b/hdstack/hotspot_proxy_v2.py index 1e3948f..7f0f8df 100644 --- a/hdstack/hotspot_proxy_v2.py +++ b/hdstack/hotspot_proxy_v2.py @@ -2,7 +2,7 @@ from twisted.internet.protocol import DatagramProtocol from twisted.internet import reactor, task from time import time from resettabletimer import ResettableTimer -from dmr_utils3.utils import int_id +from utils import int_id import random class Proxy(DatagramProtocol): diff --git a/hotspot_proxy_v2.py b/hotspot_proxy_v2.py index 6d1e754..c72e076 100644 --- a/hotspot_proxy_v2.py +++ b/hotspot_proxy_v2.py @@ -19,7 +19,7 @@ from twisted.internet.protocol import DatagramProtocol from twisted.internet import reactor, task from time import time -from dmr_utils3.utils import int_id +from utils import int_id import random import ipaddress import os diff --git a/loro.cfg b/loro.cfg index 7779866..62acb82 100644 --- a/loro.cfg +++ b/loro.cfg @@ -182,10 +182,11 @@ SINGLE_MODE: True VOICE_IDENT: False TS1_STATIC: TS2_STATIC: +DEFAULT_DIAL_TS1: 0 +DEFAULT_DIAL_TS2: 0 DEFAULT_REFLECTOR: 0 GENERATOR: 1 ANNOUNCEMENT_LANGUAGE:es_ES ALLOW_UNREG_ID: True PROXY_CONTROL: False OVERRIDE_IDENT_TG: - diff --git a/mk_voice.py b/mk_voice.py index f4f4b27..e9881ba 100755 --- a/mk_voice.py +++ b/mk_voice.py @@ -19,9 +19,9 @@ ############################################################################### from bitarray import bitarray -from dmr_utils3 import bptc, golay, qr -from dmr_utils3.utils import bytes_3, bytes_4 -from dmr_utils3.const import EMB, SLOT_TYPE, BS_VOICE_SYNC, BS_DATA_SYNC, LC_OPT +import freedmr_dmr_codec as dmr_codec +from freedmr_dmr_codec import EMB, SLOT_TYPE, BS_VOICE_SYNC, BS_DATA_SYNC, GROUP_VOICE_LC_OPT +from utils import bytes_3, bytes_4 from random import randint from voice_lib import words @@ -44,15 +44,15 @@ def pkt_gen(_rf_src, _dst_id, _peer, _slot, _phrase): # Calculate all of the static components up-front STREAM_ID = bytes_4(randint(0x00, 0xFFFFFFFF)) SDP = _rf_src + _dst_id + _peer - LC = LC_OPT + _dst_id + _rf_src + LC = GROUP_VOICE_LC_OPT + _dst_id + _rf_src - HEAD_LC = bptc.encode_header_lc(LC) + HEAD_LC = dmr_codec.encode_header_lc(LC) HEAD_LC = [HEAD_LC[:98], HEAD_LC[-98:]] - TERM_LC = bptc.encode_terminator_lc(LC) + TERM_LC = dmr_codec.encode_terminator_lc(LC) TERM_LC = [TERM_LC[:98], TERM_LC[-98:]] - EMB_LC = bptc.encode_emblc(LC) + EMB_LC = dmr_codec.encode_emblc(LC) EMBED = [] EMBED.append( BS_VOICE_SYNC ) diff --git a/playback_file.cfg b/playback_file.cfg index 89f269d..3bc1f99 100644 --- a/playback_file.cfg +++ b/playback_file.cfg @@ -191,6 +191,8 @@ SINGLE_MODE: True VOICE_IDENT: False TS1_STATIC: TS2_STATIC: +DEFAULT_DIAL_TS1: 0 +DEFAULT_DIAL_TS2: 0 DEFAULT_REFLECTOR: 0 GENERATOR: 1 ANNOUNCEMENT_LANGUAGE:es_ES @@ -230,4 +232,3 @@ SUB_ACL: DENY:1 TGID_TS1_ACL: PERMIT:ALL TGID_TS2_ACL: PERMIT:ALL ANNOUNCEMENT_LANGUAGE: en_GB - diff --git a/tests/harness/deterministic.py b/tests/harness/deterministic.py index bbf737f..953f4bd 100644 --- a/tests/harness/deterministic.py +++ b/tests/harness/deterministic.py @@ -304,8 +304,12 @@ def minimal_config(system_names: tuple[str, ...] = ("MASTER-A", "MASTER-B")) -> "DEFAULT_UA_TIMER": 1, "SINGLE_MODE": True, "VOICE_IDENT": False, + "DIAL_A_TG": True, + "DYNAMIC_TG_ROUTING": True, "TS1_STATIC": "", "TS2_STATIC": "", + "DEFAULT_DIAL_TS1": 0, + "DEFAULT_DIAL_TS2": 0, "DEFAULT_REFLECTOR": 0, "GENERATOR": 0, "ANNOUNCEMENT_LANGUAGE": "en_GB", diff --git a/tests/harness/udp_blackbox.py b/tests/harness/udp_blackbox.py index fd0e2d9..7c74b12 100644 --- a/tests/harness/udp_blackbox.py +++ b/tests/harness/udp_blackbox.py @@ -47,7 +47,7 @@ BCST = b"BCST" BCVE = b"BCVE" FBP_VERSION = 5 FBP_PASSPHRASE = b"test-passphrase".ljust(20, b"\x00")[:20] -REQUIRED_RUNTIME_MODULES = ("bitarray", "twisted", "dmr_utils3", "setproctitle") +REQUIRED_RUNTIME_MODULES = ("bitarray", "twisted", "setproctitle") @dataclass(frozen=True) @@ -467,8 +467,12 @@ def write_bridge_master_config( global_tg2_acl: str = "PERMIT:ALL", ts1_static: str = "", ts2_static: str = "91", + dial_a_tg: bool = True, + dynamic_tg_routing: bool = True, ) -> 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" systems = [] for name, port in system_ports.items(): systems.append( @@ -490,8 +494,12 @@ TGID_TS2_ACL: PERMIT:ALL DEFAULT_UA_TIMER: 1 SINGLE_MODE: True VOICE_IDENT: False +DIAL_A_TG: {dial_a_tg_text} +DYNAMIC_TG_ROUTING: {dynamic_tg_routing_text} TS1_STATIC: {ts1_static} TS2_STATIC: {ts2_static} +DEFAULT_DIAL_TS1: 0 +DEFAULT_DIAL_TS2: 0 DEFAULT_REFLECTOR: 0 ANNOUNCEMENT_LANGUAGE: en_GB GENERATOR: 0 @@ -640,9 +648,16 @@ class HbpRepeater: except socket.timeout: return captures - def login(self) -> None: - self.send(RPTL + self.radio_id) - challenge = self.recv() + def login(self, startup_timeout: float = 8.0) -> None: + deadline = time.monotonic() + startup_timeout + while True: + self.send(RPTL + self.radio_id) + try: + challenge = self.recv(timeout=min(self.timeout, max(0.1, deadline - time.monotonic()))) + break + except (TimeoutError, socket.timeout): + if time.monotonic() >= deadline: + raise if not challenge.packet.startswith(RPTACK): raise AssertionError(f"expected RPTACK challenge, got {challenge.packet!r}") @@ -955,6 +970,8 @@ class UdpBlackBoxScenario: global_tg2_acl: str = "PERMIT:ALL", ts1_static: str = "", ts2_static: str = "91", + dial_a_tg: bool = True, + dynamic_tg_routing: bool = True, fbp_systems: dict[str, int] | None = None, fbp_proto_versions: dict[str, int] | None = None, ) -> None: @@ -975,6 +992,8 @@ class UdpBlackBoxScenario: "global_tg2_acl": global_tg2_acl, "ts1_static": ts1_static, "ts2_static": ts2_static, + "dial_a_tg": dial_a_tg, + "dynamic_tg_routing": dynamic_tg_routing, } def __enter__(self): diff --git a/tests/test_api.py b/tests/test_api.py index e96d611..c0921d9 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,25 +1,8 @@ import io import json -import sys -import types import unittest -def install_dmr_utils_stub(): - if "dmr_utils3.utils" in sys.modules: - return None - dmr_utils3 = types.ModuleType("dmr_utils3") - utils = types.ModuleType("dmr_utils3.utils") - - def bytes_4(value): - return int(value).to_bytes(4, "big") - - utils.bytes_4 = bytes_4 - sys.modules["dmr_utils3"] = dmr_utils3 - sys.modules["dmr_utils3.utils"] = utils - return ("dmr_utils3", "dmr_utils3.utils") - - class FakeRequest: def __init__(self, path, payload=None): self.postpath = [part.encode("utf-8") for part in path.strip("/").split("/") if part] @@ -47,7 +30,6 @@ class APITest(unittest.TestCase): import twisted.web.resource # noqa: F401 except ModuleNotFoundError as exc: self.skipTest(f"Twisted is not installed: {exc}") - self.stubbed_modules = install_dmr_utils_stub() import API self.api = API @@ -69,11 +51,6 @@ class APITest(unittest.TestCase): self.bridges = {} self.controller = API.FD_APIController(self.config, self.bridges) - def tearDown(self): - if self.stubbed_modules: - for module in self.stubbed_modules: - sys.modules.pop(module, None) - def test_getoptions_returns_clear_no_options_response(self): result = self.controller.getoptions("MASTER-A") diff --git a/tests/test_auxiliary_tools.py b/tests/test_auxiliary_tools.py index 5fb48d4..210e2fa 100644 --- a/tests/test_auxiliary_tools.py +++ b/tests/test_auxiliary_tools.py @@ -108,19 +108,13 @@ class AuxiliaryToolTests(unittest.TestCase): sys.modules["mysql.connector"] = connector_module def _install_proxy_stubs(self): - stubbed = ["dmr_utils3", "dmr_utils3.utils", "Pyro5", "Pyro5.api"] + stubbed = ["Pyro5", "Pyro5.api"] saved_modules = {name: sys.modules.get(name) for name in stubbed + ["hotspot_proxy_v2"]} - dmr_utils3_module = types.ModuleType("dmr_utils3") - dmr_utils3_utils_module = types.ModuleType("dmr_utils3.utils") - dmr_utils3_utils_module.int_id = lambda value: int.from_bytes(value, "big") - dmr_utils3_module.utils = dmr_utils3_utils_module pyro5_module = types.ModuleType("Pyro5") pyro5_api_module = types.ModuleType("Pyro5.api") pyro5_api_module.Proxy = object pyro5_module.api = pyro5_api_module - sys.modules["dmr_utils3"] = dmr_utils3_module - sys.modules["dmr_utils3.utils"] = dmr_utils3_utils_module sys.modules["Pyro5"] = pyro5_module sys.modules["Pyro5.api"] = pyro5_api_module sys.modules.pop("hotspot_proxy_v2", None) diff --git a/tests/test_deterministic_harness.py b/tests/test_deterministic_harness.py index b45b2e8..aa84a61 100644 --- a/tests/test_deterministic_harness.py +++ b/tests/test_deterministic_harness.py @@ -5,6 +5,7 @@ import tempfile import unittest import config as freedmr_config +from freedmr_dmr_codec import encode_embedded_lc, payload_with_embedded_lc_fragment from tests.harness.deterministic import ( HBPF_DATA_SYNC, @@ -21,6 +22,14 @@ from tests.harness.deterministic import ( ) +def embedded_lc_payloads(lc): + payload = b"\x55" * 33 + return tuple( + payload_with_embedded_lc_fragment(payload, fragment) + for fragment in encode_embedded_lc(lc) + ) + + class PacketSpecTest(unittest.TestCase): def test_packet_spec_builds_parseable_dmrd_payload(self): packet = PacketSpec( @@ -44,6 +53,9 @@ class PacketSpecTest(unittest.TestCase): class DeterministicHarnessTest(unittest.TestCase): + TA_EMB_LC_PAYLOADS = embedded_lc_payloads(bytes.fromhex("04004c43414c4c3132")) + GPS_EMB_LC_PAYLOADS = embedded_lc_payloads(bytes.fromhex("080007fcfae048b57b")) + def test_config_global_use_acl_false_is_boolean_and_stale_time_is_seconds(self): config_text = """ [GLOBAL] @@ -69,6 +81,64 @@ DEFAULT_REFLECTOR: 0 self.assertIs(parsed["GLOBAL"]["USE_ACL"], False) 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"]) + self.assertEqual(parsed["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS1"], 0) + self.assertEqual(parsed["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"], 0) + + def test_config_deprecated_default_reflector_populates_ts2_default_dial(self): + config_text = """ +[GLOBAL] +USE_ACL: False + +[ALIASES] +TRY_DOWNLOAD: False +STALE_DAYS: 1 + +[MASTER-A] +MODE: MASTER +ENABLED: True +DEFAULT_REFLECTOR: 91 +""" + fd, path = tempfile.mkstemp(suffix=".cfg") + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(config_text) + + parsed = freedmr_config.build_config(path) + finally: + os.unlink(path) + + 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) + + def test_config_default_dial_ts2_takes_precedence_over_deprecated_default_reflector(self): + config_text = """ +[GLOBAL] +USE_ACL: False + +[ALIASES] +TRY_DOWNLOAD: False +STALE_DAYS: 1 + +[MASTER-A] +MODE: MASTER +ENABLED: True +DEFAULT_REFLECTOR: 91 +DEFAULT_DIAL_TS2: 92 +""" + fd, path = tempfile.mkstemp(suffix=".cfg") + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(config_text) + + parsed = freedmr_config.build_config(path) + finally: + os.unlink(path) + + self.assertEqual(parsed["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"], 92) + self.assertEqual(parsed["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"], 92) def test_set_alias_updates_bridge_master_globals_and_shared_hblink_config(self): with DeterministicScenario() as scenario: @@ -140,13 +210,13 @@ DEFAULT_REFLECTOR: 0 dtype_vseq=HBPF_SLT_VTERM, ) - def reflector_bridge(self, bridge_name, active=True): + def reflector_bridge(self, bridge_name, active=True, slot=2): tg = int(bridge_name[1:]) return { bridge_name: [ { "SYSTEM": "MASTER-A", - "TS": 2, + "TS": slot, "TGID": bytes_3(9), "ACTIVE": active, "TIMEOUT": 60, @@ -227,6 +297,16 @@ DEFAULT_REFLECTOR: 0 self.assertEqual(captured[0].fields["stream_id"], bytes_4(0x01020304)) self.assertEqual(scenario.capture.for_system("MASTER-A"), []) + def test_dynamic_tg_routing_disabled_does_not_create_unknown_conventional_bridge(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + config["SYSTEMS"]["MASTER-A"]["DYNAMIC_TG_ROUTING"] = False + + with DeterministicScenario(config=config) as scenario: + scenario.inject_hbp("MASTER-A", PacketSpec(dst_id=12345, slot=2)) + + self.assertNotIn("12345", scenario.bridge_state) + self.assertEqual(scenario.capture.packets, []) + def test_hbp_group_packet_rewrites_only_slot_for_cross_slot_route(self): bridges = active_bridge( "91", @@ -329,11 +409,12 @@ DEFAULT_REFLECTOR: 0 scenario.bm.options_config() master_entry = next( - entry for entry in scenario.bridge_state["#91"] if entry["SYSTEM"] == "MASTER-A" + entry for entry in scenario.bridge_state["#91"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 2 ) self.assertTrue(master_entry["ACTIVE"]) self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"], 91) + self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"], 91) self.assertEqual(config["SYSTEMS"]["MASTER-A"]["OVERRIDE_IDENT_TG"], 9) def test_options_invalid_voice_does_not_block_valid_static_tg(self): @@ -378,8 +459,19 @@ DEFAULT_REFLECTOR: 0 self.assertFalse(config["SYSTEMS"]["MASTER-A"]["VOICE_IDENT"]) self.assertTrue(config["SYSTEMS"]["MASTER-A"]["SINGLE_MODE"]) + def test_options_can_disable_dial_a_tg_and_dynamic_tg_routing(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + config["SYSTEMS"]["MASTER-A"]["OPTIONS"] = "DIALTG=0;DYNAMIC=0;DIAL=0;TIMER=1" + + with DeterministicScenario(config=config) as scenario: + scenario.bm.options_config() + + self.assertFalse(config["SYSTEMS"]["MASTER-A"]["DIAL_A_TG"]) + self.assertFalse(config["SYSTEMS"]["MASTER-A"]["DYNAMIC_TG_ROUTING"]) + def test_options_empty_dial_disables_default_reflector(self): config = minimal_config(("MASTER-A", "MASTER-B")) + config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"] = 91 config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"] = 91 config["SYSTEMS"]["MASTER-A"]["OPTIONS"] = "DIAL=;TIMER=1" @@ -388,11 +480,64 @@ DEFAULT_REFLECTOR: 0 scenario.bm.options_config() master_entry = next( - entry for entry in scenario.bridge_state["#91"] if entry["SYSTEM"] == "MASTER-A" + entry for entry in scenario.bridge_state["#91"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 2 ) self.assertFalse(master_entry["ACTIVE"]) self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"], 0) + self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"], 0) + + def test_options_default_dial_ts1_creates_ts1_default(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + config["SYSTEMS"]["MASTER-A"]["OPTIONS"] = "default_dial_ts1=91;TIMER=1" + + with DeterministicScenario(config=config) as scenario: + scenario.bm.options_config() + + ts1_entry = next( + entry for entry in scenario.bridge_state["#91"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 1 + ) + ts2_entry = next( + entry for entry in scenario.bridge_state["#91"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 2 + ) + + self.assertTrue(ts1_entry["ACTIVE"]) + self.assertFalse(ts2_entry["ACTIVE"]) + self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS1"], 91) + self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"], 0) + self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"], 0) + + def test_options_default_dial_ts2_creates_ts2_default(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + config["SYSTEMS"]["MASTER-A"]["OPTIONS"] = "default_dial_ts2=92;TIMER=1" + + with DeterministicScenario(config=config) as scenario: + scenario.bm.options_config() + + ts2_entry = next( + entry for entry in scenario.bridge_state["#92"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 2 + ) + + self.assertTrue(ts2_entry["ACTIVE"]) + self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"], 92) + self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"], 92) + + def test_options_default_dial_ts2_takes_precedence_over_deprecated_aliases(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + config["SYSTEMS"]["MASTER-A"]["OPTIONS"] = ( + "DIAL=91;DEFAULT_REFLECTOR=92;StartRef=93;DEFAULT_DIAL_TS2=94;TIMER=1" + ) + + with DeterministicScenario(config=config) as scenario: + scenario.bm.options_config() + + ts2_entry = next( + entry for entry in scenario.bridge_state["#94"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 2 + ) + + self.assertTrue(ts2_entry["ACTIVE"]) + self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"], 94) + self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"], 94) def test_options_invalid_timer_does_not_block_valid_static_tgs(self): config = minimal_config(("MASTER-A", "MASTER-B")) @@ -507,54 +652,114 @@ DEFAULT_REFLECTOR: 0 self.assertEqual(captured[0].fields["stream_id"], bytes_4(stream_id)) self.assertEqual(captured[1].fields["stream_id"], bytes_4(stream_id + 1)) - def test_dial_a_tg_private_call_on_slot_1_activates_ts2_reflector(self): + def test_dial_a_tg_private_call_on_slot_1_activates_ts1_reflector(self): with DeterministicScenario() as scenario: scenario.inject_hbp("MASTER-A", self.private_call(235, slot=1)) bridge = scenario.bridge_state["#235"] - source_entry = [entry for entry in bridge if entry["SYSTEM"] == "MASTER-A"][0] + source_entry = [ + entry for entry in bridge if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 1 + ][0] - self.assertEqual(source_entry["TS"], 2) + self.assertEqual(source_entry["TS"], 1) self.assertEqual(source_entry["TGID"], bytes_3(9)) self.assertTrue(source_entry["ACTIVE"]) self.assertEqual(scenario.capture.packets, []) - def test_dial_a_tg_private_call_on_slot_1_retunes_active_ts2_reflector(self): + def test_dial_a_tg_private_call_on_slot_1_retunes_only_ts1_reflector(self): bridges = {} - bridges.update(self.reflector_bridge("#235", active=True)) + bridges.update(self.reflector_bridge("#235", active=True, slot=1)) with DeterministicScenario(bridges=bridges) as scenario: scenario.inject_hbp("MASTER-A", self.private_call(236, slot=1)) old_entry = [ - entry for entry in scenario.bridge_state["#235"] if entry["SYSTEM"] == "MASTER-A" + entry for entry in scenario.bridge_state["#235"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 1 ][0] new_entry = [ - entry for entry in scenario.bridge_state["#236"] if entry["SYSTEM"] == "MASTER-A" + entry for entry in scenario.bridge_state["#236"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 1 ][0] - self.assertEqual(old_entry["TS"], 2) + self.assertEqual(old_entry["TS"], 1) self.assertFalse(old_entry["ACTIVE"]) - self.assertEqual(new_entry["TS"], 2) + self.assertEqual(new_entry["TS"], 1) self.assertTrue(new_entry["ACTIVE"]) self.assertEqual(scenario.capture.packets, []) - def test_dial_a_tg_disconnect_on_slot_1_deactivates_active_ts2_reflector(self): + def test_dial_a_tg_disconnect_on_slot_1_deactivates_active_ts1_reflector(self): bridges = {} - bridges.update(self.reflector_bridge("#235", active=True)) + bridges.update(self.reflector_bridge("#235", active=True, slot=1)) with DeterministicScenario(bridges=bridges) as scenario: scenario.inject_hbp("MASTER-A", self.private_call(4000, slot=1)) source_entry = [ - entry for entry in scenario.bridge_state["#235"] if entry["SYSTEM"] == "MASTER-A" + entry for entry in scenario.bridge_state["#235"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 1 ][0] - self.assertEqual(source_entry["TS"], 2) + self.assertEqual(source_entry["TS"], 1) self.assertFalse(source_entry["ACTIVE"]) self.assertEqual(scenario.capture.packets, []) - def test_dial_a_tg_disconnect_is_scoped_to_receiving_master(self): + def test_dial_a_tg_private_call_on_slot_1_does_not_control_ts2_reflector(self): + bridges = {} + bridges.update(self.reflector_bridge("#235", active=True, slot=2)) + + with DeterministicScenario(bridges=bridges) as scenario: + scenario.inject_hbp("MASTER-A", self.private_call(236, slot=1)) + + ts2_entry = [ + entry for entry in scenario.bridge_state["#235"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 2 + ][0] + ts1_entry = [ + entry for entry in scenario.bridge_state["#236"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 1 + ][0] + + self.assertTrue(ts2_entry["ACTIVE"]) + self.assertTrue(ts1_entry["ACTIVE"]) + self.assertEqual(scenario.capture.packets, []) + + def test_dial_a_tg_private_call_on_slot_2_controls_ts2_reflector(self): + with DeterministicScenario() as scenario: + scenario.inject_hbp("MASTER-A", self.private_call(235, slot=2)) + + bridge = scenario.bridge_state["#235"] + source_entry = [ + entry for entry in bridge if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 2 + ][0] + other_slot_entry = [ + entry for entry in bridge if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 1 + ][0] + + self.assertTrue(source_entry["ACTIVE"]) + self.assertFalse(other_slot_entry["ACTIVE"]) + self.assertEqual(scenario.capture.packets, []) + + def test_dial_a_tg_disabled_rejects_private_control(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + config["SYSTEMS"]["MASTER-A"]["DIAL_A_TG"] = False + + with DeterministicScenario(config=config) as scenario: + spoken = [] + scenario.bm.words["en_GB"].update( + { + "busy": b"busy", + "linkedto": b"linkedto", + "silence": b"silence", + "to": b"to", + } + ) + scenario.bm.pkt_gen = lambda source_id, dst_id, peer_id, slot, say: spoken.append(list(say)) or iter(()) + scenario.inject_hbp("MASTER-A", self.private_call(235, slot=1)) + scenario.inject_hbp("MASTER-A", self.private_call_terminator(235, slot=1)) + + self.assertNotIn("#235", scenario.bridge_state) + self.assertTrue(spoken) + self.assertIn(b"busy", spoken[-1]) + self.assertNotIn(b"linkedto", spoken[-1]) + self.assertEqual(scenario.capture.packets, []) + + def test_dial_a_tg_disconnect_on_slot_1_does_not_deactivate_ts2_reflector(self): bridges = { "#235": [ { @@ -594,8 +799,8 @@ DEFAULT_REFLECTOR: 0 entry for entry in scenario.bridge_state["#235"] if entry["SYSTEM"] == "MASTER-B" ][0] - self.assertFalse(master_a_entry["ACTIVE"]) - self.assertEqual(master_a_entry["TIMER"], scenario.clock.now) + self.assertTrue(master_a_entry["ACTIVE"]) + self.assertEqual(master_a_entry["TIMER"], 111) self.assertTrue(master_b_entry["ACTIVE"]) self.assertEqual(master_b_entry["TIMER"], 222) self.assertEqual(scenario.capture.packets, []) @@ -752,10 +957,10 @@ DEFAULT_REFLECTOR: 0 scenario.inject_hbp("MASTER-A", self.private_call_terminator(9990, slot=1)) source_entry = [ - entry for entry in scenario.bridge_state["#9990"] if entry["SYSTEM"] == "MASTER-A" + entry for entry in scenario.bridge_state["#9990"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 1 ][0] - self.assertEqual(source_entry["TS"], 2) + self.assertEqual(source_entry["TS"], 1) self.assertEqual(source_entry["TGID"], bytes_3(9)) self.assertTrue(source_entry["ACTIVE"]) self.assertEqual(source_entry["TIMEOUT"], 60) @@ -781,10 +986,10 @@ DEFAULT_REFLECTOR: 0 scenario.inject_hbp("MASTER-A", self.private_call_terminator(999999, slot=1)) source_entry = [ - entry for entry in scenario.bridge_state["#999999"] if entry["SYSTEM"] == "MASTER-A" + entry for entry in scenario.bridge_state["#999999"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 1 ][0] - self.assertEqual(source_entry["TS"], 2) + self.assertEqual(source_entry["TS"], 1) self.assertTrue(source_entry["ACTIVE"]) self.assertTrue(spoken) self.assertIn(b"linkedto", spoken[-1]) @@ -795,6 +1000,7 @@ DEFAULT_REFLECTOR: 0 for default_reflector in (6, 7, 8): with self.subTest(default_reflector=default_reflector): config = minimal_config(("MASTER-A", "MASTER-B")) + config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"] = default_reflector config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"] = default_reflector with DeterministicScenario(config=config) as scenario: @@ -802,32 +1008,51 @@ DEFAULT_REFLECTOR: 0 scenario.bm.make_default_reflectors() self.assertNotIn(f"#{default_reflector}", scenario.bridge_state) + self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"], 0) self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"], 0) - def test_startup_default_reflector_accepts_linkable_target(self): + def test_startup_default_dial_ts1_accepts_linkable_target(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS1"] = 91 + + with DeterministicScenario(config=config) as scenario: + scenario.bm.make_default_reflectors() + + master_entry = next( + entry for entry in scenario.bridge_state["#91"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 1 + ) + + self.assertTrue(master_entry["ACTIVE"]) + self.assertEqual(master_entry["TS"], 1) + self.assertEqual(master_entry["ON"], [bytes_3(91)]) + + def test_startup_default_dial_ts2_accepts_linkable_target(self): config = minimal_config(("MASTER-A", "MASTER-B")) + config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"] = 91 config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"] = 91 with DeterministicScenario(config=config) as scenario: scenario.bm.make_default_reflectors() master_entry = next( - entry for entry in scenario.bridge_state["#91"] if entry["SYSTEM"] == "MASTER-A" + entry for entry in scenario.bridge_state["#91"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 2 ) self.assertTrue(master_entry["ACTIVE"]) self.assertEqual(master_entry["TS"], 2) self.assertEqual(master_entry["ON"], [bytes_3(91)]) + self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"], 91) - def test_startup_default_reflector_accepts_policy_max_999999(self): + def test_startup_default_dial_ts2_accepts_policy_max_999999(self): config = minimal_config(("MASTER-A", "MASTER-B")) + config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"] = 999999 config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"] = 999999 with DeterministicScenario(config=config) as scenario: scenario.bm.make_default_reflectors() master_entry = next( - entry for entry in scenario.bridge_state["#999999"] if entry["SYSTEM"] == "MASTER-A" + entry for entry in scenario.bridge_state["#999999"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 2 ) self.assertTrue(master_entry["ACTIVE"]) @@ -836,6 +1061,7 @@ DEFAULT_REFLECTOR: 0 def test_startup_default_reflector_rejects_above_policy_max(self): config = minimal_config(("MASTER-A", "MASTER-B")) + config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"] = 1000000 config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"] = 1000000 with DeterministicScenario(config=config) as scenario: @@ -843,17 +1069,19 @@ DEFAULT_REFLECTOR: 0 scenario.bm.make_default_reflectors() self.assertNotIn("#1000000", scenario.bridge_state) + self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"], 0) self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"], 0) def test_startup_default_reflector_logs_invalid_value(self): config = minimal_config(("MASTER-A", "MASTER-B")) + config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"] = 1000000 config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"] = 1000000 with DeterministicScenario(config=config) as scenario: with self.assertLogs(scenario.bm.logger, level="WARNING") as logs: scenario.bm.make_default_reflectors() - self.assertIn("MASTER-A default dial-a-tg 1000000 is invalid", "\n".join(logs.output)) + self.assertIn("MASTER-A DEFAULT_DIAL_TS2 1000000 is invalid", "\n".join(logs.output)) self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"], 0) def test_options_default_reflector_rejects_above_policy_max(self): @@ -868,23 +1096,25 @@ DEFAULT_REFLECTOR: 0 def test_options_invalid_default_reflector_disables_active_default(self): config = minimal_config(("MASTER-A", "MASTER-B")) + config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"] = 91 config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"] = 91 config["SYSTEMS"]["MASTER-A"]["OPTIONS"] = "DIAL=1000000;TIMER=1" with DeterministicScenario(config=config) as scenario: scenario.bm.make_default_reflectors() active_entry = next( - entry for entry in scenario.bridge_state["#91"] if entry["SYSTEM"] == "MASTER-A" + entry for entry in scenario.bridge_state["#91"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 2 ) self.assertTrue(active_entry["ACTIVE"]) scenario.bm.options_config() disabled_entry = next( - entry for entry in scenario.bridge_state["#91"] if entry["SYSTEM"] == "MASTER-A" + entry for entry in scenario.bridge_state["#91"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 2 ) self.assertFalse(disabled_entry["ACTIVE"]) + self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_DIAL_TS2"], 0) self.assertEqual(config["SYSTEMS"]["MASTER-A"]["DEFAULT_REFLECTOR"], 0) def test_dial_a_tg_above_policy_max_reports_busy_without_retune(self): @@ -924,11 +1154,11 @@ DEFAULT_REFLECTOR: 0 scenario.inject_hbp("MASTER-A", self.private_call(235, slot=1)) bridge = scenario.bridge_state["#235"] - source_entry = [entry for entry in bridge if entry["SYSTEM"] == "MASTER-A"][0] + source_entry = [entry for entry in bridge if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 1][0] fbp_entry = [entry for entry in bridge if entry["SYSTEM"] == "OBP-1"][0] self.assertTrue(source_entry["ACTIVE"]) - self.assertEqual(source_entry["TS"], 2) + self.assertEqual(source_entry["TS"], 1) self.assertEqual(source_entry["TGID"], bytes_3(9)) self.assertTrue(fbp_entry["ACTIVE"]) self.assertEqual(fbp_entry["TS"], 1) @@ -947,7 +1177,7 @@ DEFAULT_REFLECTOR: 0 scenario.inject_hbp( "MASTER-A", - PacketSpec(dst_id=9, slot=2, stream_id=stream_id), + PacketSpec(dst_id=9, slot=1, stream_id=stream_id), ) self.assertEqual(scenario.capture.for_system("OBP-1"), []) @@ -1089,7 +1319,7 @@ DEFAULT_REFLECTOR: 0 self.assertTrue(slot["TX_PROMPT_CANCEL"]) self.assertFalse(slot["TX_PROMPT_ACTIVE"]) - def test_real_hbp_voice_cancels_generated_prompt_and_keeps_late_entry_lc_rewrite(self): + def test_real_hbp_voice_cancels_generated_prompt_and_preserves_same_tg_emb_lc(self): config = minimal_config(("MASTER-A", "MASTER-B")) bridges = active_bridge( "91", @@ -1129,7 +1359,7 @@ DEFAULT_REFLECTOR: 0 self.assertEqual(captured[0].fields["dst_id"], bytes_3(9)) self.assertEqual(captured[1].fields["dst_id"], bytes_3(91)) self.assertEqual(captured[1].fields["dtype_vseq"], 1) - self.assertNotEqual( + self.assertEqual( captured[1].fields["dmr_payload"], real_packet.data()[20:53], ) @@ -1200,9 +1430,9 @@ DEFAULT_REFLECTOR: 0 self.assertNotIn("#235", scenario.bridge_state) self.assertEqual(scenario.capture.packets, []) - def test_dial_a_tg_status_on_slot_1_reports_active_ts2_reflector(self): + def test_dial_a_tg_status_on_slot_1_reports_active_ts1_reflector(self): bridges = {} - bridges.update(self.reflector_bridge("#235", active=True)) + bridges.update(self.reflector_bridge("#235", active=True, slot=1)) with DeterministicScenario(bridges=bridges) as scenario: spoken = [] @@ -1222,11 +1452,11 @@ DEFAULT_REFLECTOR: 0 scenario.inject_hbp("MASTER-A", self.private_call_terminator(5000, slot=1)) source_entry = [ - entry for entry in scenario.bridge_state["#235"] if entry["SYSTEM"] == "MASTER-A" + entry for entry in scenario.bridge_state["#235"] if entry["SYSTEM"] == "MASTER-A" and entry["TS"] == 1 ][0] self.assertTrue(source_entry["ACTIVE"]) - self.assertEqual(source_entry["TS"], 2) + self.assertEqual(source_entry["TS"], 1) self.assertTrue(spoken) self.assertIn(b"linkedto", spoken[-1]) self.assertIn(b"2", spoken[-1]) @@ -1237,8 +1467,8 @@ DEFAULT_REFLECTOR: 0 def test_dial_a_tg_status_reports_only_one_active_reflector(self): bridges = {} - bridges.update(self.reflector_bridge("#235", active=True)) - bridges.update(self.reflector_bridge("#236", active=True)) + bridges.update(self.reflector_bridge("#235", active=True, slot=1)) + bridges.update(self.reflector_bridge("#236", active=True, slot=1)) with DeterministicScenario(bridges=bridges) as scenario: spoken = [] @@ -2002,6 +2232,227 @@ DEFAULT_REFLECTOR: 0 self.assertEqual(len(captured), 1) self.assertEqual(captured[0].fields["dmr_payload"], payload) + def test_same_tg_voice_embedded_lc_payload_is_preserved(self): + cases = [] + + config = minimal_config(("MASTER-A", "MASTER-B")) + cases.append(( + "hbp-to-hbp", + config, + active_bridge("91", 91, (("MASTER-A", 2), ("MASTER-B", 2))), + "MASTER-A", + "MASTER-B", + "hbp", + 2, + )) + + config = minimal_config(("MASTER-A",)) + add_openbridge_system(config, "OBP-1", network_id=3001) + cases.append(( + "hbp-to-obp", + config, + active_bridge("91", 91, (("MASTER-A", 2), ("OBP-1", 1))), + "MASTER-A", + "OBP-1", + "hbp", + 2, + )) + + config = minimal_config(("MASTER-A",)) + add_openbridge_system(config, "OBP-1", network_id=3001) + cases.append(( + "obp-to-hbp", + config, + active_bridge("91", 91, (("OBP-1", 1), ("MASTER-A", 2))), + "OBP-1", + "MASTER-A", + "obp", + 1, + )) + + config = minimal_config(()) + add_openbridge_system(config, "OBP-1", network_id=3001) + add_openbridge_system(config, "OBP-2", network_id=3002) + cases.append(( + "obp-to-obp", + config, + active_bridge("91", 91, (("OBP-1", 1), ("OBP-2", 1))), + "OBP-1", + "OBP-2", + "obp", + 1, + )) + + payload = bytes((index * 7) % 256 for index in range(33)) + for name, config, bridges, source, target, transport, slot in cases: + with self.subTest(name=name): + with DeterministicScenario(config=config, bridges=bridges) as scenario: + packet = PacketSpec( + dst_id=91, + slot=slot, + frame_type=HBPF_VOICE, + dtype_vseq=1, + payload=payload, + ) + + if transport == "hbp": + scenario.inject_hbp(source, packet) + else: + scenario.inject_obp(source, packet) + + captured = scenario.capture.for_system(target) + self.assertEqual(len(captured), 1) + self.assertEqual(captured[0].fields["dmr_payload"], payload) + + def test_tg_mapped_voice_embedded_lc_payload_is_rewritten(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + bridges = active_bridge( + "91-to-235", + 91, + ( + ("MASTER-A", 2), + ("MASTER-B", 2), + ), + ) + bridges["91-to-235"][1]["TGID"] = bytes_3(235) + payload = bytes((index * 11) % 256 for index in range(33)) + + with DeterministicScenario(config=config, bridges=bridges) as scenario: + packet = PacketSpec( + dst_id=91, + slot=2, + frame_type=HBPF_VOICE, + dtype_vseq=1, + payload=payload, + ) + + scenario.inject_hbp("MASTER-A", packet) + + captured = scenario.capture.for_system("MASTER-B") + self.assertEqual(len(captured), 1) + self.assertEqual(captured[0].fields["dst_id"], bytes_3(235)) + self.assertNotEqual(captured[0].fields["dmr_payload"], payload) + + def test_hbp_late_entry_synthetic_lc_uses_normal_group_voice_options(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + bridges = active_bridge("91", 91, (("MASTER-A", 2), ("MASTER-B", 2))) + stream_id = 0x01020304 + + with DeterministicScenario(config=config, bridges=bridges) as scenario: + scenario.inject_hbp( + "MASTER-A", + PacketSpec( + dst_id=91, + slot=2, + stream_id=stream_id, + frame_type=HBPF_VOICE, + dtype_vseq=1, + ), + ) + + self.assertEqual( + scenario.systems["MASTER-A"].STATUS[2]["RX_LC"], + bytes.fromhex("00000000005b2f9b81"), + ) + + def test_obp_late_entry_synthetic_lc_uses_normal_group_voice_options(self): + config = minimal_config(()) + add_openbridge_system(config, "OBP-1", network_id=3001) + add_openbridge_system(config, "OBP-2", network_id=3002) + bridges = active_bridge("91", 91, (("OBP-1", 1), ("OBP-2", 1))) + stream_id = 0x01020304 + + with DeterministicScenario(config=config, bridges=bridges) as scenario: + scenario.inject_obp( + "OBP-1", + PacketSpec( + dst_id=91, + slot=1, + stream_id=stream_id, + frame_type=HBPF_VOICE, + dtype_vseq=1, + ), + ) + + self.assertEqual( + scenario.systems["OBP-1"].STATUS[bytes_4(stream_id)]["LC"], + bytes.fromhex("00000000005b2f9b81"), + ) + + def test_hbp_real_voice_header_lc_is_preserved_with_service_options(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + bridges = active_bridge("91", 91, (("MASTER-A", 2), ("MASTER-B", 2))) + + with DeterministicScenario(config=config, bridges=bridges) as scenario: + scenario.inject_hbp( + "MASTER-A", + PacketSpec( + dst_id=91, + slot=2, + frame_type=HBPF_DATA_SYNC, + dtype_vseq=HBPF_SLT_VHEAD, + payload=bytes.fromhex( + "2b6004101f842dd00df07d41046dff57d75df5de30152e2070b20f803f88c695e2" + ), + ), + ) + + self.assertEqual( + scenario.systems["MASTER-A"].STATUS[2]["RX_LC"], + bytes.fromhex("001020000c302f9be5"), + ) + + def test_in_call_talker_alias_embedded_lc_is_logged(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + 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: + for index, payload in enumerate(self.TA_EMB_LC_PAYLOADS, start=1): + scenario.inject_hbp( + "MASTER-A", + PacketSpec( + dst_id=91, + slot=2, + stream_id=0x01020330, + seq=index, + frame_type=HBPF_VOICE, + dtype_vseq=index, + payload=payload, + ), + ) + + log_output = "\n".join(logs.output) + self.assertIn("*IN-CALL TA*", log_output) + self.assertIn("TEXT: 'CALL12'", log_output) + self.assertIn("LC: 04004c43414c4c3132", log_output) + + def test_in_call_gps_embedded_lc_is_logged(self): + config = minimal_config(("MASTER-A", "MASTER-B")) + 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: + for index, payload in enumerate(self.GPS_EMB_LC_PAYLOADS, start=1): + scenario.inject_hbp( + "MASTER-A", + PacketSpec( + dst_id=91, + slot=2, + stream_id=0x01020331, + seq=index, + frame_type=HBPF_VOICE, + dtype_vseq=index, + payload=payload, + ), + ) + + log_output = "\n".join(logs.output) + self.assertIn("*IN-CALL GPS*", log_output) + self.assertIn("LAT: 51.123451", log_output) + self.assertIn("LON: -2.123451", log_output) + self.assertIn("LC: 080007fcfae048b57b", log_output) + def test_hbp_vcsbk_data_reports_specific_rx_event_without_generic_duplicate(self): config = minimal_config(("MASTER-A", "MASTER-B")) config["REPORTS"]["REPORT"] = True @@ -2644,6 +3095,7 @@ DEFAULT_REFLECTOR: 0 ("OBP-2", 1), ), ) + bridges["91"][1]["TGID"] = bytes_3(235) stream_id = 0x01020304 header = PacketSpec( dst_id=91, @@ -2676,6 +3128,7 @@ DEFAULT_REFLECTOR: 0 captured = scenario.capture.for_system("OBP-2") self.assertEqual(len(captured), 1) self.assertEqual(captured[0].fields["dtype_vseq"], HBPF_SLT_VHEAD) + self.assertEqual(captured[0].fields["dst_id"], bytes_3(235)) def test_hbp_voice_sequence_zero_duplicate_is_dropped(self): bridges = active_bridge( diff --git a/tests/test_freedmr_dmr_codec.py b/tests/test_freedmr_dmr_codec.py new file mode 100644 index 0000000..5010517 --- /dev/null +++ b/tests/test_freedmr_dmr_codec.py @@ -0,0 +1,265 @@ +import unittest + +from freedmr_dmr_codec import ( + EmbeddedLCError, + FullLCError, + LC_SERVICE_OPTIONS_HBLINK_LEGACY, + LC_SERVICE_OPTIONS_NORMAL, + build_group_voice_lc, + SlotTypeError, + decode_embedded_lc, + decode_full_lc, + decode_slot_type, + embedded_lc_fragment_from_payload, + encode_embedded_lc, + encode_full_lc, + encode_full_lc_parity, + encode_slot_type, + 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" +) + + +class FreeDmrDmrCodecTest(unittest.TestCase): + 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() diff --git a/tests/test_udp_blackbox_harness.py b/tests/test_udp_blackbox_harness.py index 2d006b4..c7ca613 100644 --- a/tests/test_udp_blackbox_harness.py +++ b/tests/test_udp_blackbox_harness.py @@ -4,6 +4,8 @@ import time import unittest from pathlib import Path +from freedmr_dmr_codec import encode_embedded_lc, payload_with_embedded_lc_fragment + from tests.harness.deterministic import ( HBPF_DATA_SYNC, HBPF_SLT_VHEAD, @@ -22,7 +24,18 @@ from tests.harness.udp_blackbox import ( ) +def embedded_lc_payloads(lc): + payload = b"\x55" * 33 + return tuple( + payload_with_embedded_lc_fragment(payload, fragment) + for fragment in encode_embedded_lc(lc) + ) + + class UdpBlackBoxHarnessTest(unittest.TestCase): + TA_EMB_LC_PAYLOADS = embedded_lc_payloads(bytes.fromhex("04004c43414c4c3132")) + GPS_EMB_LC_PAYLOADS = embedded_lc_payloads(bytes.fromhex("080007fcfae048b57b")) + def private_call(self, dst_id, peer_id, slot=1, stream_id=0x01020304): return PacketSpec( peer_id=peer_id, @@ -121,6 +134,102 @@ class UdpBlackBoxHarnessTest(unittest.TestCase): self.assertEqual(captured.fields["dtype_vseq"], 7) self.assertEqual(captured.fields["dmr_payload"], payload) + def test_hbp_same_tg_voice_embedded_lc_payload_is_preserved(self): + require_udp_integration_enabled() + + payload = bytes((index * 7) % 256 for index in range(33)) + with UdpBlackBoxScenario() as scenario: + master_a = scenario.repeater("MASTER-A", 1001) + master_b = scenario.repeater("MASTER-B", 1002) + try: + master_a.login() + master_b.login() + + master_a.send_dmr( + PacketSpec( + peer_id=1001, + rf_src=3120001, + dst_id=91, + slot=2, + frame_type=HBPF_VOICE, + dtype_vseq=1, + payload=payload, + ) + ) + captured = master_b.recv(timeout=2.0) + finally: + master_a.close() + master_b.close() + + self.assertEqual(captured.fields["frame_type"], HBPF_VOICE) + self.assertEqual(captured.fields["dtype_vseq"], 1) + self.assertEqual(captured.fields["dmr_payload"], payload) + + def test_hbp_in_call_talker_alias_is_logged(self): + require_udp_integration_enabled() + + with UdpBlackBoxScenario() as scenario: + master_a = scenario.repeater("MASTER-A", 1001) + master_b = scenario.repeater("MASTER-B", 1002) + try: + master_a.login() + master_b.login() + + for index, payload in enumerate(self.TA_EMB_LC_PAYLOADS, start=1): + master_a.send_dmr( + PacketSpec( + peer_id=1001, + rf_src=3120001, + dst_id=91, + slot=2, + stream_id=0x01020330, + seq=index, + frame_type=HBPF_VOICE, + dtype_vseq=index, + payload=payload, + ) + ) + output = scenario.process.wait_for_log("*IN-CALL TA*", timeout=2.0) + finally: + master_a.close() + master_b.close() + + self.assertIn("TEXT: 'CALL12'", output) + self.assertIn("LC: 04004c43414c4c3132", output) + + def test_hbp_in_call_gps_is_logged(self): + require_udp_integration_enabled() + + with UdpBlackBoxScenario() as scenario: + master_a = scenario.repeater("MASTER-A", 1001) + master_b = scenario.repeater("MASTER-B", 1002) + try: + master_a.login() + master_b.login() + + for index, payload in enumerate(self.GPS_EMB_LC_PAYLOADS, start=1): + master_a.send_dmr( + PacketSpec( + peer_id=1001, + rf_src=3120001, + dst_id=91, + slot=2, + stream_id=0x01020331, + seq=index, + frame_type=HBPF_VOICE, + dtype_vseq=index, + payload=payload, + ) + ) + output = scenario.process.wait_for_log("*IN-CALL GPS*", timeout=2.0) + finally: + master_a.close() + master_b.close() + + self.assertIn("LAT: 51.123451", output) + self.assertIn("LON: -2.123451", output) + self.assertIn("LC: 080007fcfae048b57b", output) + def test_hbp_malformed_short_dmrd_is_ignored_and_later_packet_routes(self): require_udp_integration_enabled() @@ -442,6 +551,64 @@ class UdpBlackBoxHarnessTest(unittest.TestCase): self.assertEqual(captured.fields["slot"], 2) self.assertEqual(leaked, []) + def test_dial_a_tg_slot_1_routes_local_tg9_to_fbp_reflector_tg(self): + require_udp_integration_enabled() + + with UdpBlackBoxScenario( + ts2_static="", + fbp_systems={"OBP-1": 3001}, + ) as scenario: + master_a = scenario.repeater("MASTER-A", 1001) + fbp_peer = scenario.fbp_peer("OBP-1") + try: + master_a.login() + fbp_peer.send_bcka() + fbp_peer.send_bcve() + fbp_peer.drain(seconds=0.2) + + master_a.send_dmr(self.private_call(235, peer_id=1001, slot=1)) + master_a.send_dmr( + PacketSpec( + peer_id=1001, + rf_src=3120001, + dst_id=9, + slot=1, + stream_id=0x01020321, + ) + ) + captured = fbp_peer.recv_dmre(timeout=2.0) + finally: + master_a.close() + + self.assertEqual(captured.fields["dst_id"], bytes_3(235)) + self.assertEqual(captured.fields["slot"], 1) + + def test_dynamic_tg_routing_disabled_does_not_create_unknown_hbp_route(self): + require_udp_integration_enabled() + + with UdpBlackBoxScenario(ts2_static="", dynamic_tg_routing=False) as scenario: + master_a = scenario.repeater("MASTER-A", 1001) + master_b = scenario.repeater("MASTER-B", 1002) + try: + master_a.login() + master_b.login() + + master_a.send_dmr( + PacketSpec( + peer_id=1001, + rf_src=3120001, + dst_id=12345, + slot=2, + stream_id=0x01020322, + ) + ) + leaked = master_b.drain(seconds=0.4) + finally: + master_a.close() + master_b.close() + + self.assertEqual(leaked, []) + def test_real_hbp_voice_interrupts_generated_prompt_and_routes(self): require_udp_integration_enabled() diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..518222f --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,24 @@ +import unittest + +from utils import bytes_3, bytes_4, get_alias, int_id + + +class FreeDmrUtilsTest(unittest.TestCase): + def test_integer_byte_helpers_are_big_endian(self): + self.assertEqual(bytes_3(0x010203), b"\x01\x02\x03") + self.assertEqual(bytes_4(0x01020304), b"\x01\x02\x03\x04") + self.assertEqual(int_id(b"\x01\x02\x03\x04"), 0x01020304) + + def test_get_alias_returns_matching_record_or_original_id(self): + aliases = { + 3120001: {"callsign": "M0ABC", "name": "Test"}, + 235: "TG235", + } + + self.assertEqual(get_alias(bytes_3(3120001), aliases, "callsign"), ["M0ABC"]) + self.assertEqual(get_alias(235, aliases), "TG235") + self.assertEqual(get_alias(999, aliases), 999) + + +if __name__ == "__main__": + unittest.main() diff --git a/utils.py b/utils.py index df85a89..51fa24c 100644 --- a/utils.py +++ b/utils.py @@ -18,11 +18,13 @@ # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA ############################################################################### -#Some utilty functions from dmr_utils3 have been modified. These live here. - -# Also new FreeDMR specific functions. +# FreeDMR utility functions. +# +# Some helpers mirror the old dmr_utils3 utility surface so active runtime code +# does not depend on dmr_utils3 for simple ID/byte conversions. import ssl +from binascii import b2a_hex as ahex from time import time from os.path import isfile, getmtime from urllib.request import urlopen @@ -30,6 +32,34 @@ from json import load as jload, dump as jdump import hashlib +def bytes_3(_int_id): + return _int_id.to_bytes(3, 'big') + + +def bytes_4(_int_id): + return _int_id.to_bytes(4, 'big') + + +def int_id(_hex_bytes): + return int(ahex(_hex_bytes), 16) + + +def get_alias(_id, _dict, *args): + if type(_id) == bytes: + _id = int_id(_id) + if _id in _dict: + if args: + retValue = [] + for _item in args: + try: + retValue.append(_dict[_id][_item]) + except TypeError: + return _dict[_id] + return retValue + else: + return _dict[_id] + return _id + #Use this try_download instead of that from dmr_utils3 def try_download(_path, _file, _url, _stale,):