diff --git a/.gitattributes b/.gitattributes old mode 100644 new mode 100755 diff --git a/Dockerfile b/Dockerfile deleted file mode 100644 index 842c902..0000000 --- a/Dockerfile +++ /dev/null @@ -1,30 +0,0 @@ -FROM python:3.7-slim-stretch - -RUN apt update && \ - apt install -y git && \ - cd /usr/src/ && \ - git clone https://github.com/n0mjs710/dmr_utils3 && \ - cd /usr/src/dmr_utils3 && \ - ./install.sh && \ - rm -rf /var/lib/apt/lists/* && \ - cd /opt && \ - rm -rf /usr/src/dmr_utils3 && \ - git clone https://github.com/n0mjs710/hblink3 -ENV AAA BBBB -RUN cd /opt/hblink3/ && \ - sed -i s/.*python.*//g requirements.txt && \ - pip install --no-cache-dir -r requirements.txt - - -ADD entrypoint /entrypoint - -RUN adduser -u 54000 radio && \ - adduser radio radio && \ - chmod 755 /entrypoint && \ - chown radio:radio /entrypoint && \ - chown radio /opt/hblink3 - -USER radio -EXPOSE 54000 - -ENTRYPOINT [ "/entrypoint" ] diff --git a/app_template.py b/app_template.py deleted file mode 100755 index b919960..0000000 --- a/app_template.py +++ /dev/null @@ -1,150 +0,0 @@ -#!/usr/bin/env python -# -############################################################################### -# Copyright (C) 2016-2019 Cortney T. Buffington, N0MJS -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program; if not, write to the Free Software Foundation, -# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA -############################################################################### - -# Python modules we need -import sys -from bitarray import bitarray -from time import time -from importlib import import_module - -# Twisted is pretty important, so I keep it separate -from twisted.internet.protocol import Factory, Protocol -from twisted.protocols.basic import NetstringReceiver -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 config -import log -from const import * - -# Stuff for socket reporting -import pickle -# REMOVE LATER from datetime import datetime -# The module needs logging, but handlers, etc. are controlled by the parent -import logging -logger = logging.getLogger(__name__) - - -# Does anybody read this stuff? There's a PEP somewhere that says I should do this. -__author__ = 'Cortney T. Buffington, N0MJS' -__copyright__ = 'Copyright (c) 2016-2018 Cortney T. Buffington, N0MJS and the K0USY Group' -__credits__ = 'Colin Durbridge, G4EML, Steve Zingman, N4IRS; Mike Zingman, N4IRR; Jonathan Naylor, G4KLX; Hans Barthen, DL5DI; Torsten Shultze, DG1HT' -__license__ = 'GNU GPLv3' -__maintainer__ = 'Cort Buffington, N0MJS' -__email__ = 'n0mjs@me.com' - -# Module gobal varaibles - - -class OBP(OPENBRIDGE): - - def __init__(self, _name, _config, _report): - OPENBRIDGE.__init__(self, _name, _config, _report) - - - def dmrd_received(self, _peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data): - pass - - -class HBP(HBSYSTEM): - - def __init__(self, _name, _config, _report): - HBSYSTEM.__init__(self, _name, _config, _report) - - def dmrd_received(self, _peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data): - pass - - - -#************************************************ -# MAIN PROGRAM LOOP STARTS HERE -#************************************************ - -if __name__ == '__main__': - - import argparse - import sys - import os - import signal - - # Change the current directory to the location of the application - os.chdir(os.path.dirname(os.path.realpath(sys.argv[0]))) - - # CLI argument parser - handles picking up the config file from the command line, and sending a "help" message - parser = argparse.ArgumentParser() - parser.add_argument('-c', '--config', action='store', dest='CONFIG_FILE', help='/full/path/to/config.file (usually hblink.cfg)') - parser.add_argument('-l', '--logging', action='store', dest='LOG_LEVEL', help='Override config file logging level.') - cli_args = parser.parse_args() - - # Ensure we have a path for the config file, if one wasn't specified, then use the default (top of file) - if not cli_args.CONFIG_FILE: - cli_args.CONFIG_FILE = os.path.dirname(os.path.abspath(__file__))+'/hblink.cfg' - - # Call the external routine to build the configuration dictionary - CONFIG = config.build_config(cli_args.CONFIG_FILE) - - # Start the system logger - if cli_args.LOG_LEVEL: - CONFIG['LOGGER']['LOG_LEVEL'] = cli_args.LOG_LEVEL - logger = log.config_logging(CONFIG['LOGGER']) - logger.info('\n\nCopyright (c) 2013, 2014, 2015, 2016, 2018\n\tThe Regents of the K0USY Group. All rights reserved.\n') - logger.debug('(GLOBAL) Logging system started, anything from here on gets logged') - - # Set up the signal handler - def sig_handler(_signal, _frame): - logger.info('(GLOBAL) SHUTDOWN: CONFBRIDGE IS TERMINATING WITH SIGNAL %s', str(_signal)) - hblink_handler(_signal, _frame) - logger.info('(GLOBAL) SHUTDOWN: ALL SYSTEM HANDLERS EXECUTED - STOPPING REACTOR') - reactor.stop() - - # Set signal handers so that we can gracefully exit if need be - for sig in [signal.SIGINT, signal.SIGTERM]: - signal.signal(sig, sig_handler) - - # Create the name-number mapping dictionaries - peer_ids, subscriber_ids, talkgroup_ids = mk_aliases(CONFIG) - - # INITIALIZE THE REPORTING LOOP - if CONFIG['REPORTS']['REPORT']: - report_server = config_reports(CONFIG, bridgeReportFactory) - else: - report_server = None - logger.info('(REPORT) TCP Socket reporting not configured') - - # HBlink instance creation - logger.info('(GLOBAL) HBlink \'bridge.py\' -- SYSTEM STARTING...') - for system in CONFIG['SYSTEMS']: - if CONFIG['SYSTEMS'][system]['ENABLED']: - if CONFIG['SYSTEMS'][system]['MODE'] == 'OPENBRIDGE': - systems[system] = OBP(system, CONFIG, report_server) - else: - systems[system] = HBP(system, CONFIG, report_server) - reactor.listenUDP(CONFIG['SYSTEMS'][system]['PORT'], systems[system], interface=CONFIG['SYSTEMS'][system]['IP']) - logger.debug('(GLOBAL) %s instance created: %s, %s', CONFIG['SYSTEMS'][system]['MODE'], system, systems[system]) - - def loopingErrHandle(failure): - logger.error('(GLOBAL) STOPPING REACTOR TO AVOID MEMORY LEAK: Unhandled error in timed loop.\n %s', failure) - reactor.stop() - - - reactor.run() diff --git a/blank_app.py b/blank_app.py deleted file mode 100755 index b13e29a..0000000 --- a/blank_app.py +++ /dev/null @@ -1,143 +0,0 @@ -#!/usr/bin/env python -# -############################################################################### -# Copyright (C) 2020 Cortney T. Buffington, N0MJS -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program; if not, write to the Free Software Foundation, -# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA -############################################################################### - -''' -This is a blank application template to build on top of hblink.py. It contains -only the things you need to, essentially, run hblink.py underneath and do nothing -else. The expected behaviour is to override the dmrd_received function from -hblink.py to do somethign meaningful, so that framework is completed, but as -it stands, still does nothing with the DMRD packet. -''' - -# Python modules we need -import sys -from bitarray import bitarray -from time import time -from importlib import import_module -from types import ModuleType - -# Twisted is pretty important, so I keep it separate -from twisted.internet.protocol import Factory, Protocol -from twisted.protocols.basic import NetstringReceiver -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, config_reports, mk_aliases, acl_check -from dmr_utils3.utils import bytes_3, int_id, get_alias -from dmr_utils3 import decode, bptc, const -import config -import log -import const - -# The module needs logging logging, but handlers, etc. are controlled by the parent -import logging -logger = logging.getLogger(__name__) - - -# Does anybody read this stuff? There's a PEP somewhere that says I should do this. -__author__ = 'Cortney T. Buffington, N0MJS' -__copyright__ = 'Copyright (c) 2020 Cortney T. Buffington' -__credits__ = 'Colin Durbridge, G4EML, Steve Zingman, N4IRS; Mike Zingman, N4IRR; Jonathan Naylor, G4KLX; Hans Barthen, DL5DI; Torsten Shultze, DG1HT' -__license__ = 'GNU GPLv3' -__maintainer__ = 'Cort Buffington, N0MJS' -__email__ = 'n0mjs@me.com' -__status__ = 'pre-alpha' - -# Module gobal varaibles - - -class blankSYSTEM(HBSYSTEM): - - def __init__(self, _name, _config, _report): - HBSYSTEM.__init__(self, _name, _config, _report) - - def dmrd_received(self, _peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data): - pass - - -#************************************************ -# MAIN PROGRAM LOOP STARTS HERE -#************************************************ - -if __name__ == '__main__': - import argparse - import sys - import os - import signal - from dmr_utils3.utils import try_download, mk_id_dict - - # Change the current directory to the location of the application - os.chdir(os.path.dirname(os.path.realpath(sys.argv[0]))) - - # CLI argument parser - handles picking up the config file from the command line, and sending a "help" message - parser = argparse.ArgumentParser() - parser.add_argument('-c', '--config', action='store', dest='CONFIG_FILE', help='/full/path/to/config.file (usually hblink.cfg)') - parser.add_argument('-l', '--logging', action='store', dest='LOG_LEVEL', help='Override config file logging level.') - cli_args = parser.parse_args() - - # Ensure we have a path for the config file, if one wasn't specified, then use the default (top of file) - if not cli_args.CONFIG_FILE: - cli_args.CONFIG_FILE = os.path.dirname(os.path.abspath(__file__))+'/hblink.cfg' - - # Call the external routine to build the configuration dictionary - CONFIG = config.build_config(cli_args.CONFIG_FILE) - - # Start the system logger - if cli_args.LOG_LEVEL: - CONFIG['LOGGER']['LOG_LEVEL'] = cli_args.LOG_LEVEL - logger = log.config_logging(CONFIG['LOGGER']) - logger.info('\n\nCopyright (c) 2013, 2014, 2015, 2016, 2018, 2019\n\tThe Regents of the K0USY Group. All rights reserved.\n') - logger.debug('Logging system started, anything from here on gets logged') - - # Set up the signal handler - def sig_handler(_signal, _frame): - logger.info('SHUTDOWN: >>>BLANK APP<<< IS TERMINATING WITH SIGNAL %s', str(_signal)) - hblink_handler(_signal, _frame) - logger.info('SHUTDOWN: ALL SYSTEM HANDLERS EXECUTED - STOPPING REACTOR') - reactor.stop() - - # Set signal handers so that we can gracefully exit if need be - for sig in [signal.SIGTERM, signal.SIGINT]: - signal.signal(sig, sig_handler) - - # Create the name-number mapping dictionaries - peer_ids, subscriber_ids, talkgroup_ids = mk_aliases(CONFIG) - - - # INITIALIZE THE REPORTING LOOP - if CONFIG['REPORTS']['REPORT']: - report_server = config_reports(CONFIG, reportFactory) - else: - report_server = None - logger.info('(REPORT) TCP Socket reporting not configured') - - # HBlink instance creation - logger.info('HBlink \'blank_app.py\' -- SYSTEM STARTING...') - for system in CONFIG['SYSTEMS']: - if CONFIG['SYSTEMS'][system]['ENABLED']: - if CONFIG['SYSTEMS'][system]['MODE'] == 'OPENBRIDGE': - systems[system] = OPENBRIDGE(system, CONFIG, report_server) - else: - systems[system] = HBSYSTEM(system, CONFIG, report_server) - - reactor.listenUDP(CONFIG['SYSTEMS'][system]['PORT'], systems[system], interface=CONFIG['SYSTEMS'][system]['IP']) - logger.debug('%s instance created: %s, %s', CONFIG['SYSTEMS'][system]['MODE'], system, systems[system]) - - reactor.run() diff --git a/bridge.py b/bridge.py index ae526d6..fc40c19 100755 --- a/bridge.py +++ b/bridge.py @@ -67,6 +67,13 @@ __email__ = 'n0mjs@me.com' # Module gobal varaibles +# Dictionary for dynamically mapping unit (subscriber) to a system. +# This is for pruning unit-to-uint calls to not broadcast once the +# target system for a unit is identified +# format 'unit_id': ('SYSTEM', time) +UNIT_MAP = {} + + # Timed loop used for reporting HBP status # # REPORT BASED ON THE TYPE SELECTED IN THE MAIN CONFIG FILE @@ -116,6 +123,7 @@ def make_bridges(_rules): # Run this every minute for rule timer updates def rule_timer_loop(): + global UNIT_MAP logger.debug('(ROUTER) routerHBP Rule timer loop started') _now = time() @@ -144,6 +152,18 @@ def rule_timer_loop(): else: logger.debug('(ROUTER) Conference Bridge NO ACTION: System: %s, Bridge: %s, TS: %s, TGID: %s', _system['SYSTEM'], _bridge, _system['TS'], int_id(_system['TGID'])) + _then = _now - 60 + remove_list = [] + for unit in UNIT_MAP: + if UNIT_MAP[unit][1] < (_then): + remove_list.append(unit) + + for unit in remove_list: + del UNIT_MAP[unit] + + logger.debug('Removed unit(s) %s from UNIT_MAP', remove_list) + + if CONFIG['REPORTS']['REPORT']: report_server.send_clients(b'bridge updated') @@ -186,10 +206,14 @@ def stream_trimmer_loop(): if stream_id in systems[system].STATUS: _stream = systems[system].STATUS[stream_id] _sysconfig = CONFIG['SYSTEMS'][system] - logger.info('(%s) *TIME OUT* STREAM ID: %s SUB: %s PEER: %s TGID: %s TS 1 Duration: %.2f', \ - system, int_id(stream_id), get_alias(int_id(_stream['RFS']), subscriber_ids), get_alias(int_id(_sysconfig['NETWORK_ID']), peer_ids), get_alias(int_id(_stream['TGID']), talkgroup_ids), _stream['LAST'] - _stream['START']) + if systems[system].STATUS[stream_id]['ACTIVE']: + logger.info('(%s) *TIME OUT* STREAM ID: %s SUB: %s PEER: %s TYPE: %s DST ID: %s TS 1 Duration: %.2f', \ + system, int_id(stream_id), get_alias(int_id(_stream['RFS']), subscriber_ids), get_alias(int_id(_sysconfig['NETWORK_ID']), peer_ids), _stream['TYPE'], get_alias(int_id(_stream['DST']), talkgroup_ids), _stream['LAST'] - _stream['START']) if CONFIG['REPORTS']['REPORT']: - systems[system]._report.send_bridgeEvent('GROUP VOICE,END,RX,{},{},{},{},{},{},{:.2f}'.format(system, int_id(stream_id), int_id(_sysconfig['NETWORK_ID']), int_id(_stream['RFS']), 1, int_id(_stream['TGID']), _stream['LAST'] - _stream['START']).encode(encoding='utf-8', errors='ignore')) + if _stream['TYPE'] == 'GROUP': + systems[system]._report.send_bridgeEvent('GROUP VOICE,END,RX,{},{},{},{},{},{},{:.2f}'.format(system, int_id(stream_id), int_id(_sysconfig['NETWORK_ID']), int_id(_stream['RFS']), 1, int_id(_stream['DST']), _stream['LAST'] - _stream['START']).encode(encoding='utf-8', errors='ignore')) + elif _stream['TYPE'] == 'UNIT': + systems[system]._report.send_bridgeEvent('UNIT VOICE,END,RX,{},{},{},{},{},{},{:.2f}'.format(system, int_id(stream_id), int_id(_sysconfig['NETWORK_ID']), int_id(_stream['RFS']), 1, int_id(_stream['DST']), _stream['LAST'] - _stream['START']).encode(encoding='utf-8', errors='ignore')) removed = systems[system].STATUS.pop(stream_id) else: logger.error('(%s) Attemped to remove OpenBridge Stream ID %s not in the Stream ID list: %s', system, int_id(stream_id), [id for id in systems[system].STATUS]) @@ -198,205 +222,373 @@ class routerOBP(OPENBRIDGE): def __init__(self, _name, _config, _report): OPENBRIDGE.__init__(self, _name, _config, _report) + self.name = _name self.STATUS = {} + + # list of self._targets for unit (subscriber, private) calls + self._targets = [] - - def dmrd_received(self, _peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data): + def group_received(self, _peer_id, _rf_src, _dst_id, _seq, _slot, _frame_type, _dtype_vseq, _stream_id, _data): pkt_time = time() dmrpkt = _data[20:53] _bits = _data[15] + + # Is this a new call stream? + if (_stream_id not in self.STATUS): + # This is a new call stream + self.STATUS[_stream_id] = { + 'START': pkt_time, + 'CONTENTION':False, + 'RFS': _rf_src, + 'TYPE': 'GROUP', + 'DST': _dst_id, + 'ACTIVE': True + } + + # 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) + 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 + + + logger.info('(%s) *GROUP CALL START* OBP STREAM ID: %s SUB: %s (%s) PEER: %s (%s) TGID %s (%s), TS %s', \ + self._system, int_id(_stream_id), get_alias(_rf_src, subscriber_ids), int_id(_rf_src), get_alias(_peer_id, peer_ids), int_id(_peer_id), get_alias(_dst_id, talkgroup_ids), int_id(_dst_id), _slot) + if CONFIG['REPORTS']['REPORT']: + self._report.send_bridgeEvent('GROUP VOICE,START,RX,{},{},{},{},{},{}'.format(self._system, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id)).encode(encoding='utf-8', errors='ignore')) + + self.STATUS[_stream_id]['LAST'] = pkt_time + + + for _bridge in BRIDGES: + for _system in BRIDGES[_bridge]: + + if (_system['SYSTEM'] == self._system and _system['TGID'] == _dst_id and _system['TS'] == _slot and _system['ACTIVE'] == True): + + for _target in BRIDGES[_bridge]: + if (_target['SYSTEM'] != self._system) and (_target['ACTIVE']): + _target_status = systems[_target['SYSTEM']].STATUS + _target_system = self._CONFIG['SYSTEMS'][_target['SYSTEM']] + if _target_system['MODE'] == 'OPENBRIDGE': + # Is this a new call stream on the target? + if (_stream_id not in _target_status): + # This is a new call stream on the target + _target_status[_stream_id] = { + 'START': pkt_time, + 'CONTENTION':False, + 'RFS': _rf_src, + 'TYPE': 'GROUP', + 'DST': _dst_id, + 'ACTIVE': True + } + # 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) + + 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']: + systems[_target['SYSTEM']]._report.send_bridgeEvent('GROUP VOICE,START,TX,{},{},{},{},{},{}'.format(_target['SYSTEM'], int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _target['TS'], int_id(_target['TGID'])).encode(encoding='utf-8', errors='ignore')) + + # Record the time of this packet so we can later identify a stale stream + _target_status[_stream_id]['LAST'] = pkt_time + # Clear the TS bit -- all OpenBridge streams are effectively on TS1 + _tmp_bits = _bits & ~(1 << 7) + + # Assemble transmit HBP packet header + _tmp_data = b''.join([_data[:8], _target['TGID'], _data[11:15], _tmp_bits.to_bytes(1, 'big'), _data[16:20]]) + + # MUST TEST FOR NEW STREAM AND IF SO, RE-WRITE THE LC FOR THE TARGET + # MUST RE-WRITE DESTINATION TGID IF DIFFERENT + # if _dst_id != rule['DST_GROUP']: + dmrbits = bitarray(endian='big') + dmrbits.frombytes(dmrpkt) + # Create a voice header packet (FULL LC) + if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD: + dmrbits = _target_status[_stream_id]['H_LC'][0:98] + dmrbits[98:166] + _target_status[_stream_id]['H_LC'][98:197] + # Create a voice terminator packet (FULL LC) + elif _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VTERM: + dmrbits = _target_status[_stream_id]['T_LC'][0:98] + dmrbits[98:166] + _target_status[_stream_id]['T_LC'][98:197] + if CONFIG['REPORTS']['REPORT']: + call_duration = pkt_time - _target_status[_stream_id]['START'] + _target_status[_stream_id]['ACTIVE'] = False + 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 _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]) + + else: + # BEGIN CONTENTION HANDLING + # + # The rules for each of the 4 "ifs" below are listed here for readability. The Frame To Send is: + # From a different group than last RX from this HBSystem, but it has been less than Group Hangtime + # From a different group than last TX to this HBSystem, but it has been less than Group Hangtime + # From the same group as the last RX from this HBSystem, but from a different subscriber, and it has been less than stream timeout + # From the same group as the last TX to this HBSystem, but from a different subscriber, and it has been less than stream timeout + # The "continue" at the end of each means the next iteration of the for loop that tests for matching rules + # + if ((_target['TGID'] != _target_status[_target['TS']]['RX_TGID']) and ((pkt_time - _target_status[_target['TS']]['RX_TIME']) < _target_system['GROUP_HANGTIME'])): + if self.STATUS[_stream_id]['CONTENTION'] == False: + self.STATUS[_stream_id]['CONTENTION'] = True + logger.info('(%s) Call not routed to TGID %s, target active or in group hangtime: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_target['TGID']), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['RX_TGID'])) + continue + if ((_target['TGID'] != _target_status[_target['TS']]['TX_TGID']) and ((pkt_time - _target_status[_target['TS']]['TX_TIME']) < _target_system['GROUP_HANGTIME'])): + if self.STATUS[_stream_id]['CONTENTION'] == False: + self.STATUS[_stream_id]['CONTENTION'] = True + logger.info('(%s) Call not routed to TGID%s, target in group hangtime: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_target['TGID']), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['TX_TGID'])) + continue + if (_target['TGID'] == _target_status[_target['TS']]['RX_TGID']) and ((pkt_time - _target_status[_target['TS']]['RX_TIME']) < STREAM_TO): + if self.STATUS[_stream_id]['CONTENTION'] == False: + self.STATUS[_stream_id]['CONTENTION'] = True + logger.info('(%s) Call not routed to TGID%s, matching call already active on target: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_target['TGID']), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['RX_TGID'])) + continue + if (_target['TGID'] == _target_status[_target['TS']]['TX_TGID']) and (_rf_src != _target_status[_target['TS']]['TX_RFS']) and ((pkt_time - _target_status[_target['TS']]['TX_TIME']) < STREAM_TO): + if self.STATUS[_stream_id]['CONTENTION'] == False: + self.STATUS[_stream_id]['CONTENTION'] = True + logger.info('(%s) Call not routed for subscriber %s, call route in progress on target: HBSystem: %s, TS: %s, TGID: %s, SUB: %s', self._system, int_id(_rf_src), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['TX_TGID']), int_id(_target_status[_target['TS']]['TX_RFS'])) + continue + + # Is this a new call stream? + if (_target_status[_target['TS']]['TX_STREAM_ID'] != _stream_id): + # Record the DST TGID and Stream ID + _target_status[_target['TS']]['TX_START'] = pkt_time + _target_status[_target['TS']]['TX_TGID'] = _target['TGID'] + _target_status[_target['TS']]['TX_STREAM_ID'] = _stream_id + _target_status[_target['TS']]['TX_RFS'] = _rf_src + _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) + 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']: + systems[_target['SYSTEM']]._report.send_bridgeEvent('GROUP VOICE,START,TX,{},{},{},{},{},{}'.format(_target['SYSTEM'], int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _target['TS'], int_id(_target['TGID'])).encode(encoding='utf-8', errors='ignore')) + + # Set other values for the contention handler to test next time there is a frame to forward + _target_status[_target['TS']]['TX_TIME'] = pkt_time + _target_status[_target['TS']]['TX_TYPE'] = _dtype_vseq + + # Handle any necessary re-writes for the destination + if _system['TS'] != _target['TS']: + _tmp_bits = _bits ^ 1 << 7 + else: + _tmp_bits = _bits + + # Assemble transmit HBP packet header + _tmp_data = b''.join([_data[:8], _target['TGID'], _data[11:15], _tmp_bits.to_bytes(1, 'big'), _data[16:20]]) + + # MUST TEST FOR NEW STREAM AND IF SO, RE-WRITE THE LC FOR THE TARGET + # MUST RE-WRITE DESTINATION TGID IF DIFFERENT + # if _dst_id != rule['DST_GROUP']: + dmrbits = bitarray(endian='big') + dmrbits.frombytes(dmrpkt) + # Create a voice header packet (FULL LC) + if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD: + dmrbits = _target_status[_target['TS']]['TX_H_LC'][0:98] + dmrbits[98:166] + _target_status[_target['TS']]['TX_H_LC'][98:197] + # Create a voice terminator packet (FULL LC) + elif _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VTERM: + dmrbits = _target_status[_target['TS']]['TX_T_LC'][0:98] + dmrbits[98:166] + _target_status[_target['TS']]['TX_T_LC'][98:197] + if CONFIG['REPORTS']['REPORT']: + 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 _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] + + # Transmit the packet to the destination system + systems[_target['SYSTEM']].send_system(_tmp_data) + #logger.debug('(%s) Packet routed by bridge: %s to system: %s TS: %s, TGID: %s', self._system, _bridge, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) + + + # Final actions - Is this a voice terminator? + if (_frame_type == HBPF_DATA_SYNC) and (_dtype_vseq == HBPF_SLT_VTERM): + call_duration = pkt_time - self.STATUS[_stream_id]['START'] + logger.info('(%s) *GROUP CALL END* STREAM ID: %s SUB: %s (%s) PEER: %s (%s) TGID %s (%s), TS %s, Duration: %.2f', \ + self._system, int_id(_stream_id), get_alias(_rf_src, subscriber_ids), int_id(_rf_src), get_alias(_peer_id, peer_ids), int_id(_peer_id), get_alias(_dst_id, talkgroup_ids), int_id(_dst_id), _slot, call_duration) + if CONFIG['REPORTS']['REPORT']: + self._report.send_bridgeEvent('GROUP VOICE,END,RX,{},{},{},{},{},{},{:.2f}'.format(self._system, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id), call_duration).encode(encoding='utf-8', errors='ignore')) + self.STATUS[_stream_id]['ACTIVE'] = False + logger.debug('(%s) OpenBridge sourced call stream end, remove terminated Stream ID: %s', self._system, int_id(_stream_id)) + + + def unit_received(self, _peer_id, _rf_src, _dst_id, _seq, _slot, _frame_type, _dtype_vseq, _stream_id, _data): + global UNIT_MAP + pkt_time = time() + dmrpkt = _data[20:53] + _bits = _data[15] + + # Make/update this unit in the UNIT_MAP cache + UNIT_MAP[_rf_src] = (self.name, pkt_time) + + + # Is this a new call stream? + if (_stream_id not in self.STATUS): + # This is a new call stream + self.STATUS[_stream_id] = { + 'START': pkt_time, + 'CONTENTION':False, + 'RFS': _rf_src, + 'TYPE': 'UNIT', + 'DST': _dst_id, + 'ACTIVE': True + } + + # Create a destination list for the call: + if _dst_id in UNIT_MAP: + if UNIT_MAP[_dst_id][0] != self._system: + self._targets = [UNIT_MAP[_dst_id][0]] + else: + self._targets = [] + logger.error('UNIT call to a subscriber on the same system, send nothing') + else: + self._targets = list(UNIT) + self._targets.remove(self._system) + + + # This is a new call stream, so log & report + logger.info('(%s) *UNIT CALL START* STREAM ID: %s SUB: %s (%s) PEER: %s (%s) UNIT: %s (%s), TS: %s, FORWARD: %s', \ + self._system, int_id(_stream_id), get_alias(_rf_src, subscriber_ids), int_id(_rf_src), get_alias(_peer_id, peer_ids), int_id(_peer_id), get_alias(_dst_id, talkgroup_ids), int_id(_dst_id), _slot, self._targets) + if CONFIG['REPORTS']['REPORT']: + self._report.send_bridgeEvent('UNIT VOICE,START,RX,{},{},{},{},{},{},{}'.format(self._system, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id), self._targets).encode(encoding='utf-8', errors='ignore')) + + # Record the time of this packet so we can later identify a stale stream + self.STATUS[_stream_id]['LAST'] = pkt_time + + for _target in self._targets: + _target_status = systems[_target].STATUS + _target_system = self._CONFIG['SYSTEMS'][_target] + + if self._CONFIG['SYSTEMS'][_target]['MODE'] == 'OPENBRIDGE': + if (_stream_id not in _target_status): + # This is a new call stream on the target + _target_status[_stream_id] = { + 'START': pkt_time, + 'CONTENTION':False, + 'RFS': _rf_src, + 'TYPE': 'UNIT', + 'DST': _dst_id, + 'ACTIVE': True + } + + logger.info('(%s) Unit call bridged to OBP System: %s TS: %s, TGID: %s', self._system, _target, _slot if _target_system['BOTH_SLOTS'] else 1, int_id(_dst_id)) + if CONFIG['REPORTS']['REPORT']: + systems[_target]._report.send_bridgeEvent('UNIT VOICE,START,TX,{},{},{},{},{},{}'.format(_target, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id)).encode(encoding='utf-8', errors='ignore')) + + # Record the time of this packet so we can later identify a stale stream + _target_status[_stream_id]['LAST'] = pkt_time + # Clear the TS bit and follow propper OBP definition, unless "BOTH_SLOTS" is set. This only works for unit calls. + if _target_system['BOTH_SLOTS']: + _tmp_bits = _bits + else: + _tmp_bits = _bits & ~(1 << 7) + + # Assemble transmit HBP packet + _tmp_data = b''.join([_data[:15], _tmp_bits.to_bytes(1, 'big'), _data[16:20]]) + _data = b''.join([_tmp_data, dmrpkt]) + + if (_frame_type == HBPF_DATA_SYNC) and (_dtype_vseq == HBPF_SLT_VTERM): + _target_status[_stream_id]['ACTIVE'] = False + + else: + # BEGIN STANDARD CONTENTION HANDLING + # + # The rules for each of the 4 "ifs" below are listed here for readability. The Frame To Send is: + # From a different group than last RX from this HBSystem, but it has been less than Group Hangtime + # From a different group than last TX to this HBSystem, but it has been less than Group Hangtime + # From the same group as the last RX from this HBSystem, but from a different subscriber, and it has been less than stream timeout + # From the same group as the last TX to this HBSystem, but from a different subscriber, and it has been less than stream timeout + # The "continue" at the end of each means the next iteration of the for loop that tests for matching rules + # + ''' + if ((_dst_id != _target_status[_slot]['RX_TGID']) and ((pkt_time - _target_status[_slot]['RX_TIME']) < _target_system['GROUP_HANGTIME'])): + if self.STATUS[_stream_id]['CONTENTION'] == False: + self.STATUS[_stream_id]['CONTENTION'] = True + logger.info('(%s) Call not routed to TGID %s, target active or in group hangtime: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_dst_id), _target, _slot, int_id(_target_status[_slot]['RX_TGID'])) + continue + if ((_dst_id != _target_status[_slot]['TX_TGID']) and ((pkt_time - _target_status[_slot]['TX_TIME']) < _target_system['GROUP_HANGTIME'])): + if self.STATUS[_stream_id]['CONTENTION'] == False: + self.STATUS[_stream_id]['CONTENTION'] = True + logger.info('(%s) Call not routed to TGID%s, target in group hangtime: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_dst_id), _target, _slot, int_id(_target_status[_slot]['TX_TGID'])) + continue + ''' + if (_dst_id == _target_status[_slot]['RX_TGID']) and ((pkt_time - _target_status[_slot]['RX_TIME']) < STREAM_TO): + if self.STATUS[_stream_id]['CONTENTION'] == False: + self.STATUS[_stream_id]['CONTENTION'] = True + logger.info('(%s) Call not routed to TGID%s, matching call already active on target: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_dst_id), _target, _slot, int_id(_target_status[_slot]['RX_TGID'])) + continue + if (_dst_id == _target_status[_slot]['TX_TGID']) and (_rf_src != _target_status[_slot]['TX_RFS']) and ((pkt_time - _target_status[_slot]['TX_TIME']) < STREAM_TO): + if self.STATUS[_stream_id]['CONTENTION'] == False: + self.STATUS[_stream_id]['CONTENTION'] = True + logger.info('(%s) Call not routed for subscriber %s, call route in progress on target: HBSystem: %s, TS: %s, TGID: %s, SUB: %s', self._system, int_id(_rf_src), _target, _slot, int_id(_target_status[_slot]['TX_TGID']), int_id(_target_status[_slot]['TX_RFS'])) + continue + + # Record target information if this is a new call stream? + if (_stream_id not in self.STATUS): + # Record the DST TGID and Stream ID + _target_status[_slot]['TX_START'] = pkt_time + _target_status[_slot]['TX_TGID'] = _dst_id + _target_status[_slot]['TX_STREAM_ID'] = _stream_id + _target_status[_slot]['TX_RFS'] = _rf_src + _target_status[_slot]['TX_PEER'] = _peer_id + + logger.info('(%s) Unit call bridged to HBP System: %s TS: %s, UNIT: %s', self._system, _target, _slot, int_id(_dst_id)) + if CONFIG['REPORTS']['REPORT']: + systems[_target]._report.send_bridgeEvent('UNIT VOICE,START,TX,{},{},{},{},{},{}'.format(_target, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id)).encode(encoding='utf-8', errors='ignore')) + + # Set other values for the contention handler to test next time there is a frame to forward + _target_status[_slot]['TX_TIME'] = pkt_time + _target_status[_slot]['TX_TYPE'] = _dtype_vseq + + #send the call: + systems[_target].send_system(_data) + + if _target_system['MODE'] == 'OPENBRIDGE': + if (_frame_type == HBPF_DATA_SYNC) and (_dtype_vseq == HBPF_SLT_VTERM): + if (_stream_id in _target_status): + _target_status.pop(_stream_id) + + + # Final actions - Is this a voice terminator? + if (_frame_type == HBPF_DATA_SYNC) and (_dtype_vseq == HBPF_SLT_VTERM): + self._targets = [] + call_duration = pkt_time - self.STATUS[_stream_id]['START'] + logger.info('(%s) *UNIT CALL END* STREAM ID: %s SUB: %s (%s) PEER: %s (%s) UNIT %s (%s), TS %s, Duration: %.2f', \ + self._system, int_id(_stream_id), get_alias(_rf_src, subscriber_ids), int_id(_rf_src), get_alias(_peer_id, peer_ids), int_id(_peer_id), get_alias(_dst_id, talkgroup_ids), int_id(_dst_id), _slot, call_duration) + if CONFIG['REPORTS']['REPORT']: + self._report.send_bridgeEvent('UNIT VOICE,END,RX,{},{},{},{},{},{},{:.2f}'.format(self._system, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id), call_duration).encode(encoding='utf-8', errors='ignore')) + + + def dmrd_received(self, _peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data): if _call_type == 'group': - # Is this a new call stream? - if (_stream_id not in self.STATUS): - # This is a new call stream - self.STATUS[_stream_id] = { - 'START': pkt_time, - 'CONTENTION':False, - 'RFS': _rf_src, - 'TGID': _dst_id, - } + self.group_received(_peer_id, _rf_src, _dst_id, _seq, _slot, _frame_type, _dtype_vseq, _stream_id, _data) + elif _call_type == 'unit': + self.unit_received(_peer_id, _rf_src, _dst_id, _seq, _slot, _frame_type, _dtype_vseq, _stream_id, _data) + elif _call_type == 'vscsbk': + logger.debug('CSBK recieved, but HBlink does not process them currently') + else: + logger.error('Unknown call type recieved -- not processed') - # 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) - 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 - - - logger.info('(%s) *CALL START* STREAM ID: %s SUB: %s (%s) PEER: %s (%s) TGID %s (%s), TS %s', \ - self._system, int_id(_stream_id), get_alias(_rf_src, subscriber_ids), int_id(_rf_src), get_alias(_peer_id, peer_ids), int_id(_peer_id), get_alias(_dst_id, talkgroup_ids), int_id(_dst_id), _slot) - if CONFIG['REPORTS']['REPORT']: - self._report.send_bridgeEvent('GROUP VOICE,START,RX,{},{},{},{},{},{}'.format(self._system, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id)).encode(encoding='utf-8', errors='ignore')) - - self.STATUS[_stream_id]['LAST'] = pkt_time - - - for _bridge in BRIDGES: - for _system in BRIDGES[_bridge]: - - if (_system['SYSTEM'] == self._system and _system['TGID'] == _dst_id and _system['TS'] == _slot and _system['ACTIVE'] == True): - - for _target in BRIDGES[_bridge]: - if (_target['SYSTEM'] != self._system) and (_target['ACTIVE']): - _target_status = systems[_target['SYSTEM']].STATUS - _target_system = self._CONFIG['SYSTEMS'][_target['SYSTEM']] - if _target_system['MODE'] == 'OPENBRIDGE': - # Is this a new call stream on the target? - if (_stream_id not in _target_status): - # This is a new call stream on the target - _target_status[_stream_id] = { - 'START': pkt_time, - 'CONTENTION':False, - 'RFS': _rf_src, - 'TGID': _dst_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[_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) - - 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']: - systems[_target['SYSTEM']]._report.send_bridgeEvent('GROUP VOICE,START,TX,{},{},{},{},{},{}'.format(_target['SYSTEM'], int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _target['TS'], int_id(_target['TGID'])).encode(encoding='utf-8', errors='ignore')) - - # Record the time of this packet so we can later identify a stale stream - _target_status[_stream_id]['LAST'] = pkt_time - # Clear the TS bit -- all OpenBridge streams are effectively on TS1 - _tmp_bits = _bits & ~(1 << 7) - - # Assemble transmit HBP packet header - _tmp_data = b''.join([_data[:8], _target['TGID'], _data[11:15], _tmp_bits.to_bytes(1, 'big'), _data[16:20]]) - - # MUST TEST FOR NEW STREAM AND IF SO, RE-WRITE THE LC FOR THE TARGET - # MUST RE-WRITE DESTINATION TGID IF DIFFERENT - # if _dst_id != rule['DST_GROUP']: - dmrbits = bitarray(endian='big') - dmrbits.frombytes(dmrpkt) - # Create a voice header packet (FULL LC) - if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD: - dmrbits = _target_status[_stream_id]['H_LC'][0:98] + dmrbits[98:166] + _target_status[_stream_id]['H_LC'][98:197] - # Create a voice terminator packet (FULL LC) - elif _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VTERM: - dmrbits = _target_status[_stream_id]['T_LC'][0:98] + dmrbits[98:166] + _target_status[_stream_id]['T_LC'][98:197] - if CONFIG['REPORTS']['REPORT']: - call_duration = pkt_time - _target_status[_stream_id]['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 _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]) - - else: - # BEGIN CONTENTION HANDLING - # - # The rules for each of the 4 "ifs" below are listed here for readability. The Frame To Send is: - # From a different group than last RX from this HBSystem, but it has been less than Group Hangtime - # From a different group than last TX to this HBSystem, but it has been less than Group Hangtime - # From the same group as the last RX from this HBSystem, but from a different subscriber, and it has been less than stream timeout - # From the same group as the last TX to this HBSystem, but from a different subscriber, and it has been less than stream timeout - # The "continue" at the end of each means the next iteration of the for loop that tests for matching rules - # - if ((_target['TGID'] != _target_status[_target['TS']]['RX_TGID']) and ((pkt_time - _target_status[_target['TS']]['RX_TIME']) < _target_system['GROUP_HANGTIME'])): - if self.STATUS[_stream_id]['CONTENTION'] == False: - self.STATUS[_stream_id]['CONTENTION'] = True - logger.info('(%s) Call not routed to TGID %s, target active or in group hangtime: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_target['TGID']), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['RX_TGID'])) - continue - if ((_target['TGID'] != _target_status[_target['TS']]['TX_TGID']) and ((pkt_time - _target_status[_target['TS']]['TX_TIME']) < _target_system['GROUP_HANGTIME'])): - if self.STATUS[_stream_id]['CONTENTION'] == False: - self.STATUS[_stream_id]['CONTENTION'] = True - logger.info('(%s) Call not routed to TGID%s, target in group hangtime: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_target['TGID']), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['TX_TGID'])) - continue - if (_target['TGID'] == _target_status[_target['TS']]['RX_TGID']) and ((pkt_time - _target_status[_target['TS']]['RX_TIME']) < STREAM_TO): - if self.STATUS[_stream_id]['CONTENTION'] == False: - self.STATUS[_stream_id]['CONTENTION'] = True - logger.info('(%s) Call not routed to TGID%s, matching call already active on target: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_target['TGID']), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['RX_TGID'])) - continue - if (_target['TGID'] == _target_status[_target['TS']]['TX_TGID']) and (_rf_src != _target_status[_target['TS']]['TX_RFS']) and ((pkt_time - _target_status[_target['TS']]['TX_TIME']) < STREAM_TO): - if self.STATUS[_stream_id]['CONTENTION'] == False: - self.STATUS[_stream_id]['CONTENTION'] = True - logger.info('(%s) Call not routed for subscriber %s, call route in progress on target: HBSystem: %s, TS: %s, TGID: %s, SUB: %s', self._system, int_id(_rf_src), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['TX_TGID']), int_id(_target_status[_target['TS']]['TX_RFS'])) - continue - - # Is this a new call stream? - if (_target_status[_target['TS']]['TX_STREAM_ID'] != _stream_id): - # Record the DST TGID and Stream ID - _target_status[_target['TS']]['TX_START'] = pkt_time - _target_status[_target['TS']]['TX_TGID'] = _target['TGID'] - _target_status[_target['TS']]['TX_STREAM_ID'] = _stream_id - _target_status[_target['TS']]['TX_RFS'] = _rf_src - _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) - 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']: - systems[_target['SYSTEM']]._report.send_bridgeEvent('GROUP VOICE,START,TX,{},{},{},{},{},{}'.format(_target['SYSTEM'], int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _target['TS'], int_id(_target['TGID'])).encode(encoding='utf-8', errors='ignore')) - - # Set other values for the contention handler to test next time there is a frame to forward - _target_status[_target['TS']]['TX_TIME'] = pkt_time - _target_status[_target['TS']]['TX_TYPE'] = _dtype_vseq - - # Handle any necessary re-writes for the destination - if _system['TS'] != _target['TS']: - _tmp_bits = _bits ^ 1 << 7 - else: - _tmp_bits = _bits - - # Assemble transmit HBP packet header - _tmp_data = b''.join([_data[:8], _target['TGID'], _data[11:15], _tmp_bits.to_bytes(1, 'big'), _data[16:20]]) - - # MUST TEST FOR NEW STREAM AND IF SO, RE-WRITE THE LC FOR THE TARGET - # MUST RE-WRITE DESTINATION TGID IF DIFFERENT - # if _dst_id != rule['DST_GROUP']: - dmrbits = bitarray(endian='big') - dmrbits.frombytes(dmrpkt) - # Create a voice header packet (FULL LC) - if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD: - dmrbits = _target_status[_target['TS']]['TX_H_LC'][0:98] + dmrbits[98:166] + _target_status[_target['TS']]['TX_H_LC'][98:197] - # Create a voice terminator packet (FULL LC) - elif _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VTERM: - dmrbits = _target_status[_target['TS']]['TX_T_LC'][0:98] + dmrbits[98:166] + _target_status[_target['TS']]['TX_T_LC'][98:197] - if CONFIG['REPORTS']['REPORT']: - 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 _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] - - # Transmit the packet to the destination system - systems[_target['SYSTEM']].send_system(_tmp_data) - #logger.debug('(%s) Packet routed by bridge: %s to system: %s TS: %s, TGID: %s', self._system, _bridge, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) - - - - # Final actions - Is this a voice terminator? - if (_frame_type == HBPF_DATA_SYNC) and (_dtype_vseq == HBPF_SLT_VTERM): - call_duration = pkt_time - self.STATUS[_stream_id]['START'] - logger.info('(%s) *CALL END* STREAM ID: %s SUB: %s (%s) PEER: %s (%s) TGID %s (%s), TS %s, Duration: %.2f', \ - self._system, int_id(_stream_id), get_alias(_rf_src, subscriber_ids), int_id(_rf_src), get_alias(_peer_id, peer_ids), int_id(_peer_id), get_alias(_dst_id, talkgroup_ids), int_id(_dst_id), _slot, call_duration) - if CONFIG['REPORTS']['REPORT']: - self._report.send_bridgeEvent('GROUP VOICE,END,RX,{},{},{},{},{},{},{:.2f}'.format(self._system, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id), call_duration).encode(encoding='utf-8', errors='ignore')) - removed = self.STATUS.pop(_stream_id) - logger.debug('(%s) OpenBridge sourced call stream end, remove terminated Stream ID: %s', self._system, int_id(_stream_id)) - if not removed: - selflogger.error('(%s) *CALL END* STREAM ID: %s NOT IN LIST -- THIS IS A REAL PROBLEM', self._system, int_id(_stream_id)) class routerHBP(HBSYSTEM): def __init__(self, _name, _config, _report): HBSYSTEM.__init__(self, _name, _config, _report) + self.name = _name + + # list of self._targets for unit (subscriber, private) calls + self._targets = [] # Status information for the system, TS1 & TS2 # 1 & 2 are "timeslot" @@ -456,252 +648,412 @@ class routerHBP(HBSYSTEM): } } - def dmrd_received(self, _peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data): + + def group_received(self, _peer_id, _rf_src, _dst_id, _seq, _slot, _frame_type, _dtype_vseq, _stream_id, _data): + global UNIT_MAP pkt_time = time() dmrpkt = _data[20:53] _bits = _data[15] + + # Make/update an entry in the UNIT_MAP for this subscriber + UNIT_MAP[_rf_src] = (self.name, pkt_time) - if _call_type == 'group': + # Is this a new call stream? + if (_stream_id != self.STATUS[_slot]['RX_STREAM_ID']): + if (self.STATUS[_slot]['RX_TYPE'] != HBPF_SLT_VTERM) and (pkt_time < (self.STATUS[_slot]['RX_TIME'] + STREAM_TO)) and (_rf_src != self.STATUS[_slot]['RX_RFS']): + logger.warning('(%s) Packet received with STREAM ID: %s SUB: %s PEER: %s TGID %s, SLOT %s collided with existing call', self._system, int_id(_stream_id), int_id(_rf_src), int_id(_peer_id), int_id(_dst_id), _slot) + return - # Is this a new call stream? - if (_stream_id != self.STATUS[_slot]['RX_STREAM_ID']): - if (self.STATUS[_slot]['RX_TYPE'] != HBPF_SLT_VTERM) and (pkt_time < (self.STATUS[_slot]['RX_TIME'] + STREAM_TO)) and (_rf_src != self.STATUS[_slot]['RX_RFS']): - logger.warning('(%s) Packet received with STREAM ID: %s SUB: %s PEER: %s TGID %s, SLOT %s collided with existing call', self._system, int_id(_stream_id), int_id(_rf_src), int_id(_peer_id), int_id(_dst_id), _slot) - return + # This is a new call stream + self.STATUS[_slot]['RX_START'] = pkt_time + logger.info('(%s) *GROUP CALL START* STREAM ID: %s SUB: %s (%s) PEER: %s (%s) TGID %s (%s), TS %s', \ + self._system, int_id(_stream_id), get_alias(_rf_src, subscriber_ids), int_id(_rf_src), get_alias(_peer_id, peer_ids), int_id(_peer_id), get_alias(_dst_id, talkgroup_ids), int_id(_dst_id), _slot) + if CONFIG['REPORTS']['REPORT']: + self._report.send_bridgeEvent('GROUP VOICE,START,RX,{},{},{},{},{},{}'.format(self._system, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id)).encode(encoding='utf-8', errors='ignore')) - # This is a new call stream - self.STATUS[_slot]['RX_START'] = pkt_time - logger.info('(%s) *CALL START* STREAM ID: %s SUB: %s (%s) PEER: %s (%s) TGID %s (%s), TS %s', \ - self._system, int_id(_stream_id), get_alias(_rf_src, subscriber_ids), int_id(_rf_src), get_alias(_peer_id, peer_ids), int_id(_peer_id), get_alias(_dst_id, talkgroup_ids), int_id(_dst_id), _slot) - if CONFIG['REPORTS']['REPORT']: - self._report.send_bridgeEvent('GROUP VOICE,START,RX,{},{},{},{},{},{}'.format(self._system, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id)).encode(encoding='utf-8', errors='ignore')) + # 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) + self.STATUS[_slot]['RX_LC'] = decoded['LC'] - # 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) - 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 - # 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 + for _bridge in BRIDGES: + for _system in BRIDGES[_bridge]: + + if (_system['SYSTEM'] == self._system and _system['TGID'] == _dst_id and _system['TS'] == _slot and _system['ACTIVE'] == True): + + for _target in BRIDGES[_bridge]: + if _target['SYSTEM'] != self._system: + if _target['ACTIVE']: + _target_status = systems[_target['SYSTEM']].STATUS + _target_system = self._CONFIG['SYSTEMS'][_target['SYSTEM']] + + if _target_system['MODE'] == 'OPENBRIDGE': + # Is this a new call stream on the target? + if (_stream_id not in _target_status): + # This is a new call stream on the target + _target_status[_stream_id] = { + 'START': pkt_time, + 'CONTENTION':False, + 'RFS': _rf_src, + 'TYPE': 'GROUP', + 'DST': _dst_id, + 'ACTIVE': True, + } + # 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) + + 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']: + systems[_target['SYSTEM']]._report.send_bridgeEvent('GROUP VOICE,START,TX,{},{},{},{},{},{}'.format(_target['SYSTEM'], int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _target['TS'], int_id(_target['TGID'])).encode(encoding='utf-8', errors='ignore')) + + # Record the time of this packet so we can later identify a stale stream + _target_status[_stream_id]['LAST'] = pkt_time + # Clear the TS bit -- all OpenBridge streams are effectively on TS1 + _tmp_bits = _bits & ~(1 << 7) + + # Assemble transmit HBP packet header + _tmp_data = b''.join([_data[:8], _target['TGID'], _data[11:15], _tmp_bits.to_bytes(1, 'big'), _data[16:20]]) + + # MUST TEST FOR NEW STREAM AND IF SO, RE-WRITE THE LC FOR THE TARGET + # MUST RE-WRITE DESTINATION TGID IF DIFFERENT + # if _dst_id != rule['DST_GROUP']: + dmrbits = bitarray(endian='big') + dmrbits.frombytes(dmrpkt) + # Create a voice header packet (FULL LC) + if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD: + dmrbits = _target_status[_stream_id]['H_LC'][0:98] + dmrbits[98:166] + _target_status[_stream_id]['H_LC'][98:197] + # Create a voice terminator packet (FULL LC) + elif _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VTERM: + dmrbits = _target_status[_stream_id]['T_LC'][0:98] + dmrbits[98:166] + _target_status[_stream_id]['T_LC'][98:197] + if CONFIG['REPORTS']['REPORT']: + call_duration = pkt_time - _target_status[_stream_id]['START'] + _target_status[_stream_id]['ACTIVE'] = False + 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 _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]) + + else: + # BEGIN STANDARD CONTENTION HANDLING + # + # The rules for each of the 4 "ifs" below are listed here for readability. The Frame To Send is: + # From a different group than last RX from this HBSystem, but it has been less than Group Hangtime + # From a different group than last TX to this HBSystem, but it has been less than Group Hangtime + # From the same group as the last RX from this HBSystem, but from a different subscriber, and it has been less than stream timeout + # From the same group as the last TX to this HBSystem, but from a different subscriber, and it has been less than stream timeout + # The "continue" at the end of each means the next iteration of the for loop that tests for matching rules + # + if ((_target['TGID'] != _target_status[_target['TS']]['RX_TGID']) and ((pkt_time - _target_status[_target['TS']]['RX_TIME']) < _target_system['GROUP_HANGTIME'])): + if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD and self.STATUS[_slot]['RX_STREAM_ID'] != _stream_id: + logger.info('(%s) Call not routed to TGID %s, target active or in group hangtime: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_target['TGID']), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['RX_TGID'])) + continue + if ((_target['TGID'] != _target_status[_target['TS']]['TX_TGID']) and ((pkt_time - _target_status[_target['TS']]['TX_TIME']) < _target_system['GROUP_HANGTIME'])): + if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD and self.STATUS[_slot]['RX_STREAM_ID'] != _stream_id: + logger.info('(%s) Call not routed to TGID%s, target in group hangtime: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_target['TGID']), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['TX_TGID'])) + continue + if (_target['TGID'] == _target_status[_target['TS']]['RX_TGID']) and ((pkt_time - _target_status[_target['TS']]['RX_TIME']) < STREAM_TO): + if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD and self.STATUS[_slot]['RX_STREAM_ID'] != _stream_id: + logger.info('(%s) Call not routed to TGID%s, matching call already active on target: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_target['TGID']), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['RX_TGID'])) + continue + if (_target['TGID'] == _target_status[_target['TS']]['TX_TGID']) and (_rf_src != _target_status[_target['TS']]['TX_RFS']) and ((pkt_time - _target_status[_target['TS']]['TX_TIME']) < STREAM_TO): + if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD and self.STATUS[_slot]['RX_STREAM_ID'] != _stream_id: + logger.info('(%s) Call not routed for subscriber %s, call route in progress on target: HBSystem: %s, TS: %s, TGID: %s, SUB: %s', self._system, int_id(_rf_src), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['TX_TGID']), int_id(_target_status[_target['TS']]['TX_RFS'])) + continue + + # Is this a new call stream? + if (_stream_id != self.STATUS[_slot]['RX_STREAM_ID']): + # Record the DST TGID and Stream ID + _target_status[_target['TS']]['TX_START'] = pkt_time + _target_status[_target['TS']]['TX_TGID'] = _target['TGID'] + _target_status[_target['TS']]['TX_STREAM_ID'] = _stream_id + _target_status[_target['TS']]['TX_RFS'] = _rf_src + _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) + 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']: + systems[_target['SYSTEM']]._report.send_bridgeEvent('GROUP VOICE,START,TX,{},{},{},{},{},{}'.format(_target['SYSTEM'], int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _target['TS'], int_id(_target['TGID'])).encode(encoding='utf-8', errors='ignore')) + + # Set other values for the contention handler to test next time there is a frame to forward + _target_status[_target['TS']]['TX_TIME'] = pkt_time + _target_status[_target['TS']]['TX_TYPE'] = _dtype_vseq + + # Handle any necessary re-writes for the destination + if _system['TS'] != _target['TS']: + _tmp_bits = _bits ^ 1 << 7 + else: + _tmp_bits = _bits + + # Assemble transmit HBP packet header + _tmp_data = b''.join([_data[:8], _target['TGID'], _data[11:15], _tmp_bits.to_bytes(1, 'big'), _data[16:20]]) + + dmrbits = bitarray(endian='big') + dmrbits.frombytes(dmrpkt) + # Create a voice header packet (FULL LC) + if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD: + dmrbits = _target_status[_target['TS']]['TX_H_LC'][0:98] + dmrbits[98:166] + _target_status[_target['TS']]['TX_H_LC'][98:197] + # Create a voice terminator packet (FULL LC) + elif _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VTERM: + dmrbits = _target_status[_target['TS']]['TX_T_LC'][0:98] + dmrbits[98:166] + _target_status[_target['TS']]['TX_T_LC'][98:197] + if CONFIG['REPORTS']['REPORT']: + 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 _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, _data[53:55]]) + + # Transmit the packet to the destination system + systems[_target['SYSTEM']].send_system(_tmp_data) + #logger.debug('(%s) Packet routed by bridge: %s to system: %s TS: %s, TGID: %s', self._system, _bridge, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) + + if _target_system['MODE'] == 'OPENBRIDGE': + if (_frame_type == HBPF_DATA_SYNC) and (_dtype_vseq == HBPF_SLT_VTERM) and (self.STATUS[_slot]['RX_TYPE'] != HBPF_SLT_VTERM): + if (_stream_id in _target_status): + _target_status.pop(_stream_id) + + + # Final actions - Is this a voice terminator? + if (_frame_type == HBPF_DATA_SYNC) and (_dtype_vseq == HBPF_SLT_VTERM) and (self.STATUS[_slot]['RX_TYPE'] != HBPF_SLT_VTERM): + call_duration = pkt_time - self.STATUS[_slot]['RX_START'] + logger.info('(%s) *GROUP CALL END* STREAM ID: %s SUB: %s (%s) PEER: %s (%s) TGID %s (%s), TS %s, Duration: %.2f', \ + self._system, int_id(_stream_id), get_alias(_rf_src, subscriber_ids), int_id(_rf_src), get_alias(_peer_id, peer_ids), int_id(_peer_id), get_alias(_dst_id, talkgroup_ids), int_id(_dst_id), _slot, call_duration) + if CONFIG['REPORTS']['REPORT']: + self._report.send_bridgeEvent('GROUP VOICE,END,RX,{},{},{},{},{},{},{:.2f}'.format(self._system, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id), call_duration).encode(encoding='utf-8', errors='ignore')) + + # + # Begin in-band signalling for call end. This has nothign to do with routing traffic directly. + # + + # Iterate the rules dictionary for _bridge in BRIDGES: for _system in BRIDGES[_bridge]: + if _system['SYSTEM'] == self._system: - if (_system['SYSTEM'] == self._system and _system['TGID'] == _dst_id and _system['TS'] == _slot and _system['ACTIVE'] == True): + # TGID matches a rule source, reset its timer + if _slot == _system['TS'] and _dst_id == _system['TGID'] and ((_system['TO_TYPE'] == 'ON' and (_system['ACTIVE'] == True)) or (_system['TO_TYPE'] == 'OFF' and _system['ACTIVE'] == False)): + _system['TIMER'] = pkt_time + _system['TIMEOUT'] + logger.info('(%s) Transmission match for Bridge: %s. Reset timeout to %s', self._system, _bridge, _system['TIMER']) - for _target in BRIDGES[_bridge]: - if _target['SYSTEM'] != self._system: - if _target['ACTIVE']: - _target_status = systems[_target['SYSTEM']].STATUS - _target_system = self._CONFIG['SYSTEMS'][_target['SYSTEM']] - - if _target_system['MODE'] == 'OPENBRIDGE': - # Is this a new call stream on the target? - if (_stream_id not in _target_status): - # This is a new call stream on the target - _target_status[_stream_id] = { - 'START': pkt_time, - 'CONTENTION':False, - 'RFS': _rf_src, - 'TGID': _dst_id, - } - # 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) - - 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']: - systems[_target['SYSTEM']]._report.send_bridgeEvent('GROUP VOICE,START,TX,{},{},{},{},{},{}'.format(_target['SYSTEM'], int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _target['TS'], int_id(_target['TGID'])).encode(encoding='utf-8', errors='ignore')) - - # Record the time of this packet so we can later identify a stale stream - _target_status[_stream_id]['LAST'] = pkt_time - # Clear the TS bit -- all OpenBridge streams are effectively on TS1 - _tmp_bits = _bits & ~(1 << 7) - - # Assemble transmit HBP packet header - _tmp_data = b''.join([_data[:8], _target['TGID'], _data[11:15], _tmp_bits.to_bytes(1, 'big'), _data[16:20]]) - - # MUST TEST FOR NEW STREAM AND IF SO, RE-WRITE THE LC FOR THE TARGET - # MUST RE-WRITE DESTINATION TGID IF DIFFERENT - # if _dst_id != rule['DST_GROUP']: - dmrbits = bitarray(endian='big') - dmrbits.frombytes(dmrpkt) - # Create a voice header packet (FULL LC) - if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD: - dmrbits = _target_status[_stream_id]['H_LC'][0:98] + dmrbits[98:166] + _target_status[_stream_id]['H_LC'][98:197] - # Create a voice terminator packet (FULL LC) - elif _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VTERM: - dmrbits = _target_status[_stream_id]['T_LC'][0:98] + dmrbits[98:166] + _target_status[_stream_id]['T_LC'][98:197] - if CONFIG['REPORTS']['REPORT']: - call_duration = pkt_time - _target_status[_stream_id]['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 _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]) - - else: - # BEGIN STANDARD CONTENTION HANDLING - # - # The rules for each of the 4 "ifs" below are listed here for readability. The Frame To Send is: - # From a different group than last RX from this HBSystem, but it has been less than Group Hangtime - # From a different group than last TX to this HBSystem, but it has been less than Group Hangtime - # From the same group as the last RX from this HBSystem, but from a different subscriber, and it has been less than stream timeout - # From the same group as the last TX to this HBSystem, but from a different subscriber, and it has been less than stream timeout - # The "continue" at the end of each means the next iteration of the for loop that tests for matching rules - # - if ((_target['TGID'] != _target_status[_target['TS']]['RX_TGID']) and ((pkt_time - _target_status[_target['TS']]['RX_TIME']) < _target_system['GROUP_HANGTIME'])): - if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD and self.STATUS[_slot]['RX_STREAM_ID'] != _stream_id: - logger.info('(%s) Call not routed to TGID %s, target active or in group hangtime: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_target['TGID']), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['RX_TGID'])) - continue - if ((_target['TGID'] != _target_status[_target['TS']]['TX_TGID']) and ((pkt_time - _target_status[_target['TS']]['TX_TIME']) < _target_system['GROUP_HANGTIME'])): - if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD and self.STATUS[_slot]['RX_STREAM_ID'] != _stream_id: - logger.info('(%s) Call not routed to TGID%s, target in group hangtime: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_target['TGID']), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['TX_TGID'])) - continue - if (_target['TGID'] == _target_status[_target['TS']]['RX_TGID']) and ((pkt_time - _target_status[_target['TS']]['RX_TIME']) < STREAM_TO): - if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD and self.STATUS[_slot]['RX_STREAM_ID'] != _stream_id: - logger.info('(%s) Call not routed to TGID%s, matching call already active on target: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_target['TGID']), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['RX_TGID'])) - continue - if (_target['TGID'] == _target_status[_target['TS']]['TX_TGID']) and (_rf_src != _target_status[_target['TS']]['TX_RFS']) and ((pkt_time - _target_status[_target['TS']]['TX_TIME']) < STREAM_TO): - if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD and self.STATUS[_slot]['RX_STREAM_ID'] != _stream_id: - logger.info('(%s) Call not routed for subscriber %s, call route in progress on target: HBSystem: %s, TS: %s, TGID: %s, SUB: %s', self._system, int_id(_rf_src), _target['SYSTEM'], _target['TS'], int_id(_target_status[_target['TS']]['TX_TGID']), int_id(_target_status[_target['TS']]['TX_RFS'])) - continue - - # Is this a new call stream? - if (_stream_id != self.STATUS[_slot]['RX_STREAM_ID']): - # Record the DST TGID and Stream ID - _target_status[_target['TS']]['TX_START'] = pkt_time - _target_status[_target['TS']]['TX_TGID'] = _target['TGID'] - _target_status[_target['TS']]['TX_STREAM_ID'] = _stream_id - _target_status[_target['TS']]['TX_RFS'] = _rf_src - _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) - 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']: - systems[_target['SYSTEM']]._report.send_bridgeEvent('GROUP VOICE,START,TX,{},{},{},{},{},{}'.format(_target['SYSTEM'], int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _target['TS'], int_id(_target['TGID'])).encode(encoding='utf-8', errors='ignore')) - - # Set other values for the contention handler to test next time there is a frame to forward - _target_status[_target['TS']]['TX_TIME'] = pkt_time - _target_status[_target['TS']]['TX_TYPE'] = _dtype_vseq - - # Handle any necessary re-writes for the destination - if _system['TS'] != _target['TS']: - _tmp_bits = _bits ^ 1 << 7 - else: - _tmp_bits = _bits - - # Assemble transmit HBP packet header - _tmp_data = b''.join([_data[:8], _target['TGID'], _data[11:15], _tmp_bits.to_bytes(1, 'big'), _data[16:20]]) - - # MUST TEST FOR NEW STREAM AND IF SO, RE-WRITE THE LC FOR THE TARGET - # MUST RE-WRITE DESTINATION TGID IF DIFFERENT - # if _dst_id != rule['DST_GROUP']: - dmrbits = bitarray(endian='big') - dmrbits.frombytes(dmrpkt) - # Create a voice header packet (FULL LC) - if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD: - dmrbits = _target_status[_target['TS']]['TX_H_LC'][0:98] + dmrbits[98:166] + _target_status[_target['TS']]['TX_H_LC'][98:197] - # Create a voice terminator packet (FULL LC) - elif _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VTERM: - dmrbits = _target_status[_target['TS']]['TX_T_LC'][0:98] + dmrbits[98:166] + _target_status[_target['TS']]['TX_T_LC'][98:197] - if CONFIG['REPORTS']['REPORT']: - 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 _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, _data[53:55]]) - - # Transmit the packet to the destination system - systems[_target['SYSTEM']].send_system(_tmp_data) - #logger.debug('(%s) Packet routed by bridge: %s to system: %s TS: %s, TGID: %s', self._system, _bridge, _target['SYSTEM'], _target['TS'], int_id(_target['TGID'])) - - - - # Final actions - Is this a voice terminator? - if (_frame_type == HBPF_DATA_SYNC) and (_dtype_vseq == HBPF_SLT_VTERM) and (self.STATUS[_slot]['RX_TYPE'] != HBPF_SLT_VTERM): - call_duration = pkt_time - self.STATUS[_slot]['RX_START'] - logger.info('(%s) *CALL END* STREAM ID: %s SUB: %s (%s) PEER: %s (%s) TGID %s (%s), TS %s, Duration: %.2f', \ - self._system, int_id(_stream_id), get_alias(_rf_src, subscriber_ids), int_id(_rf_src), get_alias(_peer_id, peer_ids), int_id(_peer_id), get_alias(_dst_id, talkgroup_ids), int_id(_dst_id), _slot, call_duration) - if CONFIG['REPORTS']['REPORT']: - self._report.send_bridgeEvent('GROUP VOICE,END,RX,{},{},{},{},{},{},{:.2f}'.format(self._system, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id), call_duration).encode(encoding='utf-8', errors='ignore')) - - # - # Begin in-band signalling for call end. This has nothign to do with routing traffic directly. - # - - # Iterate the rules dictionary - - for _bridge in BRIDGES: - for _system in BRIDGES[_bridge]: - if _system['SYSTEM'] == self._system: - - # TGID matches a rule source, reset its timer - if _slot == _system['TS'] and _dst_id == _system['TGID'] and ((_system['TO_TYPE'] == 'ON' and (_system['ACTIVE'] == True)) or (_system['TO_TYPE'] == 'OFF' and _system['ACTIVE'] == False)): + # TGID matches an ACTIVATION trigger + if (_dst_id in _system['ON'] or _dst_id in _system['RESET']) and _slot == _system['TS']: + # Set the matching rule as ACTIVE + if _dst_id in _system['ON']: + if _system['ACTIVE'] == False: + _system['ACTIVE'] = True + _system['TIMER'] = pkt_time + _system['TIMEOUT'] + logger.info('(%s) Bridge: %s, connection changed to state: %s', self._system, _bridge, _system['ACTIVE']) + # Cancel the timer if we've enabled an "OFF" type timeout + if _system['TO_TYPE'] == 'OFF': + _system['TIMER'] = pkt_time + logger.info('(%s) Bridge: %s set to "OFF" with an on timer rule: timeout timer cancelled', self._system, _bridge) + # Reset the timer for the rule + if _system['ACTIVE'] == True and _system['TO_TYPE'] == 'ON': _system['TIMER'] = pkt_time + _system['TIMEOUT'] - logger.info('(%s) Transmission match for Bridge: %s. Reset timeout to %s', self._system, _bridge, _system['TIMER']) + logger.info('(%s) Bridge: %s, timeout timer reset to: %s', self._system, _bridge, _system['TIMER'] - pkt_time) - # TGID matches an ACTIVATION trigger - if (_dst_id in _system['ON'] or _dst_id in _system['RESET']) and _slot == _system['TS']: - # Set the matching rule as ACTIVE - if _dst_id in _system['ON']: - if _system['ACTIVE'] == False: - _system['ACTIVE'] = True - _system['TIMER'] = pkt_time + _system['TIMEOUT'] - logger.info('(%s) Bridge: %s, connection changed to state: %s', self._system, _bridge, _system['ACTIVE']) - # Cancel the timer if we've enabled an "OFF" type timeout - if _system['TO_TYPE'] == 'OFF': - _system['TIMER'] = pkt_time - logger.info('(%s) Bridge: %s set to "OFF" with an on timer rule: timeout timer cancelled', self._system, _bridge) - # Reset the timer for the rule - if _system['ACTIVE'] == True and _system['TO_TYPE'] == 'ON': - _system['TIMER'] = pkt_time + _system['TIMEOUT'] - logger.info('(%s) Bridge: %s, timeout timer reset to: %s', self._system, _bridge, _system['TIMER'] - pkt_time) + # TGID matches an DE-ACTIVATION trigger + if (_dst_id in _system['OFF'] or _dst_id in _system['RESET']) and _slot == _system['TS']: + # Set the matching rule as ACTIVE + if _dst_id in _system['OFF']: + if _system['ACTIVE'] == True: + _system['ACTIVE'] = False + logger.info('(%s) Bridge: %s, connection changed to state: %s', self._system, _bridge, _system['ACTIVE']) + # Cancel the timer if we've enabled an "ON" type timeout + if _system['TO_TYPE'] == 'ON': + _system['TIMER'] = pkt_time + logger.info('(%s) Bridge: %s set to ON with and "OFF" timer rule: timeout timer cancelled', self._system, _bridge) + # Reset the timer for the rule + if _system['ACTIVE'] == False and _system['TO_TYPE'] == 'OFF': + _system['TIMER'] = pkt_time + _system['TIMEOUT'] + logger.info('(%s) Bridge: %s, timeout timer reset to: %s', self._system, _bridge, _system['TIMER'] - pkt_time) + # Cancel the timer if we've enabled an "ON" type timeout + if _system['ACTIVE'] == True and _system['TO_TYPE'] == 'ON' and _dst_group in _system['OFF']: + _system['TIMER'] = pkt_time + logger.info('(%s) Bridge: %s set to ON with and "OFF" timer rule: timeout timer cancelled', self._system, _bridge) - # TGID matches an DE-ACTIVATION trigger - if (_dst_id in _system['OFF'] or _dst_id in _system['RESET']) and _slot == _system['TS']: - # Set the matching rule as ACTIVE - if _dst_id in _system['OFF']: - if _system['ACTIVE'] == True: - _system['ACTIVE'] = False - logger.info('(%s) Bridge: %s, connection changed to state: %s', self._system, _bridge, _system['ACTIVE']) - # Cancel the timer if we've enabled an "ON" type timeout - if _system['TO_TYPE'] == 'ON': - _system['TIMER'] = pkt_time - logger.info('(%s) Bridge: %s set to ON with and "OFF" timer rule: timeout timer cancelled', self._system, _bridge) - # Reset the timer for the rule - if _system['ACTIVE'] == False and _system['TO_TYPE'] == 'OFF': - _system['TIMER'] = pkt_time + _system['TIMEOUT'] - logger.info('(%s) Bridge: %s, timeout timer reset to: %s', self._system, _bridge, _system['TIMER'] - pkt_time) - # Cancel the timer if we've enabled an "ON" type timeout - if _system['ACTIVE'] == True and _system['TO_TYPE'] == 'ON' and _dst_group in _system['OFF']: - _system['TIMER'] = pkt_time - logger.info('(%s) Bridge: %s set to ON with and "OFF" timer rule: timeout timer cancelled', self._system, _bridge) - - # - # END IN-BAND SIGNALLING - # + # + # END IN-BAND SIGNALLING + # + # Mark status variables for use later + self.STATUS[_slot]['RX_PEER'] = _peer_id + self.STATUS[_slot]['RX_SEQ'] = _seq + self.STATUS[_slot]['RX_RFS'] = _rf_src + self.STATUS[_slot]['RX_TYPE'] = _dtype_vseq + self.STATUS[_slot]['RX_TGID'] = _dst_id + self.STATUS[_slot]['RX_TIME'] = pkt_time + self.STATUS[_slot]['RX_STREAM_ID'] = _stream_id - # Mark status variables for use later - self.STATUS[_slot]['RX_PEER'] = _peer_id - self.STATUS[_slot]['RX_SEQ'] = _seq - self.STATUS[_slot]['RX_RFS'] = _rf_src - self.STATUS[_slot]['RX_TYPE'] = _dtype_vseq - self.STATUS[_slot]['RX_TGID'] = _dst_id - self.STATUS[_slot]['RX_TIME'] = pkt_time - self.STATUS[_slot]['RX_STREAM_ID'] = _stream_id + def unit_received(self, _peer_id, _rf_src, _dst_id, _seq, _slot, _frame_type, _dtype_vseq, _stream_id, _data): + global UNIT_MAP + pkt_time = time() + dmrpkt = _data[20:53] + _bits = _data[15] + + # Make/update this unit in the UNIT_MAP cache + UNIT_MAP[_rf_src] = (self.name, pkt_time) + + + # Is this a new call stream? + if (_stream_id != self.STATUS[_slot]['RX_STREAM_ID']): + + # Collision in progress, bail out! + if (self.STATUS[_slot]['RX_TYPE'] != HBPF_SLT_VTERM) and (pkt_time < (self.STATUS[_slot]['RX_TIME'] + STREAM_TO)) and (_rf_src != self.STATUS[_slot]['RX_RFS']): + logger.warning('(%s) Packet received with STREAM ID: %s SUB: %s PEER: %s UNIT %s, SLOT %s collided with existing call', self._system, int_id(_stream_id), int_id(_rf_src), int_id(_peer_id), int_id(_dst_id), _slot) + return + + # Create a destination list for the call: + if _dst_id in UNIT_MAP: + if UNIT_MAP[_dst_id][0] != self._system: + self._targets = [UNIT_MAP[_dst_id][0]] + else: + self._targets = [] + logger.error('UNIT call to a subscriber on the same system, send nothing') + else: + self._targets = list(UNIT) + self._targets.remove(self._system) + + # This is a new call stream, so log & report + self.STATUS[_slot]['RX_START'] = pkt_time + logger.info('(%s) *UNIT CALL START* STREAM ID: %s SUB: %s (%s) PEER: %s (%s) UNIT: %s (%s), TS: %s, FORWARD: %s', \ + self._system, int_id(_stream_id), get_alias(_rf_src, subscriber_ids), int_id(_rf_src), get_alias(_peer_id, peer_ids), int_id(_peer_id), get_alias(_dst_id, talkgroup_ids), int_id(_dst_id), _slot, self._targets) + if CONFIG['REPORTS']['REPORT']: + self._report.send_bridgeEvent('UNIT VOICE,START,RX,{},{},{},{},{},{},{}'.format(self._system, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id), self._targets).encode(encoding='utf-8', errors='ignore')) + + for _target in self._targets: + + _target_status = systems[_target].STATUS + _target_system = self._CONFIG['SYSTEMS'][_target] + + if self._CONFIG['SYSTEMS'][_target]['MODE'] == 'OPENBRIDGE': + if (_stream_id not in _target_status): + # This is a new call stream on the target + _target_status[_stream_id] = { + 'START': pkt_time, + 'CONTENTION':False, + 'RFS': _rf_src, + 'TYPE': 'UNIT', + 'DST': _dst_id, + 'ACTIVE': True + } + + logger.info('(%s) Unit call bridged to OBP System: %s TS: %s, UNIT: %s', self._system, _target, _slot if _target_system['BOTH_SLOTS'] else 1, int_id(_dst_id)) + if CONFIG['REPORTS']['REPORT']: + systems[_target]._report.send_bridgeEvent('UNIT VOICE,START,TX,{},{},{},{},{},{}'.format(_target, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id)).encode(encoding='utf-8', errors='ignore')) + + # Record the time of this packet so we can later identify a stale stream + _target_status[_stream_id]['LAST'] = pkt_time + # Clear the TS bit and follow propper OBP definition, unless "BOTH_SLOTS" is set. This only works for unit calls. + if _target_system['BOTH_SLOTS']: + _tmp_bits = _bits + else: + _tmp_bits = _bits & ~(1 << 7) + + # Assemble transmit HBP packet + _tmp_data = b''.join([_data[:15], _tmp_bits.to_bytes(1, 'big'), _data[16:20]]) + _data = b''.join([_tmp_data, dmrpkt]) + + if (_frame_type == HBPF_DATA_SYNC) and (_dtype_vseq == HBPF_SLT_VTERM): + _target_status[_stream_id]['ACTIVE'] = False + + else: + # BEGIN STANDARD CONTENTION HANDLING + # + # The rules for each of the 4 "ifs" below are listed here for readability. The Frame To Send is: + # From a different group than last RX from this HBSystem, but it has been less than Group Hangtime + # From a different group than last TX to this HBSystem, but it has been less than Group Hangtime + # From the same group as the last RX from this HBSystem, but from a different subscriber, and it has been less than stream timeout + # From the same group as the last TX to this HBSystem, but from a different subscriber, and it has been less than stream timeout + # The "continue" at the end of each means the next iteration of the for loop that tests for matching rules + # + ''' + if ((_dst_id != _target_status[_slot]['RX_TGID']) and ((pkt_time - _target_status[_slot]['RX_TIME']) < _target_system['GROUP_HANGTIME'])): + if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD and self.STATUS[_slot]['RX_STREAM_ID'] != _stream_id: + logger.info('(%s) Call not routed to destination %s, target active or in group hangtime: HBSystem: %s, TS: %s, DEST: %s', self._system, int_id(_dst_id), _target, _slot, int_id(_target_status[_slot]['RX_TGID'])) + continue + if ((_dst_id != _target_status[_slot]['TX_TGID']) and ((pkt_time - _target_status[_slot]['TX_TIME']) < _target_system['GROUP_HANGTIME'])): + if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD and self.STATUS[_slot]['RX_STREAM_ID'] != _stream_id: + logger.info('(%s) Call not routed to destination %s, target in group hangtime: HBSystem: %s, TS: %s, DEST: %s', self._system, int_id(_dst_id), _target, _slot, int_id(_target_status[_slot]['TX_TGID'])) + continue + ''' + if (_dst_id == _target_status[_slot]['RX_TGID']) and ((pkt_time - _target_status[_slot]['RX_TIME']) < STREAM_TO): + if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD and self.STATUS[_slot]['RX_STREAM_ID'] != _stream_id: + logger.info('(%s) Call not routed to destination %s, matching call already active on target: HBSystem: %s, TS: %s, DEST: %s', self._system, int_id(_dst_id), _target, _slot, int_id(_target_status[_slot]['RX_TGID'])) + continue + if (_dst_id == _target_status[_slot]['TX_TGID']) and (_rf_src != _target_status[_slot]['TX_RFS']) and ((pkt_time - _target_status[_slot]['TX_TIME']) < STREAM_TO): + if _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD and self.STATUS[_slot]['RX_STREAM_ID'] != _stream_id: + logger.info('(%s) Call not routed for subscriber %s, call route in progress on target: HBSystem: %s, TS: %s, DEST: %s, SUB: %s', self._system, int_id(_rf_src), _target, _slot, int_id(_target_status[_slot]['TX_TGID']), int_id(_target_status[_slot]['TX_RFS'])) + continue + + # Record target information if this is a new call stream? + if (_stream_id != self.STATUS[_slot]['RX_STREAM_ID']): + # Record the DST TGID and Stream ID + _target_status[_slot]['TX_START'] = pkt_time + _target_status[_slot]['TX_TGID'] = _dst_id + _target_status[_slot]['TX_STREAM_ID'] = _stream_id + _target_status[_slot]['TX_RFS'] = _rf_src + _target_status[_slot]['TX_PEER'] = _peer_id + + logger.info('(%s) Unit call bridged to HBP System: %s TS: %s, UNIT: %s', self._system, _target, _slot, int_id(_dst_id)) + if CONFIG['REPORTS']['REPORT']: + systems[_target]._report.send_bridgeEvent('UNIT VOICE,START,TX,{},{},{},{},{},{}'.format(_target, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id)).encode(encoding='utf-8', errors='ignore')) + + # Set other values for the contention handler to test next time there is a frame to forward + _target_status[_slot]['TX_TIME'] = pkt_time + _target_status[_slot]['TX_TYPE'] = _dtype_vseq + + #send the call: + systems[_target].send_system(_data) + + + # Final actions - Is this a voice terminator? + if (_frame_type == HBPF_DATA_SYNC) and (_dtype_vseq == HBPF_SLT_VTERM) and (self.STATUS[_slot]['RX_TYPE'] != HBPF_SLT_VTERM): + self._targets = [] + call_duration = pkt_time - self.STATUS[_slot]['RX_START'] + logger.info('(%s) *UNIT CALL END* STREAM ID: %s SUB: %s (%s) PEER: %s (%s) UNIT %s (%s), TS %s, Duration: %.2f', \ + self._system, int_id(_stream_id), get_alias(_rf_src, subscriber_ids), int_id(_rf_src), get_alias(_peer_id, peer_ids), int_id(_peer_id), get_alias(_dst_id, talkgroup_ids), int_id(_dst_id), _slot, call_duration) + if CONFIG['REPORTS']['REPORT']: + self._report.send_bridgeEvent('UNIT VOICE,END,RX,{},{},{},{},{},{},{:.2f}'.format(self._system, int_id(_stream_id), int_id(_peer_id), int_id(_rf_src), _slot, int_id(_dst_id), call_duration).encode(encoding='utf-8', errors='ignore')) + + # Mark status variables for use later + self.STATUS[_slot]['RX_PEER'] = _peer_id + self.STATUS[_slot]['RX_SEQ'] = _seq + self.STATUS[_slot]['RX_RFS'] = _rf_src + self.STATUS[_slot]['RX_TYPE'] = _dtype_vseq + self.STATUS[_slot]['RX_TGID'] = _dst_id + self.STATUS[_slot]['RX_TIME'] = pkt_time + self.STATUS[_slot]['RX_STREAM_ID'] = _stream_id + + + def dmrd_received(self, _peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data): + if _call_type == 'group': + self.group_received(_peer_id, _rf_src, _dst_id, _seq, _slot, _frame_type, _dtype_vseq, _stream_id, _data) + elif _call_type == 'unit': + if self._system not in UNIT: + logger.error('(%s) *UNIT CALL NOT FORWARDED* UNIT calling is disabled for this system (INGRESS)', self._system) + else: + self.unit_received(_peer_id, _rf_src, _dst_id, _seq, _slot, _frame_type, _dtype_vseq, _stream_id, _data) + elif _call_type == 'vscsbk': + logger.debug('CSBK recieved, but HBlink does not process them currently') + else: + logger.error('Unknown call type recieved -- not processed') # # Socket-based reporting section @@ -754,7 +1106,7 @@ if __name__ == '__main__': if cli_args.LOG_LEVEL: CONFIG['LOGGER']['LOG_LEVEL'] = cli_args.LOG_LEVEL logger = log.config_logging(CONFIG['LOGGER']) - logger.info('\n\nCopyright (c) 2013, 2014, 2015, 2016, 2018, 2019\n\tThe Regents of the K0USY Group. All rights reserved.\n') + logger.info('\n\nCopyright (c) 2013, 2014, 2015, 2016, 2018, 2019, 2020\n\tThe Regents of the K0USY Group. All rights reserved.\n') logger.debug('(GLOBAL) Logging system started, anything from here on gets logged') # Set up the signal handler @@ -782,6 +1134,9 @@ if __name__ == '__main__': # Build the routing rules file BRIDGES = make_bridges(rules_module.BRIDGES) + + # Get rule parameter for private calls + UNIT = rules_module.UNIT # INITIALIZE THE REPORTING LOOP if CONFIG['REPORTS']['REPORT']: diff --git a/config.py b/config.py index 3913a33..e76fbf4 100755 --- a/config.py +++ b/config.py @@ -199,52 +199,6 @@ def build_config(_config_file): 'LAST_PING_ACK_TIME': 0, }}) - if config.get(section, 'MODE') == 'XLXPEER': - CONFIG['SYSTEMS'].update({section: { - 'MODE': config.get(section, 'MODE'), - 'ENABLED': config.getboolean(section, 'ENABLED'), - 'LOOSE': config.getboolean(section, 'LOOSE'), - 'SOCK_ADDR': (gethostbyname(config.get(section, 'IP')), config.getint(section, 'PORT')), - 'IP': gethostbyname(config.get(section, 'IP')), - 'PORT': config.getint(section, 'PORT'), - 'MASTER_SOCKADDR': (gethostbyname(config.get(section, 'MASTER_IP')), config.getint(section, 'MASTER_PORT')), - 'MASTER_IP': gethostbyname(config.get(section, 'MASTER_IP')), - 'MASTER_PORT': config.getint(section, 'MASTER_PORT'), - 'PASSPHRASE': bytes(config.get(section, 'PASSPHRASE'), 'utf-8'), - 'CALLSIGN': bytes(config.get(section, 'CALLSIGN').ljust(8)[:8], 'utf-8'), - 'RADIO_ID': config.getint(section, 'RADIO_ID').to_bytes(4, 'big'), - 'RX_FREQ': bytes(config.get(section, 'RX_FREQ').ljust(9)[:9], 'utf-8'), - 'TX_FREQ': bytes(config.get(section, 'TX_FREQ').ljust(9)[:9], 'utf-8'), - 'TX_POWER': bytes(config.get(section, 'TX_POWER').rjust(2,'0'), 'utf-8'), - 'COLORCODE': bytes(config.get(section, 'COLORCODE').rjust(2,'0'), 'utf-8'), - 'LATITUDE': bytes(config.get(section, 'LATITUDE').ljust(8)[:8], 'utf-8'), - 'LONGITUDE': bytes(config.get(section, 'LONGITUDE').ljust(9)[:9], 'utf-8'), - 'HEIGHT': bytes(config.get(section, 'HEIGHT').rjust(3,'0'), 'utf-8'), - 'LOCATION': bytes(config.get(section, 'LOCATION').ljust(20)[:20], 'utf-8'), - 'DESCRIPTION': bytes(config.get(section, 'DESCRIPTION').ljust(19)[:19], 'utf-8'), - 'SLOTS': bytes(config.get(section, 'SLOTS'), 'utf-8'), - 'URL': bytes(config.get(section, 'URL').ljust(124)[:124], 'utf-8'), - 'SOFTWARE_ID': bytes(config.get(section, 'SOFTWARE_ID').ljust(40)[:40], 'utf-8'), - 'PACKAGE_ID': bytes(config.get(section, 'PACKAGE_ID').ljust(40)[:40], 'utf-8'), - 'GROUP_HANGTIME': config.getint(section, 'GROUP_HANGTIME'), - 'XLXMODULE': config.getint(section, 'XLXMODULE'), - 'OPTIONS': '', - 'USE_ACL': config.getboolean(section, 'USE_ACL'), - 'SUB_ACL': config.get(section, 'SUB_ACL'), - 'TG1_ACL': config.get(section, 'TGID_TS1_ACL'), - 'TG2_ACL': config.get(section, 'TGID_TS2_ACL') - }}) - CONFIG['SYSTEMS'][section].update({'XLXSTATS': { - 'CONNECTION': 'NO', # NO, RTPL_SENT, AUTHENTICATED, CONFIG-SENT, YES - 'CONNECTED': None, - 'PINGS_SENT': 0, - 'PINGS_ACKD': 0, - 'NUM_OUTSTANDING': 0, - 'PING_OUTSTANDING': False, - 'LAST_PING_TX_TIME': 0, - 'LAST_PING_ACK_TIME': 0, - }}) - elif config.get(section, 'MODE') == 'MASTER': CONFIG['SYSTEMS'].update({section: { 'MODE': config.get(section, 'MODE'), @@ -274,6 +228,7 @@ def build_config(_config_file): 'TARGET_SOCK': (gethostbyname(config.get(section, 'TARGET_IP')), config.getint(section, 'TARGET_PORT')), 'TARGET_IP': gethostbyname(config.get(section, 'TARGET_IP')), 'TARGET_PORT': config.getint(section, 'TARGET_PORT'), + 'BOTH_SLOTS': config.getboolean(section, 'BOTH_SLOTS'), 'USE_ACL': config.getboolean(section, 'USE_ACL'), 'SUB_ACL': config.get(section, 'SUB_ACL'), 'TG1_ACL': config.get(section, 'TGID_ACL'), diff --git a/entrypoint b/entrypoint deleted file mode 100755 index 422f0f3..0000000 --- a/entrypoint +++ /dev/null @@ -1,19 +0,0 @@ -#!/bin/sh - -if [ -z "$GIT_REPO" ]; then - mkdir -p /var/tmp/config - cd /var/tmp/config - git clone https://${GIT_USER}:${GIT_PASSWORD}@${GIT_REPO} - - DIR=$(echo ${GIT_REPO}| sed s/.git$//g | sed s#^.*/##g) - - cp -a /var/tmp/config/${DIR}/*cfg /opt/hblink3/ - cp -a /var/tmp/config/${DIR}/*csv /opt/hblink3/ - cp -a /var/tmp/config/${DIR}/*json /opt/hblink3/ -else - cp -a /opt/config/*cfg /opt/hblink3/ - cp -a /opt/config/*csv /opt/hblink3/ - cp -a /opt/config/*json /opt/hblink3/ -fi -cd /opt/hblink3 -python /opt/hblink3/hblink.py -c hblink.cfg diff --git a/hblink-750.cfg b/hblink-750.cfg new file mode 100755 index 0000000..aecfff4 --- /dev/null +++ b/hblink-750.cfg @@ -0,0 +1,127 @@ +[GLOBAL] +PATH: ./ +PING_TIME: 5 +MAX_MISSED: 3 +USE_ACL: False +REG_ACL: DENY:1 +SUB_ACL: DENY:1 +TGID_TS1_ACL: PERMIT:ALL +TGID_TS2_ACL: PERMIT:ALL + +[REPORTS] +REPORT: False +REPORT_INTERVAL: 60 +REPORT_PORT: 4321 +REPORT_CLIENTS: 127.0.0.1 + +[LOGGER] +LOG_FILE: /tmp/hblink.log +LOG_HANDLERS: console-timed +LOG_LEVEL: INFO +LOG_NAME: 444.750 + +[ALIASES] +TRY_DOWNLOAD: False +PATH: ./ +PEER_FILE: peer_ids.json +SUBSCRIBER_FILE: subscriber_ids.json +TGID_FILE: talkgroup_ids.json +PEER_URL: https://www.radioid.net/static/rptrs.json +SUBSCRIBER_URL: https://www.radioid.net/static/users.json +STALE_DAYS: 7 + +[OBP] +MODE: OPENBRIDGE +ENABLED: True +IP: +PORT: 50100 +NETWORK_ID: 1 +PASSPHRASE: deadbeef +#TARGET_IP: olympic.k0usy.org +TARGET_IP: 127.0.0.1 +#TARGET_PORT: 50666 +TARGET_PORT: 50101 +BOTH_SLOTS: True +USE_ACL: False +SUB_ACL: DENY:1 +TGID_ACL: PERMIT:ALL + +[444.750] +MODE: MASTER +ENABLED: True +REPEAT: True +MAX_PEERS: 5 +EXPORT_AMBE: False +IP: +PORT: 50001 +PASSPHRASE: jimmy +GROUP_HANGTIME: 10 +USE_ACL: False +REG_ACL: DENY:1 +SUB_ACL: DENY:1 +TGID_TS1_ACL: DENY:8 +TGID_TS2_ACL: PERMIT:3120 + +[TWO] +MODE: MASTER +ENABLED: False +REPEAT: True +MAX_PEERS: 5 +EXPORT_AMBE: False +IP: +PORT:50002 +PASSPHRASE: jimmy +GROUP_HANGTIME: 10 +USE_ACL: False +REG_ACL: DENY:1 +SUB_ACL: DENY:1 +TGID_TS1_ACL: DENY:8 +TGID_TS2_ACL: PERMIT:3120 + +[THREE] +MODE: MASTER +ENABLED: False +REPEAT: True +MAX_PEERS: 5 +EXPORT_AMBE: False +IP: +PORT:50003 +PASSPHRASE: jimmy +GROUP_HANGTIME: 10 +USE_ACL: False +REG_ACL: DENY:1 +SUB_ACL: DENY:1 +TGID_TS1_ACL: DENY:8 +TGID_TS2_ACL: PERMIT:3120 + +[KS-DMR] +MODE: PEER +ENABLED: False +LOOSE: False +EXPORT_AMBE: False +IP: +PORT: 54001 +MASTER_IP: olympic.k0usy.org +MASTER_PORT: 62071 +PASSPHRASE: c0ffee +CALLSIGN: W1ABC +RADIO_ID: 312312 +RX_FREQ: 449000000 +TX_FREQ: 444000000 +TX_POWER: 25 +COLORCODE: 1 +SLOTS: 1 +LATITUDE: 38.0000 +LONGITUDE: -095.0000 +HEIGHT: 75 +LOCATION: Anywhere, USA +DESCRIPTION: This is a cool repeater +URL: www.w1abc.org +SOFTWARE_ID: 20170620 +PACKAGE_ID: MMDVM_HBlink +GROUP_HANGTIME: 5 +OPTIONS: +USE_ACL: True +SUB_ACL: DENY:1 +TGID_TS1_ACL: PERMIT:ALL +TGID_TS2_ACL: PERMIT:ALL diff --git a/hblink-800.cfg b/hblink-800.cfg new file mode 100755 index 0000000..0fe46db --- /dev/null +++ b/hblink-800.cfg @@ -0,0 +1,127 @@ +[GLOBAL] +PATH: ./ +PING_TIME: 5 +MAX_MISSED: 3 +USE_ACL: False +REG_ACL: DENY:1 +SUB_ACL: DENY:1 +TGID_TS1_ACL: PERMIT:ALL +TGID_TS2_ACL: PERMIT:ALL + +[REPORTS] +REPORT: False +REPORT_INTERVAL: 60 +REPORT_PORT: 4321 +REPORT_CLIENTS: 127.0.0.1 + +[LOGGER] +LOG_FILE: /tmp/hblink.log +LOG_HANDLERS: console-timed +LOG_LEVEL: INFO +LOG_NAME: 444.800 + +[ALIASES] +TRY_DOWNLOAD: False +PATH: ./ +PEER_FILE: peer_ids.json +SUBSCRIBER_FILE: subscriber_ids.json +TGID_FILE: talkgroup_ids.json +PEER_URL: https://www.radioid.net/static/rptrs.json +SUBSCRIBER_URL: https://www.radioid.net/static/users.json +STALE_DAYS: 7 + +[OBP] +MODE: OPENBRIDGE +ENABLED: True +IP: +PORT: 50101 +NETWORK_ID: 2 +PASSPHRASE: deadbeef +#TARGET_IP: olympic.k0usy.org +TARGET_IP: 127.0.0.1 +#TARGET_PORT: 50666 +TARGET_PORT: 50100 +BOTH_SLOTS: True +USE_ACL: False +SUB_ACL: DENY:1 +TGID_ACL: PERMIT:ALL + +[444.800] +MODE: MASTER +ENABLED: True +REPEAT: True +MAX_PEERS: 5 +EXPORT_AMBE: False +IP: +PORT: 50011 +PASSPHRASE: jimmy +GROUP_HANGTIME: 10 +USE_ACL: False +REG_ACL: DENY:1 +SUB_ACL: DENY:1 +TGID_TS1_ACL: DENY:8 +TGID_TS2_ACL: PERMIT:3120 + +[TWO] +MODE: MASTER +ENABLED: False +REPEAT: True +MAX_PEERS: 5 +EXPORT_AMBE: False +IP: +PORT:50012 +PASSPHRASE: jimmy +GROUP_HANGTIME: 10 +USE_ACL: False +REG_ACL: DENY:1 +SUB_ACL: DENY:1 +TGID_TS1_ACL: DENY:8 +TGID_TS2_ACL: PERMIT:3120 + +[THREE] +MODE: MASTER +ENABLED: False +REPEAT: True +MAX_PEERS: 5 +EXPORT_AMBE: False +IP: +PORT:50013 +PASSPHRASE: jimmy +GROUP_HANGTIME: 10 +USE_ACL: False +REG_ACL: DENY:1 +SUB_ACL: DENY:1 +TGID_TS1_ACL: DENY:8 +TGID_TS2_ACL: PERMIT:3120 + +[KS-DMR] +MODE: PEER +ENABLED: False +LOOSE: False +EXPORT_AMBE: False +IP: +PORT: 54011 +MASTER_IP: olympic.k0usy.org +MASTER_PORT: 62071 +PASSPHRASE: c0ffee +CALLSIGN: W1ABC +RADIO_ID: 312312 +RX_FREQ: 449000000 +TX_FREQ: 444000000 +TX_POWER: 25 +COLORCODE: 1 +SLOTS: 1 +LATITUDE: 38.0000 +LONGITUDE: -095.0000 +HEIGHT: 75 +LOCATION: Anywhere, USA +DESCRIPTION: This is a cool repeater +URL: www.w1abc.org +SOFTWARE_ID: 20170620 +PACKAGE_ID: MMDVM_HBlink +GROUP_HANGTIME: 5 +OPTIONS: +USE_ACL: True +SUB_ACL: DENY:1 +TGID_TS1_ACL: PERMIT:ALL +TGID_TS2_ACL: PERMIT:ALL diff --git a/hblink-SAMPLE.cfg b/hblink-SAMPLE.cfg index a3b616c..2a50767 100755 --- a/hblink-SAMPLE.cfg +++ b/hblink-SAMPLE.cfg @@ -120,7 +120,9 @@ STALE_DAYS: 7 # # ACLs: # OpenBridge does not 'register', so registration ACL is meaningless. -# OpenBridge passes all traffic on TS1, so there is only 1 TGID ACL. +# Proper OpenBridge passes all traffic on TS1. +# HBlink can extend OPB to use both slots for unit calls only. +# Setting "BOTH_SLOTS" True ONLY affects unit traffic! # Otherwise ACLs work as described in the global stanza [OBP-1] MODE: OPENBRIDGE @@ -131,6 +133,7 @@ NETWORK_ID: 3129100 PASSPHRASE: password TARGET_IP: 1.2.3.4 TARGET_PORT: 62035 +BOTH_SLOTS: True USE_ACL: True SUB_ACL: DENY:1 TGID_ACL: PERMIT:ALL @@ -206,36 +209,4 @@ OPTIONS: USE_ACL: True SUB_ACL: DENY:1 TGID_TS1_ACL: PERMIT:ALL -TGID_TS2_ACL: PERMIT:ALL - -[XLX-1] -MODE: XLXPEER -ENABLED: True -LOOSE: True -EXPORT_AMBE: False -IP: -PORT: 54002 -MASTER_IP: 172.16.1.1 -MASTER_PORT: 62030 -PASSPHRASE: passw0rd -CALLSIGN: W1ABC -RADIO_ID: 312000 -RX_FREQ: 449000000 -TX_FREQ: 444000000 -TX_POWER: 25 -COLORCODE: 1 -SLOTS: 1 -LATITUDE: 38.0000 -LONGITUDE: -095.0000 -HEIGHT: 75 -LOCATION: Anywhere, USA -DESCRIPTION: This is a cool repeater -URL: www.w1abc.org -SOFTWARE_ID: 20170620 -PACKAGE_ID: MMDVM_HBlink -GROUP_HANGTIME: 5 -XLXMODULE: 4004 -USE_ACL: True -SUB_ACL: DENY:1 -TGID_TS1_ACL: PERMIT:ALL -TGID_TS2_ACL: PERMIT:ALL +TGID_TS2_ACL: PERMIT:ALL \ No newline at end of file diff --git a/hblink.py b/hblink.py index 273fe3e..c7566ee 100755 --- a/hblink.py +++ b/hblink.py @@ -160,8 +160,8 @@ class OPENBRIDGE(DatagramProtocol): _stream_id = _data[16:20] #logger.debug('(%s) DMRD - Seqence: %s, RF Source: %s, Destination ID: %s', self._system, int_id(_seq), int_id(_rf_src), int_id(_dst_id)) - # Sanity check for OpenBridge -- all calls must be on Slot 1 - if _slot != 1: + # Sanity check for OpenBridge -- all calls must be on Slot 1 for Brandmeister or DMR+. Other HBlinks can process timeslot on OPB if the flag is set + if _slot != 1 and not self._config['BOTH_SLOTS'] and not _call_type == 'unit': logger.error('(%s) OpenBridge packet discarded because it was not received on slot 1. SID: %s, TGID %s', self._system, int_id(_rf_src), int_id(_dst_id)) return @@ -223,13 +223,6 @@ class HBSYSTEM(DatagramProtocol): self.datagramReceived = self.peer_datagramReceived self.dereg = self.peer_dereg - elif self._config['MODE'] == 'XLXPEER': - self._stats = self._config['XLXSTATS'] - self.send_system = self.send_master - self.maintenance_loop = self.peer_maintenance_loop - self.datagramReceived = self.peer_datagramReceived - self.dereg = self.peer_dereg - def startProtocol(self): # Set up periodic loop for tracking pings from peers. Run every 'PING_TIME' seconds self._system_maintenance = task.LoopingCall(self.maintenance_loop) @@ -289,29 +282,6 @@ class HBSYSTEM(DatagramProtocol): # KEEP THE FOLLOWING COMMENTED OUT UNLESS YOU'RE DEBUGGING DEEPLY!!!! # logger.debug('(%s) TX Packet to %s:%s -- %s', self._system, self._config['MASTER_IP'], self._config['MASTER_PORT'], ahex(_packet)) - def send_xlxmaster(self, radio, xlx, mastersock): - radio3 = int.from_bytes(radio, 'big').to_bytes(3, 'big') - radio4 = int.from_bytes(radio, 'big').to_bytes(4, 'big') - xlx3 = xlx.to_bytes(3, 'big') - streamid = randint(0,255).to_bytes(1, 'big')+randint(0,255).to_bytes(1, 'big')+randint(0,255).to_bytes(1, 'big')+randint(0,255).to_bytes(1, 'big') - # Wait for .5 secs for the XLX to log us in - for packetnr in range(5): - if packetnr < 3: - # First 3 packets, voice start, stream type e1 - strmtype = 225 - payload = bytearray.fromhex('4f2e00b501ae3a001c40a0c1cc7dff57d75df5d5065026f82880bd616f13f185890000') - else: - # Last 2 packets, voice end, stream type e2 - strmtype = 226 - payload = bytearray.fromhex('4f410061011e3a781c30a061ccbdff57d75df5d2534425c02fe0b1216713e885ba0000') - packetnr1 = packetnr.to_bytes(1, 'big') - strmtype1 = strmtype.to_bytes(1, 'big') - _packet = b''.join([DMRD, packetnr1, radio3, xlx3, radio4, strmtype1, streamid, payload]) - self.transport.write(_packet, mastersock) - # KEEP THE FOLLOWING COMMENTED OUT UNLESS YOU'RE DEBUGGING DEEPLY!!!! - #logger.debug('(%s) XLX Module Change Packet: %s', self._system, ahex(_packet)) - return - def dmrd_received(self, _peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data): pass @@ -656,11 +626,7 @@ class HBSYSTEM(DatagramProtocol): self._stats['CONNECTION'] = 'YES' self._stats['CONNECTED'] = time() logger.info('(%s) Connection to Master Completed', self._system) - # If we are an XLX, send the XLX module request here. - if self._config['MODE'] == 'XLXPEER': - self.send_xlxmaster(self._config['RADIO_ID'], int(4000), self._config['MASTER_SOCKADDR']) - self.send_xlxmaster(self._config['RADIO_ID'], self._config['XLXMODULE'], self._config['MASTER_SOCKADDR']) - logger.info('(%s) Sending XLX Module request', self._system) + else: self._stats['CONNECTION'] = 'NO' logger.error('(%s) Master ACK Contained wrong ID - Connection Reset', self._system) @@ -797,7 +763,7 @@ if __name__ == '__main__': if cli_args.LOG_LEVEL: CONFIG['LOGGER']['LOG_LEVEL'] = cli_args.LOG_LEVEL logger = log.config_logging(CONFIG['LOGGER']) - logger.info('\n\nCopyright (c) 2013, 2014, 2015, 2016, 2018, 2019\n\tThe Regents of the K0USY Group. All rights reserved.\n') + logger.info('\n\nCopyright (c) 2013, 2014, 2015, 2016, 2018, 2019, 2020\n\tThe Regents of the K0USY Group. All rights reserved.\n') logger.debug('(GLOBAL) Logging system started, anything from here on gets logged') # Set up the signal handler diff --git a/rules-750.py b/rules-750.py new file mode 100755 index 0000000..b0c1b5f --- /dev/null +++ b/rules-750.py @@ -0,0 +1,16 @@ +BRIDGES = { + '1/2': [ + {'SYSTEM': '444.750', 'TS': 1, 'TGID': 2, 'ACTIVE': True, 'TIMEOUT': 5,'TO_TYPE': 'NONE', 'ON': [], 'OFF': [], 'RESET': []}, + {'SYSTEM': 'OBP', 'TS': 1, 'TGID': 2, 'ACTIVE': True, 'TIMEOUT': 5,'TO_TYPE': 'NONE', 'ON': [], 'OFF': [], 'RESET': []} + ], + 'KANSAS': [ + {'SYSTEM': '444.750', 'TS': 2, 'TGID': 3120, 'ACTIVE': True, 'TIMEOUT': 5,'TO_TYPE': 'NONE', 'ON': [], 'OFF': [], 'RESET': []}, + {'SYSTEM': 'OBP', 'TS': 1, 'TGID': 3120, 'ACTIVE': True, 'TIMEOUT': 5,'TO_TYPE': 'NONE', 'ON': [], 'OFF': [], 'RESET': []} + ] +} + +UNIT = ['444.750', 'OBP'] + +if __name__ == '__main__': + from pprint import pprint + pprint(BRIDGES) diff --git a/rules-800.py b/rules-800.py new file mode 100755 index 0000000..0d10f53 --- /dev/null +++ b/rules-800.py @@ -0,0 +1,16 @@ +BRIDGES = { + '1/2': [ + {'SYSTEM': '444.800', 'TS': 1, 'TGID': 2, 'ACTIVE': True, 'TIMEOUT': 5,'TO_TYPE': 'NONE', 'ON': [], 'OFF': [], 'RESET': []}, + {'SYSTEM': 'OBP', 'TS': 1, 'TGID': 2, 'ACTIVE': True, 'TIMEOUT': 5,'TO_TYPE': 'NONE', 'ON': [], 'OFF': [], 'RESET': []} + ], + 'KANSAS': [ + {'SYSTEM': '444.800', 'TS': 2, 'TGID': 3120, 'ACTIVE': True, 'TIMEOUT': 5,'TO_TYPE': 'NONE', 'ON': [], 'OFF': [], 'RESET': []}, + {'SYSTEM': 'OBP', 'TS': 1, 'TGID': 3120, 'ACTIVE': True, 'TIMEOUT': 5,'TO_TYPE': 'NONE', 'ON': [], 'OFF': [], 'RESET': []} + ] +} + +UNIT = ["444.800", "OBP"] + +if __name__ == '__main__': + from pprint import pprint + pprint(BRIDGES) diff --git a/rules_SAMPLE.py b/rules_SAMPLE.py index a3c3701..b46a46a 100755 --- a/rules_SAMPLE.py +++ b/rules_SAMPLE.py @@ -47,6 +47,18 @@ BRIDGES = { ] } +''' +list the names of each system that should bridge unit to unit (individual) calls. +''' + +UNIT = ['ONE', 'TWO'] + +''' +This is for testing the syntax of the file. It won't eliminate all errors, but running this file +like it were a Python program itself will tell you if the syntax is correct! +''' + if __name__ == '__main__': from pprint import pprint pprint(BRIDGES) + print(UNIT) diff --git a/voice_lib.py b/voice_lib.py old mode 100644 new mode 100755