2383 lines
143 KiB
Python
Executable File
2383 lines
143 KiB
Python
Executable File
#!/usr/bin/env python
|
|
#
|
|
###############################################################################
|
|
# Copyright (C) 2020 Simon Adlem, G7RZU <g7rzu@gb7fr.org.uk>
|
|
# Copyright (C) 2016-2019 Cortney T. Buffington, N0MJS <n0mjs@me.com>
|
|
#
|
|
# This program is free software; you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation; either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program; if not, write to the Free Software Foundation,
|
|
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
|
|
###############################################################################
|
|
|
|
'''
|
|
This application, in conjuction with it's rule file (rules.py) will
|
|
work like a "conference bridge". This is similar to what most hams think of as a
|
|
reflector. You define conference bridges and any system joined to that conference
|
|
bridge will both receive traffic from, and send traffic to any other system
|
|
joined to the same conference bridge. It does not provide end-to-end connectivity
|
|
as each end system must individually be joined to a conference bridge (a name
|
|
you create in the configuraiton file) to pass traffic.
|
|
|
|
This program currently only works with group voice calls.
|
|
'''
|
|
|
|
# Python modules we need
|
|
import sys
|
|
from bitarray import bitarray
|
|
from time import time,sleep,perf_counter
|
|
import importlib.util
|
|
import re
|
|
import copy
|
|
from setproctitle import setproctitle
|
|
|
|
# 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, acl_check
|
|
from dmr_utils3.utils import bytes_3, int_id, get_alias, bytes_4
|
|
from dmr_utils3 import decode, bptc, const
|
|
import config
|
|
from config import acl_build
|
|
import log
|
|
from const import *
|
|
from mk_voice import pkt_gen
|
|
#from voice_lib import words
|
|
|
|
#Read voices
|
|
from read_ambe import readAMBE
|
|
#Remap some words for certain languages
|
|
from i8n_voice_map import voiceMap
|
|
|
|
|
|
#MySQL
|
|
from mysql_config import useMYSQL
|
|
|
|
# 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__)
|
|
|
|
#REGEX
|
|
import re
|
|
|
|
|
|
# Does anybody read this stuff? There's a PEP somewhere that says I should do this.
|
|
__author__ = 'Cortney T. Buffington, N0MJS, Forked by Simon Adlem - G7RZU'
|
|
__copyright__ = 'Copyright (c) 2016-2019 Cortney T. Buffington, N0MJS and the K0USY Group, Simon Adlem, G7RZU 2020,2021'
|
|
__credits__ = 'Colin Durbridge, G4EML, Steve Zingman, N4IRS; Mike Zingman, N4IRR; Jonathan Naylor, G4KLX; Hans Barthen, DL5DI; Torsten Shultze, DG1HT; Jon Lee, G4TSN; Norman Williams, M6NBP'
|
|
__license__ = 'GNU GPLv3'
|
|
__maintainer__ = 'Simon Adlem G7RZU'
|
|
__email__ = 'simon@gb7fr.org.uk'
|
|
|
|
# Module gobal varaibles
|
|
|
|
# Timed loop used for reporting HBP status
|
|
#
|
|
# REPORT BASED ON THE TYPE SELECTED IN THE MAIN CONFIG FILE
|
|
def config_reports(_config, _factory):
|
|
if True: #_config['REPORTS']['REPORT']:
|
|
def reporting_loop(logger, _server):
|
|
logger.debug('(REPORT) Periodic reporting loop started')
|
|
_server.send_config()
|
|
_server.send_bridge()
|
|
i = 0
|
|
for system in CONFIG['SYSTEMS']:
|
|
if 'PEERS' in CONFIG['SYSTEMS'][system] and CONFIG['SYSTEMS'][system]['PEERS']:
|
|
i = i +1
|
|
logger.info('(REPORT) %s systems have at least one peer',i)
|
|
|
|
logger.info('(REPORT) HBlink TCP reporting server configured')
|
|
|
|
report_server = _factory(_config)
|
|
report_server.clients = []
|
|
reactor.listenTCP(_config['REPORTS']['REPORT_PORT'], report_server)
|
|
|
|
reporting = task.LoopingCall(reporting_loop, logger, report_server)
|
|
reporting.start(_config['REPORTS']['REPORT_INTERVAL'])
|
|
|
|
return report_server
|
|
|
|
|
|
# Import Bridging rules
|
|
# Note: A stanza *must* exist for any MASTER or CLIENT configured in the main
|
|
# configuration file and listed as "active". It can be empty,
|
|
# but it has to exist.
|
|
def make_bridges(_rules):
|
|
# Convert integer GROUP ID numbers from the config into hex strings
|
|
# we need to send in the actual data packets.
|
|
for _bridge in _rules:
|
|
for _system in _rules[_bridge]:
|
|
if _system['SYSTEM'] not in CONFIG['SYSTEMS']:
|
|
sys.exit('ERROR: Conference bridge "{}" references a system named "{}" that is not enabled in the main configuration'.format(_bridge, _system['SYSTEM']))
|
|
|
|
_system['TGID'] = bytes_3(_system['TGID'])
|
|
for i, e in enumerate(_system['ON']):
|
|
_system['ON'][i] = bytes_3(_system['ON'][i])
|
|
for i, e in enumerate(_system['OFF']):
|
|
_system['OFF'][i] = bytes_3(_system['OFF'][i])
|
|
_system['TIMEOUT'] = _system['TIMEOUT']*60
|
|
if _system['ACTIVE'] == True:
|
|
_system['TIMER'] = time() + _system['TIMEOUT']
|
|
else:
|
|
_system['TIMER'] = time()
|
|
|
|
# if _bridge[0:1] == '#':
|
|
# continue
|
|
|
|
for _confsystem in CONFIG['SYSTEMS']:
|
|
#if _confsystem[0:3] == 'OBP':
|
|
if CONFIG['SYSTEMS'][_confsystem]['MODE'] != 'MASTER':
|
|
continue
|
|
ts1 = False
|
|
ts2 = False
|
|
for i,e in enumerate(_rules[_bridge]):
|
|
if e['SYSTEM'] == _confsystem and e['TS'] == 1:
|
|
ts1 = True
|
|
if e['SYSTEM'] == _confsystem and e['TS'] == 2:
|
|
ts2 = True
|
|
if _bridge[0:1] != '#':
|
|
_tmout = CONFIG['SYSTEMS'][_confsystem]['DEFAULT_UA_TIMER']
|
|
if ts1 == False:
|
|
_rules[_bridge].append({'SYSTEM': _confsystem, 'TS': 1, 'TGID': bytes_3(int(_bridge)),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(int(_bridge)),],'RESET': [], 'TIMER': time()})
|
|
if ts2 == False:
|
|
_rules[_bridge].append({'SYSTEM': _confsystem, 'TS': 2, 'TGID': bytes_3(int(_bridge)),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(int(_bridge)),],'RESET': [], 'TIMER': time()})
|
|
else:
|
|
_tmout = CONFIG['SYSTEMS'][_confsystem]['DEFAULT_UA_TIMER']
|
|
if ts2 == False:
|
|
_rules[_bridge].append({'SYSTEM': _confsystem, 'TS': 2, 'TGID': bytes_3(9),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [bytes_3(4000)],'ON': [],'RESET': [], 'TIMER': time()})
|
|
|
|
return _rules
|
|
|
|
#Make a single bridge - used for on-the-fly UA bridges
|
|
def make_single_bridge(_tgid,_sourcesystem,_slot,_tmout):
|
|
_tgid_s = str(int_id(_tgid))
|
|
#Always a 1 min timeout for Echo
|
|
if _tgid_s == '9990':
|
|
_tmout = 1
|
|
BRIDGES[_tgid_s] = []
|
|
for _system in CONFIG['SYSTEMS']:
|
|
if _system[0:3] != 'OBP':
|
|
#if CONFIG['SYSTEMS'][system]['MODE'] == 'MASTER':
|
|
#_tmout = CONFIG['SYSTEMS'][_system]['DEFAULT_UA_TIMER']
|
|
if _system == _sourcesystem:
|
|
if _slot == 1:
|
|
BRIDGES[_tgid_s].append({'SYSTEM': _system, 'TS': 1, 'TGID': _tgid,'ACTIVE': True,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [_tgid,],'RESET': [], 'TIMER': time() + (_tmout * 60)})
|
|
BRIDGES[_tgid_s].append({'SYSTEM': _system, 'TS': 2, 'TGID': _tgid,'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [_tgid,],'RESET': [], 'TIMER': time()})
|
|
else:
|
|
BRIDGES[_tgid_s].append({'SYSTEM': _system, 'TS': 2, 'TGID': _tgid,'ACTIVE': True,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [_tgid,],'RESET': [], 'TIMER': time() + (_tmout * 60)})
|
|
BRIDGES[_tgid_s].append({'SYSTEM': _system, 'TS': 1, 'TGID': _tgid,'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [_tgid,],'RESET': [], 'TIMER': time()})
|
|
else:
|
|
BRIDGES[_tgid_s].append({'SYSTEM': _system, 'TS': 1, 'TGID': _tgid,'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [_tgid,],'RESET': [], 'TIMER': time()})
|
|
BRIDGES[_tgid_s].append({'SYSTEM': _system, 'TS': 2, 'TGID': _tgid,'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [_tgid,],'RESET': [], 'TIMER': time()})
|
|
|
|
if _system[0:3] == 'OBP' and (int_id(_tgid) >= 79 and (int_id(_tgid) < 9990 or int_id(_tgid) > 9999)):
|
|
BRIDGES[_tgid_s].append({'SYSTEM': _system, 'TS': 1, 'TGID': _tgid,'ACTIVE': True,'TIMEOUT': '','TO_TYPE': 'NONE','OFF': [],'ON': [],'RESET': [], 'TIMER': time()})
|
|
|
|
#Make static bridge - used for on-the-fly relay bridges
|
|
def make_stat_bridge(_tgid):
|
|
_tgid_s = str(int_id(_tgid))
|
|
BRIDGES[_tgid_s] = []
|
|
for _system in CONFIG['SYSTEMS']:
|
|
if _system[0:3] != 'OBP':
|
|
if CONFIG['SYSTEMS'][_system]['MODE'] == 'MASTER':
|
|
_tmout = CONFIG['SYSTEMS'][_system]['DEFAULT_UA_TIMER']
|
|
BRIDGES[_tgid_s].append({'SYSTEM': _system, 'TS': 1, 'TGID': _tgid,'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [_tgid,],'RESET': [], 'TIMER': time()})
|
|
BRIDGES[_tgid_s].append({'SYSTEM': _system, 'TS': 2, 'TGID': _tgid,'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [_tgid,],'RESET': [], 'TIMER': time()})
|
|
|
|
if _system[0:3] == 'OBP':
|
|
BRIDGES[_tgid_s].append({'SYSTEM': _system, 'TS': 1, 'TGID': _tgid,'ACTIVE': True,'TIMEOUT': '','TO_TYPE': 'STAT','OFF': [],'ON': [],'RESET': [], 'TIMER': time()})
|
|
|
|
|
|
def make_default_reflector(reflector,_tmout,system):
|
|
bridge = '#'+str(reflector)
|
|
#_tmout = CONFIG['SYSTEMS'][system]['DEFAULT_UA_TIMER']
|
|
if bridge not in BRIDGES:
|
|
BRIDGES[bridge] = []
|
|
make_single_reflector(bytes_3(reflector),_tmout, system)
|
|
bridgetemp = []
|
|
for bridgesystem in BRIDGES[bridge]:
|
|
if bridgesystem['SYSTEM'] == system and bridgesystem['TS'] == 2:
|
|
bridgetemp.append({'SYSTEM': system, 'TS': 2, 'TGID': bytes_3(9),'ACTIVE': True,'TIMEOUT': _tmout * 60,'TO_TYPE': 'OFF','OFF': [],'ON': [bytes_3(reflector),],'RESET': [], 'TIMER': time() + (_tmout * 60)})
|
|
else:
|
|
bridgetemp.append(bridgesystem)
|
|
|
|
BRIDGES[bridge] = bridgetemp
|
|
|
|
def make_static_tg(tg,ts,_tmout,system):
|
|
#_tmout = CONFIG['SYSTEMS'][system]['DEFAULT_UA_TIMER']
|
|
if str(tg) not in BRIDGES:
|
|
make_single_bridge(bytes_3(tg),system,ts,_tmout)
|
|
bridgetemp = []
|
|
for bridgesystem in BRIDGES[str(tg)]:
|
|
if bridgesystem['SYSTEM'] == system and bridgesystem['TS'] == ts:
|
|
bridgetemp.append({'SYSTEM': system, 'TS': ts, 'TGID': bytes_3(tg),'ACTIVE': True,'TIMEOUT': _tmout * 60,'TO_TYPE': 'OFF','OFF': [],'ON': [bytes_3(tg),],'RESET': [], 'TIMER': time() + (_tmout * 60)})
|
|
else:
|
|
bridgetemp.append(bridgesystem)
|
|
|
|
BRIDGES[str(tg)] = bridgetemp
|
|
|
|
def reset_static_tg(tg,ts,_tmout,system):
|
|
#_tmout = CONFIG['SYSTEMS'][system]['DEFAULT_UA_TIMER']
|
|
bridgetemp = []
|
|
try:
|
|
for bridgesystem in BRIDGES[str(tg)]:
|
|
if bridgesystem['SYSTEM'] == system and bridgesystem['TS'] == ts:
|
|
bridgetemp.append({'SYSTEM': system, 'TS': ts, 'TGID': bytes_3(tg),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(tg),],'RESET': [], 'TIMER': time() + (_tmout * 60)})
|
|
else:
|
|
bridgetemp.append(bridgesystem)
|
|
|
|
BRIDGES[str(tg)] = bridgetemp
|
|
except KeyError:
|
|
logger.exception('(ERROR) KeyError in reset_static_tg() - bridge gone away?')
|
|
return
|
|
|
|
def reset_default_reflector(reflector,_tmout,system):
|
|
bridge = '#'+str(reflector)
|
|
#_tmout = CONFIG['SYSTEMS'][system]['DEFAULT_UA_TIMER']
|
|
if bridge not in BRIDGES:
|
|
BRIDGES[bridge] = []
|
|
make_single_reflector(bytes_3(reflector),_tmout, system)
|
|
bridgetemp = []
|
|
for bridgesystem in BRIDGES[bridge]:
|
|
if bridgesystem['SYSTEM'] == system and bridgesystem['TS'] == 2:
|
|
bridgetemp.append({'SYSTEM': system, 'TS': 2, 'TGID': bytes_3(9),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(reflector),],'RESET': [], 'TIMER': time() + (_tmout * 60)})
|
|
else:
|
|
bridgetemp.append(bridgesystem)
|
|
BRIDGES[bridge] = bridgetemp
|
|
|
|
def make_single_reflector(_tgid,_tmout,_sourcesystem):
|
|
_tgid_s = str(int_id(_tgid))
|
|
_bridge = '#' + _tgid_s
|
|
#1 min timeout for echo
|
|
if _tgid_s == '9990':
|
|
_tmout = 1
|
|
BRIDGES[_bridge] = []
|
|
for _system in CONFIG['SYSTEMS']:
|
|
#if _system[0:3] != 'OBP':
|
|
if CONFIG['SYSTEMS'][_system]['MODE'] == 'MASTER':
|
|
#_tmout = CONFIG['SYSTEMS'][_system]['DEFAULT_UA_TIMER']
|
|
if _system == _sourcesystem:
|
|
BRIDGES[_bridge].append({'SYSTEM': _system, 'TS': 2, 'TGID': bytes_3(9),'ACTIVE': True,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [_tgid,],'RESET': [], 'TIMER': time() + (_tmout * 60)})
|
|
else:
|
|
BRIDGES[_bridge].append({'SYSTEM': _system, 'TS': 2, 'TGID': bytes_3(9),'ACTIVE': False,'TIMEOUT': CONFIG['SYSTEMS'][_system]['DEFAULT_UA_TIMER'] * 60,'TO_TYPE': 'ON','OFF': [],'ON': [_tgid,],'RESET': [], 'TIMER': time()})
|
|
if _system[0:3] == 'OBP' and (int_id(_tgid) >= 79 and (int_id(_tgid) < 9990 or int_id(_tgid) > 9999)):
|
|
BRIDGES[_bridge].append({'SYSTEM': _system, 'TS': 1, 'TGID': _tgid,'ACTIVE': True,'TIMEOUT': '','TO_TYPE': 'NONE','OFF': [],'ON': [],'RESET': [], 'TIMER': time()})
|
|
|
|
def remove_bridge_system(system):
|
|
_bridgestemp = {}
|
|
_bridgetemp = {}
|
|
for _bridge in BRIDGES:
|
|
for _bridgesystem in BRIDGES[_bridge]:
|
|
if _bridgesystem['SYSTEM'] != system:
|
|
if _bridge not in _bridgestemp:
|
|
_bridgestemp[_bridge] = []
|
|
_bridgestemp[_bridge].append(_bridgesystem)
|
|
BRIDGES.update(_bridgestemp)
|
|
|
|
|
|
# Run this every minute for rule timer updates
|
|
def rule_timer_loop():
|
|
logger.debug('(ROUTER) routerHBP Rule timer loop started')
|
|
_now = time()
|
|
_remove_bridges = []
|
|
for _bridge in BRIDGES:
|
|
_bridge_used = False
|
|
for _system in BRIDGES[_bridge]:
|
|
if _system['TO_TYPE'] == 'ON':
|
|
if _system['ACTIVE'] == True:
|
|
_bridge_used = True
|
|
if _system['TIMER'] < _now:
|
|
_system['ACTIVE'] = False
|
|
logger.info('(ROUTER) Conference Bridge TIMEOUT: DEACTIVATE System: %s, Bridge: %s, TS: %s, TGID: %s', _system['SYSTEM'], _bridge, _system['TS'], int_id(_system['TGID']))
|
|
if _bridge[0:1] == '#':
|
|
reactor.callInThread(disconnectedVoice,_system['SYSTEM'])
|
|
else:
|
|
timeout_in = _system['TIMER'] - _now
|
|
_bridge_used = True
|
|
logger.info('(ROUTER) Conference Bridge ACTIVE (ON timer running): System: %s Bridge: %s, TS: %s, TGID: %s, Timeout in: %.2fs,', _system['SYSTEM'], _bridge, _system['TS'], int_id(_system['TGID']), timeout_in)
|
|
elif _system['ACTIVE'] == False:
|
|
logger.debug('(ROUTER) Conference Bridge INACTIVE (no change): System: %s Bridge: %s, TS: %s, TGID: %s', _system['SYSTEM'], _bridge, _system['TS'], int_id(_system['TGID']))
|
|
elif _system['TO_TYPE'] == 'OFF':
|
|
if _system['ACTIVE'] == False:
|
|
if _system['TIMER'] < _now:
|
|
_system['ACTIVE'] = True
|
|
_bridge_used = True
|
|
logger.info('(ROUTER) Conference Bridge TIMEOUT: ACTIVATE System: %s, Bridge: %s, TS: %s, TGID: %s', _system['SYSTEM'], _bridge, _system['TS'], int_id(_system['TGID']))
|
|
else:
|
|
timeout_in = _system['TIMER'] - _now
|
|
_bridge_used = True
|
|
logger.info('(ROUTER) Conference Bridge INACTIVE (OFF timer running): System: %s Bridge: %s, TS: %s, TGID: %s, Timeout in: %.2fs,', _system['SYSTEM'], _bridge, _system['TS'], int_id(_system['TGID']), timeout_in)
|
|
elif _system['ACTIVE'] == True:
|
|
_bridge_used = True
|
|
logger.debug('(ROUTER) Conference Bridge ACTIVE (no change): System: %s Bridge: %s, TS: %s, TGID: %s', _system['SYSTEM'], _bridge, _system['TS'], int_id(_system['TGID']))
|
|
else:
|
|
if _system['SYSTEM'][0:3] != 'OBP':
|
|
_bridge_used = True
|
|
elif _system['SYSTEM'][0:3] == 'OBP' and _system['TO_TYPE'] == 'STAT':
|
|
_bridge_used = True
|
|
logger.debug('(ROUTER) Conference Bridge NO ACTION: System: %s, Bridge: %s, TS: %s, TGID: %s', _system['SYSTEM'], _bridge, _system['TS'], int_id(_system['TGID']))
|
|
|
|
if _bridge_used == False:
|
|
_remove_bridges.append(_bridge)
|
|
|
|
for _bridgerem in _remove_bridges:
|
|
del BRIDGES[_bridgerem]
|
|
logger.debug('(ROUTER) Unused conference bridge %s removed',_bridgerem)
|
|
|
|
if CONFIG['REPORTS']['REPORT']:
|
|
report_server.send_clients(b'bridge updated')
|
|
|
|
def statTrimmer():
|
|
logger.debug('(ROUTER) STAT trimmer loop started')
|
|
_remove_bridges = []
|
|
for _bridge in BRIDGES:
|
|
_bridge_stat = False
|
|
_in_use = False
|
|
for _system in BRIDGES[_bridge]:
|
|
if _system['TO_TYPE'] == 'STAT':
|
|
_bridge_stat = True
|
|
if _system['TO_TYPE'] == 'ON' and _system['ACTIVE']:
|
|
_in_use = True
|
|
elif _system['TO_TYPE'] == 'OFF' and not _system['ACTIVE']:
|
|
_in_use = True
|
|
if _bridge_stat and not _in_use:
|
|
_remove_bridges.append(_bridge)
|
|
for _bridgerem in _remove_bridges:
|
|
del BRIDGES[_bridgerem]
|
|
logger.debug('(ROUTER) STAT bridge %s removed',_bridgerem)
|
|
if CONFIG['REPORTS']['REPORT']:
|
|
report_server.send_clients(b'bridge updated')
|
|
|
|
def kaReporting():
|
|
logger.debug('(ROUTER) KeepAlive reporting loop started')
|
|
for system in systems:
|
|
if CONFIG['SYSTEMS'][system]['MODE'] == 'OPENBRIDGE':
|
|
if CONFIG['SYSTEMS'][system]['ENHANCED_OBP']:
|
|
if '_bcka' not in CONFIG['SYSTEMS'][system]:
|
|
logger.warning('(ROUTER) not sending to system %s as KeepAlive never seen',system)
|
|
elif CONFIG['SYSTEMS'][system]['_bcka'] < time() - 60:
|
|
logger.warning('(ROUTER) not sending to system %s as last KeepAlive was %s seconds ago',system, int(time() - CONFIG['SYSTEMS'][system]['_bcka']))
|
|
|
|
# run this every 10 seconds to trim stream ids
|
|
def stream_trimmer_loop():
|
|
logger.debug('(ROUTER) Trimming inactive stream IDs from system lists')
|
|
_now = time()
|
|
|
|
for system in systems:
|
|
# HBP systems, master and peer
|
|
if CONFIG['SYSTEMS'][system]['MODE'] != 'OPENBRIDGE':
|
|
for slot in range(1,3):
|
|
_slot = systems[system].STATUS[slot]
|
|
|
|
# RX slot check
|
|
if _slot['RX_TYPE'] != HBPF_SLT_VTERM and _slot['RX_TIME'] < _now - 5:
|
|
_slot['RX_TYPE'] = HBPF_SLT_VTERM
|
|
logger.info('(%s) *TIME OUT* RX STREAM ID: %s SUB: %s TGID %s, TS %s, Duration: %.2f', \
|
|
system, int_id(_slot['RX_STREAM_ID']), int_id(_slot['RX_RFS']), int_id(_slot['RX_TGID']), slot, _slot['RX_TIME'] - _slot['RX_START'])
|
|
if CONFIG['REPORTS']['REPORT']:
|
|
systems[system]._report.send_bridgeEvent('GROUP VOICE,END,RX,{},{},{},{},{},{},{:.2f}'.format(system, int_id(_slot['RX_STREAM_ID']), int_id(_slot['RX_PEER']), int_id(_slot['RX_RFS']), slot, int_id(_slot['RX_TGID']), _slot['RX_TIME'] - _slot['RX_START']).encode(encoding='utf-8', errors='ignore'))
|
|
#Null stream_id - for loop control
|
|
if _slot['RX_TIME'] < _now - 60:
|
|
_slot['RX_STREAM_ID'] = b'\x00'
|
|
|
|
# TX slot check
|
|
if _slot['TX_TYPE'] != HBPF_SLT_VTERM and _slot['TX_TIME'] < _now - 5:
|
|
_slot['TX_TYPE'] = HBPF_SLT_VTERM
|
|
logger.info('(%s) *TIME OUT* TX STREAM ID: %s SUB: %s TGID %s, TS %s, Duration: %.2f', \
|
|
system, int_id(_slot['TX_STREAM_ID']), int_id(_slot['TX_RFS']), int_id(_slot['TX_TGID']), slot, _slot['TX_TIME'] - _slot['TX_START'])
|
|
if CONFIG['REPORTS']['REPORT']:
|
|
systems[system]._report.send_bridgeEvent('GROUP VOICE,END,TX,{},{},{},{},{},{},{:.2f}'.format(system, int_id(_slot['TX_STREAM_ID']), int_id(_slot['TX_PEER']), int_id(_slot['TX_RFS']), slot, int_id(_slot['TX_TGID']), _slot['TX_TIME'] - _slot['TX_START']).encode(encoding='utf-8', errors='ignore'))
|
|
|
|
# OBP systems
|
|
# We can't delete items from a dicationry that's being iterated, so we have to make a temporarly list of entrys to remove later
|
|
if CONFIG['SYSTEMS'][system]['MODE'] == 'OPENBRIDGE':
|
|
remove_list = []
|
|
fin_list = []
|
|
for stream_id in systems[system].STATUS:
|
|
|
|
#if stream already marked as finished, just remove it
|
|
if '_fin' in systems[system].STATUS[stream_id] and systems[system].STATUS[stream_id]['LAST'] < _now - 180:
|
|
logger.info('(%s) *FINISHED STREAM* STREAM ID: %s',system, int_id(stream_id))
|
|
fin_list.append(stream_id)
|
|
continue
|
|
|
|
#try:
|
|
if '_to' not in systems[system].STATUS[stream_id] and '_fin' not in systems[system].STATUS[stream_id] and systems[system].STATUS[stream_id]['LAST'] < _now - 5:
|
|
_stream = systems[system].STATUS[stream_id]
|
|
_sysconfig = CONFIG['SYSTEMS'][system]
|
|
#systems[system].STATUS[stream_id]['_fin'] = True
|
|
if '_bcsq' in _sysconfig and _stream['TGID'] in _sysconfig['_bcsq'] and _sysconfig['_bcsq'][_stream['TGID']] == stream_id:
|
|
logger.debug('(%s) *TIME OUT* STREAM ID: %s SUB: %s PEER: %s TGID: %s TS 1 (BCSQ)', \
|
|
system, int_id(stream_id), get_alias(int_id(_stream['RFS']), subscriber_ids), get_alias(int_id(_stream['RX_PEER']), peer_ids), get_alias(int_id(_stream['TGID']), talkgroup_ids))
|
|
elif '_bcsq' in systems[system].STATUS[stream_id] :
|
|
logger.debug('(%s) *TIME OUT* STREAM ID: %s SUB: %s PEER: %s TGID: %s TS 1 (BCSQ)', \
|
|
system, int_id(stream_id), get_alias(int_id(_stream['RFS']), subscriber_ids), get_alias(int_id(_stream['RX_PEER']), peer_ids), get_alias(int_id(_stream['TGID']), talkgroup_ids))
|
|
else:
|
|
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(_stream['RX_PEER']), peer_ids), get_alias(int_id(_stream['TGID']), 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(_stream['RX_PEER']), int_id(_stream['RFS']), 1, int_id(_stream['TGID']), _stream['LAST'] - _stream['START']).encode(encoding='utf-8', errors='ignore'))
|
|
systems[system].STATUS[stream_id]['_to'] = True
|
|
continue
|
|
#except:
|
|
#logger.warning("(%s) Keyerror - stream trimmer Stream ID: %s",system,stream_id)
|
|
#systems[system].STATUS[stream_id]['LAST'] = _now
|
|
#continue
|
|
|
|
|
|
try:
|
|
if systems[system].STATUS[stream_id]['LAST'] < _now - 180:
|
|
remove_list.append(stream_id)
|
|
except Exception as e:
|
|
logger.exception("(%s) Keyerror - stream trimmer Stream ID: %s",system,stream_id, exc_info=e)
|
|
systems[system].STATUS[stream_id]['LAST'] = _now
|
|
continue
|
|
|
|
#remove finished
|
|
for stream_id in fin_list:
|
|
removed = systems[system].STATUS.pop(stream_id)
|
|
|
|
for stream_id in remove_list:
|
|
if stream_id in systems[system].STATUS:
|
|
_stream = systems[system].STATUS[stream_id]
|
|
_sysconfig = CONFIG['SYSTEMS'][system]
|
|
|
|
removed = systems[system].STATUS.pop(stream_id)
|
|
|
|
try:
|
|
_bcsq_remove = []
|
|
for tgid in _sysconfig['_bcsq']:
|
|
if _sysconfig['_bcsq'][tgid] == stream_id:
|
|
_bcsq_remove.append(tgid)
|
|
for bcrm in _bcsq_remove:
|
|
removed = _sysconfig['_bcsq'].pop(bcrm)
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
logger.debug('(%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])
|
|
|
|
def sendVoicePacket(self,pkt,_source_id,_dest_id,_slot):
|
|
_stream_id = pkt[16:20]
|
|
_pkt_time = time()
|
|
if _stream_id not in systems[system].STATUS:
|
|
systems[system].STATUS[_stream_id] = {
|
|
'START': _pkt_time,
|
|
'CONTENTION':False,
|
|
'RFS': _source_id,
|
|
'TGID': _dest_id,
|
|
'LAST': _pkt_time
|
|
}
|
|
_slot['TX_TGID'] = _dest_id
|
|
else:
|
|
systems[system].STATUS[_stream_id]['LAST'] = _pkt_time
|
|
_slot['TX_TIME'] = _pkt_time
|
|
|
|
self.send_system(pkt)
|
|
|
|
def sendSpeech(self,speech):
|
|
logger.debug('(%s) Inside sendspeech thread',self._system)
|
|
sleep(1)
|
|
_nine = bytes_3(9)
|
|
_source_id = bytes_3(5000)
|
|
_slot = systems[system].STATUS[2]
|
|
while True:
|
|
try:
|
|
pkt = next(speech)
|
|
except StopIteration:
|
|
break
|
|
#Packet every 60ms
|
|
sleep(0.058)
|
|
reactor.callFromThread(sendVoicePacket,self,pkt,_source_id,_nine,_slot)
|
|
|
|
logger.debug('(%s) Sendspeech thread ended',self._system)
|
|
|
|
def disconnectedVoice(system):
|
|
_nine = bytes_3(9)
|
|
_source_id = bytes_3(5000)
|
|
_lang = CONFIG['SYSTEMS'][system]['ANNOUNCEMENT_LANGUAGE']
|
|
logger.debug('(%s) Sending disconnected voice',system)
|
|
_say = [words[_lang]['silence']]
|
|
_say.append(words[_lang]['silence'])
|
|
if CONFIG['SYSTEMS'][system]['DEFAULT_REFLECTOR'] > 0:
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['linkedto'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['to'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['silence'])
|
|
|
|
for number in str(CONFIG['SYSTEMS'][system]['DEFAULT_REFLECTOR']):
|
|
_say.append(words[_lang][number])
|
|
_say.append(words[_lang]['silence'])
|
|
else:
|
|
_say.append(words[_lang]['notlinked'])
|
|
|
|
_say.append(words[_lang]['silence'])
|
|
|
|
speech = pkt_gen(_source_id, _nine, bytes_4(9), 1, _say)
|
|
|
|
sleep(1)
|
|
_slot = systems[system].STATUS[2]
|
|
while True:
|
|
try:
|
|
pkt = next(speech)
|
|
except StopIteration:
|
|
break
|
|
#Packet every 60ms
|
|
sleep(0.058)
|
|
_stream_id = pkt[16:20]
|
|
_pkt_time = time()
|
|
reactor.callFromThread(sendVoicePacket,systems[system],pkt,_source_id,_nine,_slot)
|
|
logger.debug('(%s) disconnected voice thread end',system)
|
|
|
|
def playFileOnRequest(self,fileNumber):
|
|
system = self._system
|
|
_lang = CONFIG['SYSTEMS'][system]['ANNOUNCEMENT_LANGUAGE']
|
|
_nine = bytes_3(9)
|
|
_source_id = bytes_3(5000)
|
|
logger.debug('(%s) Sending contents of AMBE file: %s',system,fileNumber)
|
|
sleep(1)
|
|
_say = []
|
|
try:
|
|
_say.append(AMBEobj.readSingleFile('/'+_lang+'/'+str(fileNumber)+'.ambe'))
|
|
except IOError:
|
|
logger.warning('(%s) cannot read file for number %s',system,fileNumber)
|
|
return
|
|
speech = pkt_gen(_source_id, _nine, bytes_4(9), 1, _say)
|
|
sleep(1)
|
|
_slot = systems[system].STATUS[2]
|
|
while True:
|
|
try:
|
|
pkt = next(speech)
|
|
except StopIteration:
|
|
break
|
|
#Packet every 60ms
|
|
sleep(0.058)
|
|
_stream_id = pkt[16:20]
|
|
_pkt_time = time()
|
|
reactor.callFromThread(sendVoicePacket,self,pkt,_source_id,_nine,_slot)
|
|
logger.debug('(%s) Sending AMBE file %s end',system,fileNumber)
|
|
|
|
|
|
|
|
def threadIdent():
|
|
logger.debug('(IDENT) starting ident thread')
|
|
reactor.callInThread(ident)
|
|
|
|
def threadedMysql():
|
|
logger.debug('(MYSQL) Starting MySQL thread')
|
|
reactor.callInThread(mysqlGetConfig)
|
|
|
|
def ident():
|
|
for system in systems:
|
|
if CONFIG['SYSTEMS'][system]['MODE'] != 'MASTER':
|
|
continue
|
|
if CONFIG['SYSTEMS'][system]['VOICE_IDENT'] == True:
|
|
_lang = CONFIG['SYSTEMS'][system]['ANNOUNCEMENT_LANGUAGE']
|
|
if CONFIG['SYSTEMS'][system]['MAX_PEERS'] > 1:
|
|
logger.debug("(IDENT) %s System has MAX_PEERS > 1, skipping",system)
|
|
continue
|
|
_callsign = False
|
|
for _peerid in CONFIG['SYSTEMS'][system]['PEERS']:
|
|
if CONFIG['SYSTEMS'][system]['PEERS'][_peerid]['CALLSIGN']:
|
|
_callsign = CONFIG['SYSTEMS'][system]['PEERS'][_peerid]['CALLSIGN'].decode()
|
|
if not _callsign:
|
|
logger.debug("(IDENT) %s System has no peers or no recorded callsign (%s), skipping",system,_callsign)
|
|
continue
|
|
_slot = systems[system].STATUS[2]
|
|
#If slot is idle for RX and TX
|
|
#print("RX:"+str(_slot['RX_TYPE'])+" TX:"+str(_slot['TX_TYPE'])+" TIME:"+str(time() - _slot['TX_TIME']))
|
|
if (_slot['RX_TYPE'] == HBPF_SLT_VTERM) and (_slot['TX_TYPE'] == HBPF_SLT_VTERM) and (time() - _slot['TX_TIME'] > CONFIG['SYSTEMS'][system]['GROUP_HANGTIME']):
|
|
#_stream_id = hex_str_4(1234567)
|
|
logger.info('(%s) System idle. Sending voice ident',system)
|
|
_say = [words[_lang]['silence']]
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['this-is'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['silence'])
|
|
|
|
_systemcs = re.sub(r'\W+', '', _callsign)
|
|
_systemcs.upper()
|
|
for character in _systemcs:
|
|
_say.append(words[_lang][character])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['silence'])
|
|
|
|
_say.append(words[_lang]['freedmr'])
|
|
|
|
#test
|
|
#_say.append(AMBEobj.readSingleFile('alpha.ambe'))
|
|
_all_call = bytes_3(16777215)
|
|
_source_id= bytes_3(5000)
|
|
speech = pkt_gen(_source_id, _all_call, bytes_4(16777215), 1, _say)
|
|
|
|
sleep(1)
|
|
_slot = systems[system].STATUS[2]
|
|
while True:
|
|
try:
|
|
pkt = next(speech)
|
|
except StopIteration:
|
|
break
|
|
#Packet every 60ms
|
|
sleep(0.058)
|
|
|
|
_stream_id = pkt[16:20]
|
|
_pkt_time = time()
|
|
reactor.callFromThread(sendVoicePacket,systems[system],pkt,_source_id,_all_call,_slot)
|
|
|
|
def options_config():
|
|
logger.debug('(OPTIONS) Running options parser')
|
|
for _system in CONFIG['SYSTEMS']:
|
|
try:
|
|
if CONFIG['SYSTEMS'][_system]['MODE'] != 'MASTER':
|
|
continue
|
|
if CONFIG['SYSTEMS'][_system]['ENABLED'] == True:
|
|
if 'OPTIONS' in CONFIG['SYSTEMS'][_system]:
|
|
_options = {}
|
|
CONFIG['SYSTEMS'][_system]['OPTIONS'] = CONFIG['SYSTEMS'][_system]['OPTIONS'].rstrip('\x00')
|
|
CONFIG['SYSTEMS'][_system]['OPTIONS'] = CONFIG['SYSTEMS'][_system]['OPTIONS'].encode('ascii', 'ignore').decode()
|
|
CONFIG['SYSTEMS'][_system]['OPTIONS'] = re.sub("\'","",CONFIG['SYSTEMS'][_system]['OPTIONS'])
|
|
CONFIG['SYSTEMS'][_system]['OPTIONS'] = re.sub("\"","",CONFIG['SYSTEMS'][_system]['OPTIONS'])
|
|
for x in CONFIG['SYSTEMS'][_system]['OPTIONS'].split(";"):
|
|
try:
|
|
k,v = x.split('=')
|
|
except ValueError:
|
|
logger.debug('(OPTIONS) Value error %s ignoring',_system)
|
|
continue
|
|
_options[k] = v
|
|
logger.debug('(OPTIONS) Options found for %s',_system)
|
|
|
|
if 'DIAL' in _options:
|
|
_options['DEFAULT_REFLECTOR'] = _options.pop('DIAL')
|
|
if 'TIMER' in _options:
|
|
_options['DEFAULT_UA_TIMER'] = _options.pop('TIMER')
|
|
if 'TS1' in _options:
|
|
_options['TS1_STATIC'] = _options.pop('TS1')
|
|
if 'TS2' in _options:
|
|
_options['TS2_STATIC'] = _options.pop('TS2')
|
|
|
|
#DMR+ style options
|
|
if 'StartRef' in _options:
|
|
_options['DEFAULT_REFLECTOR'] = _options.pop('StartRef')
|
|
if 'RelinkTime' in _options:
|
|
_options['DEFAULT_UA_TIMER'] = _options.pop('RelinkTime')
|
|
if 'TS1_1' in _options:
|
|
_options['TS1_STATIC'] = _options.pop('TS1_1')
|
|
if 'TS1_2' in _options:
|
|
_options['TS1_STATIC'] = _options['TS1_STATIC'] + ',' + _options.pop('TS1_2')
|
|
if 'TS1_3' in _options:
|
|
_options['TS1_STATIC'] = _options['TS1_STATIC'] + ',' + _options.pop('TS1_3')
|
|
if 'TS1_4' in _options:
|
|
_options['TS1_STATIC'] = _options['TS1_STATIC'] + ',' + _options.pop('TS1_4')
|
|
if 'TS1_5' in _options:
|
|
_options['TS1_STATIC'] = _options['TS1_STATIC'] + ',' + _options.pop('TS1_5')
|
|
if 'TS1_6' in _options:
|
|
_options['TS1_STATIC'] = _options['TS1_STATIC'] + ',' + _options.pop('TS1_6')
|
|
if 'TS1_7' in _options:
|
|
_options['TS1_STATIC'] = _options['TS1_STATIC'] + ',' + _options.pop('TS1_7')
|
|
if 'TS1_8' in _options:
|
|
_options['TS1_STATIC'] = _options['TS1_STATIC'] + ',' + _options.pop('TS1_8')
|
|
if 'TS1_9' in _options:
|
|
_options['TS1_STATIC'] = _options['TS1_STATIC'] + ',' + _options.pop('TS1_9')
|
|
if 'TS2_2' in _options:
|
|
_options['TS2_STATIC'] = _options.pop('TS2_1')
|
|
if 'TS2_2' in _options:
|
|
_options['TS2_STATIC'] = _options['TS2_STATIC'] + ',' + _options.pop('TS2_2')
|
|
if 'TS2_3' in _options:
|
|
_options['TS2_STATIC'] = _options['TS2_STATIC'] + ',' + _options.pop('TS2_3')
|
|
if 'TS2_4' in _options:
|
|
_options['TS2_STATIC'] = _options['TS2_STATIC'] + ',' + _options.pop('TS2_4')
|
|
if 'TS2_5' in _options:
|
|
_options['TS2_STATIC'] = _options['TS2_STATIC'] + ',' + _options.pop('TS2_5')
|
|
if 'TS2_6' in _options:
|
|
_options['TS2_STATIC'] = _options['TS2_STATIC'] + ',' + _options.pop('TS2_6')
|
|
if 'TS2_7' in _options:
|
|
_options['TS2_STATIC'] = _options['TS2_STATIC'] + ',' + _options.pop('TS2_7')
|
|
if 'TS2_8' in _options:
|
|
_options['TS2_STATIC'] = _options['TS2_STATIC'] + ',' + _options.pop('TS2_8')
|
|
if 'TS2_9' in _options:
|
|
_options['TS2_STATIC'] = _options['TS2_STATIC'] + ',' + _options.pop('TS2_9')
|
|
|
|
if 'UserLink' in _options:
|
|
_options.pop('UserLink')
|
|
|
|
if 'TS1_STATIC' not in _options:
|
|
_options['TS1_STATIC'] = False
|
|
|
|
if 'TS2_STATIC' not in _options:
|
|
_options['TS2_STATIC'] = False
|
|
|
|
if 'DEFAULT_REFLECTOR' not in _options:
|
|
_options['DEFAULT_REFLECTOR'] = 0
|
|
|
|
if 'DEFAULT_UA_TIMER' not in _options:
|
|
_options['DEFAULT_UA_TIMER'] = CONFIG['SYSTEMS'][_system]['DEFAULT_UA_TIMER']
|
|
|
|
if 'VOICE' in _options and (CONFIG['SYSTEMS'][_system]['VOICE_IDENT'] != bool(int(_options['VOICE']))):
|
|
CONFIG['SYSTEMS'][_system]['VOICE_IDENT'] = bool(int(_options['VOICE']))
|
|
logger.debug("(OPTIONS) %s - Setting voice ident to %s",_system,CONFIG['SYSTEMS'][_system]['VOICE_IDENT'])
|
|
|
|
if 'LANG' in _options and _options['LANG'] in words and _options['LANG'] != CONFIG['SYSTEMS'][_system]['ANNOUNCEMENT_LANGUAGE'] :
|
|
CONFIG['SYSTEMS'][_system]['ANNOUNCEMENT_LANGUAGE'] = _options['LANG']
|
|
logger.debug("(OPTIONS) %s - Setting voice language to %s",_system,CONFIG['SYSTEMS'][_system]['ANNOUNCEMENT_LANGUAGE'])
|
|
|
|
if 'SINGLE' in _options and (CONFIG['SYSTEMS'][_system]['SINGLE_MODE'] != bool(int(_options['SINGLE']))):
|
|
CONFIG['SYSTEMS'][_system]['SINGLE_MODE'] = bool(int(_options['SINGLE']))
|
|
logger.debug("(OPTIONS) %s - Setting SINGLE_MODE to %s",_system,CONFIG['SYSTEMS'][_system]['SINGLE_MODE'])
|
|
|
|
if 'TS1_STATIC' not in _options or 'TS2_STATIC' not in _options or 'DEFAULT_REFLECTOR' not in _options or 'DEFAULT_UA_TIMER' not in _options:
|
|
logger.debug('(OPTIONS) %s - Required field missing, ignoring',_system)
|
|
continue
|
|
|
|
if _options['TS1_STATIC'] == '':
|
|
_options['TS1_STATIC'] = False
|
|
if _options['TS2_STATIC'] == '':
|
|
_options['TS2_STATIC'] = False
|
|
|
|
if _options['TS1_STATIC']:
|
|
re.sub("\s","",_options['TS1_STATIC'])
|
|
if re.search("![\d\,]",_options['TS1_STATIC']):
|
|
logger.debug('(OPTIONS) %s - TS1_STATIC contains characters other than numbers and comma, ignoring',_system)
|
|
continue
|
|
|
|
if _options['TS2_STATIC']:
|
|
re.sub("\s","",_options['TS2_STATIC'])
|
|
if re.search("![\d\,]",_options['TS2_STATIC']):
|
|
logger.debug('(OPTIONS) %s - TS2_STATIC contains characters other than numbers and comma, ignoring',_system)
|
|
continue
|
|
|
|
if isinstance(_options['DEFAULT_REFLECTOR'], str) and not _options['DEFAULT_REFLECTOR'].isdigit():
|
|
logger.debug('(OPTIONS) %s - DEFAULT_UA_TIMER is not an integer, ignoring',_system)
|
|
continue
|
|
|
|
if isinstance(_options['DEFAULT_UA_TIMER'], str) and not _options['DEFAULT_UA_TIMER'].isdigit():
|
|
logger.debug('(OPTIONS) %s - DEFAULT_REFLECTOR is not an integer, ignoring',_system)
|
|
continue
|
|
|
|
_tmout = int(_options['DEFAULT_UA_TIMER'])
|
|
|
|
if int(_options['DEFAULT_UA_TIMER']) != CONFIG['SYSTEMS'][_system]['DEFAULT_UA_TIMER']:
|
|
logger.debug('(OPTIONS) %s Updating DEFAULT_UA_TIMER for existing bridges.',_system)
|
|
remove_bridge_system(_system)
|
|
for _bridge in BRIDGES:
|
|
ts1 = False
|
|
ts2 = False
|
|
for i,e in enumerate(BRIDGES[_bridge]):
|
|
if e['SYSTEM'] == _system and e['TS'] == 1:
|
|
ts1 = True
|
|
if e['SYSTEM'] == _system and e['TS'] == 2:
|
|
ts2 = True
|
|
if _bridge[0:1] != '#':
|
|
if ts1 == False:
|
|
BRIDGES[_bridge].append({'SYSTEM': _system, 'TS': 1, 'TGID': bytes_3(int(_bridge)),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(int(_bridge)),],'RESET': [], 'TIMER': time()})
|
|
if ts2 == False:
|
|
BRIDGES[_bridge].append({'SYSTEM': _system, 'TS': 2, 'TGID': bytes_3(int(_bridge)),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(int(_bridge)),],'RESET': [], 'TIMER': time()})
|
|
else:
|
|
if ts2 == False:
|
|
BRIDGES[_bridge].append({'SYSTEM': _system, 'TS': 2, 'TGID': bytes_3(9),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [bytes_3(4000)],'ON': [],'RESET': [], 'TIMER': time()})
|
|
|
|
if int(_options['DEFAULT_REFLECTOR']) != CONFIG['SYSTEMS'][_system]['DEFAULT_REFLECTOR']:
|
|
if int(_options['DEFAULT_REFLECTOR']) > 0:
|
|
logger.debug('(OPTIONS) %s default reflector changed, updating',_system)
|
|
reset_default_reflector(CONFIG['SYSTEMS'][_system]['DEFAULT_REFLECTOR'],_tmout,_system)
|
|
make_default_reflector(int(_options['DEFAULT_REFLECTOR']),_tmout,_system)
|
|
else:
|
|
logger.debug('(OPTIONS) %s default reflector disabled, updating',_system)
|
|
reset_default_reflector(int(_options['DEFAULT_REFLECTOR']),_tmout,_system)
|
|
|
|
if _options['TS1_STATIC'] != CONFIG['SYSTEMS'][_system]['TS1_STATIC']:
|
|
_tmout = int(_options['DEFAULT_UA_TIMER'])
|
|
logger.debug('(OPTIONS) %s TS1 static TGs changed, updating',_system)
|
|
ts1 = []
|
|
if CONFIG['SYSTEMS'][_system]['TS1_STATIC']:
|
|
ts1 = CONFIG['SYSTEMS'][_system]['TS1_STATIC'].split(',')
|
|
for tg in ts1:
|
|
if not tg:
|
|
continue
|
|
tg = int(tg)
|
|
reset_static_tg(tg,1,_tmout,_system)
|
|
ts1 = []
|
|
if _options['TS1_STATIC']:
|
|
ts1 = _options['TS1_STATIC'].split(',')
|
|
for tg in ts1:
|
|
if not tg:
|
|
continue
|
|
tg = int(tg)
|
|
make_static_tg(tg,1,_tmout,_system)
|
|
|
|
if _options['TS2_STATIC'] != CONFIG['SYSTEMS'][_system]['TS2_STATIC']:
|
|
_tmout = int(_options['DEFAULT_UA_TIMER'])
|
|
logger.debug('(OPTIONS) %s TS2 static TGs changed, updating',_system)
|
|
ts2 = []
|
|
if CONFIG['SYSTEMS'][_system]['TS2_STATIC']:
|
|
ts2 = CONFIG['SYSTEMS'][_system]['TS2_STATIC'].split(',')
|
|
for tg in ts2:
|
|
if not tg:
|
|
continue
|
|
tg = int(tg)
|
|
reset_static_tg(tg,2,_tmout,_system)
|
|
ts2 = []
|
|
if _options['TS2_STATIC']:
|
|
ts2 = _options['TS2_STATIC'].split(',')
|
|
for tg in ts2:
|
|
if not tg:
|
|
continue
|
|
tg = int(tg)
|
|
make_static_tg(tg,2,_tmout,_system)
|
|
|
|
CONFIG['SYSTEMS'][_system]['TS1_STATIC'] = _options['TS1_STATIC']
|
|
CONFIG['SYSTEMS'][_system]['TS2_STATIC'] = _options['TS2_STATIC']
|
|
CONFIG['SYSTEMS'][_system]['DEFAULT_REFLECTOR'] = int(_options['DEFAULT_REFLECTOR'])
|
|
CONFIG['SYSTEMS'][_system]['DEFAULT_UA_TIMER'] = int(_options['DEFAULT_UA_TIMER'])
|
|
except Exception:
|
|
logger.exception('(OPTIONS) caught exception:')
|
|
continue
|
|
|
|
def mysqlGetConfig():
|
|
logger.debug('(MYSQL) Periodic config check')
|
|
SQLGETCONFIG = {}
|
|
if sql.con():
|
|
logger.debug('(MYSQL) reading config from database')
|
|
try:
|
|
SQLGETCONFIG = sql.getConfig()
|
|
except:
|
|
logger.debug('(MYSQL) problem with SQL query, aborting')
|
|
sql.close()
|
|
return
|
|
else:
|
|
logger.debug('(MYSQL) problem connecting to SQL server, aborting')
|
|
sql.close()
|
|
return
|
|
|
|
sql.close()
|
|
reactor.callFromThread(mysql_config_check,SQLGETCONFIG)
|
|
|
|
|
|
def mysql_config_check(SQLGETCONFIG):
|
|
|
|
SQLCONFIG = SQLGETCONFIG
|
|
for system in SQLGETCONFIG:
|
|
if system not in CONFIG['SYSTEMS']:
|
|
if SQLCONFIG[system]['ENABLED']:
|
|
logger.debug('(MYSQL) new enabled system %s, starting HBP listener',system)
|
|
CONFIG['SYSTEMS'][system] = SQLCONFIG[system]
|
|
systems[system] = routerHBP(system, CONFIG, report_server)
|
|
listeningPorts[system] = reactor.listenUDP(CONFIG['SYSTEMS'][system]['PORT'], systems[system], interface=CONFIG['SYSTEMS'][system]['IP'])
|
|
else:
|
|
logger.debug('(MYSQL) new disabled system %s',system)
|
|
_tmout = SQLCONFIG[system]['DEFAULT_UA_TIMER']
|
|
#Do ACL processing
|
|
# Subscriber and TGID ACLs
|
|
logger.debug('(MYSQL) building ACLs')
|
|
# Registration ACLs
|
|
SQLCONFIG[system]['REG_ACL'] = acl_build(SQLCONFIG[system]['REG_ACL'], PEER_MAX)
|
|
for acl in ['SUB_ACL', 'TG1_ACL', 'TG2_ACL']:
|
|
SQLCONFIG[system][acl] = acl_build(SQLCONFIG[system][acl], ID_MAX)
|
|
|
|
#Add system to bridges
|
|
if SQLCONFIG[system]['ENABLED']:
|
|
logger.debug('(MYSQL) adding new system to static bridges')
|
|
for _bridge in BRIDGES:
|
|
ts1 = False
|
|
ts2 = False
|
|
for i,e in enumerate(BRIDGES[_bridge]):
|
|
if e['SYSTEM'] == system and e['TS'] == 1:
|
|
ts1 = True
|
|
if e['SYSTEM'] == system and e['TS'] == 2:
|
|
ts2 = True
|
|
if _bridge[0:1] != '#':
|
|
if ts1 == False:
|
|
BRIDGES[_bridge].append({'SYSTEM': system, 'TS': 1, 'TGID': bytes_3(int(_bridge)),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(int(_bridge)),],'RESET': [], 'TIMER': time()})
|
|
if ts2 == False:
|
|
BRIDGES[_bridge].append({'SYSTEM': system, 'TS': 2, 'TGID': bytes_3(int(_bridge)),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(int(_bridge)),],'RESET': [], 'TIMER': time()})
|
|
else:
|
|
if ts2 == False:
|
|
BRIDGES[_bridge].append({'SYSTEM': system, 'TS': 2, 'TGID': bytes_3(9),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [bytes_3(4000)],'ON': [],'RESET': [], 'TIMER': time()})
|
|
|
|
if SQLCONFIG[system]['DEFAULT_REFLECTOR'] > 0:
|
|
logger.debug('(MYSQL) %s setting default reflector',system)
|
|
make_default_reflector(SQLCONFIG[system]['DEFAULT_REFLECTOR'],_tmout,system)
|
|
|
|
if SQLCONFIG[system]['TS1_STATIC']:
|
|
logger.debug('(MYSQL) %s setting static TGs on TS1',system)
|
|
ts1 = SQLCONFIG[system]['TS1_STATIC'].split(',')
|
|
for tg in ts1:
|
|
if not tg:
|
|
continue
|
|
tg = int(tg)
|
|
make_static_tg(tg,1,_tmout,system)
|
|
|
|
if SQLCONFIG[system]['TS2_STATIC']:
|
|
logger.debug('(MYSQL) %s setting static TGs on TS2',system)
|
|
ts2 = SQLCONFIG[system]['TS2_STATIC'].split(',')
|
|
for tg in ts2:
|
|
if not tg:
|
|
continue
|
|
tg = int(tg)
|
|
make_static_tg(tg,2,_tmout,system)
|
|
|
|
continue
|
|
|
|
#Preserve options line
|
|
if 'OPTIONS' in CONFIG['SYSTEMS'][system]:
|
|
SQLCONFIG[system]['OPTIONS'] = CONFIG['SYSTEMS'][system]['OPTIONS']
|
|
SQLCONFIG[system]['TS1_STATIC'] = CONFIG['SYSTEMS'][system]['TS1_STATIC']
|
|
SQLCONFIG[system]['TS2_STATIC'] = CONFIG['SYSTEMS'][system]['TS2_STATIC']
|
|
SQLCONFIG[system]['DEFAULT_UA_TIMER'] = CONFIG['SYSTEMS'][system]['DEFAULT_UA_TIMER']
|
|
SQLCONFIG[system]['DEFAULT_REFLECTOR'] = CONFIG['SYSTEMS'][system]['DEFAULT_REFLECTOR']
|
|
|
|
#logger.debug('(MYSQL) %s has HBP Options line - skipping',system)
|
|
#continue
|
|
|
|
|
|
if SQLCONFIG[system]['ENABLED'] == False and CONFIG['SYSTEMS'][system]['ENABLED'] == True:
|
|
logger.debug('(MYSQL) %s changed from enabled to disabled, killing HBP listener and removing from bridges',system)
|
|
systems[system].master_dereg()
|
|
if systems[system]._system_maintenance is not None and systems[system]._system_maintenance.running == True:
|
|
systems[system]._system_maintenance.stop()
|
|
systems[system]._system_maintenance = None
|
|
remove_bridge_system(system)
|
|
listeningPorts[system].stopListening()
|
|
|
|
if CONFIG['SYSTEMS'][system]['ENABLED'] == False and SQLCONFIG[system]['ENABLED'] == True:
|
|
logger.debug('(MYSQL) %s changed from disabled to enabled, starting HBP listener',system)
|
|
systems[system] = routerHBP(system, CONFIG, report_server)
|
|
listeningPorts[system] = 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])
|
|
logger.debug('(MYSQL) adding new system to static bridges')
|
|
_tmout = SQLCONFIG[system]['DEFAULT_UA_TIMER']
|
|
for _bridge in BRIDGES:
|
|
ts1 = False
|
|
ts2 = False
|
|
for i,e in enumerate(BRIDGES[_bridge]):
|
|
if e['SYSTEM'] == system and e['TS'] == 1:
|
|
ts1 = True
|
|
if e['SYSTEM'] == system and e['TS'] == 2:
|
|
ts2 = True
|
|
if _bridge[0:1] != '#':
|
|
if ts1 == False:
|
|
BRIDGES[_bridge].append({'SYSTEM': system, 'TS': 1, 'TGID': bytes_3(int(_bridge)),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(int(_bridge)),],'RESET': [], 'TIMER': time()})
|
|
if ts2 == False:
|
|
BRIDGES[_bridge].append({'SYSTEM': system, 'TS': 2, 'TGID': bytes_3(int(_bridge)),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(int(_bridge)),],'RESET': [], 'TIMER': time()})
|
|
else:
|
|
if ts2 == False:
|
|
BRIDGES[_bridge].append({'SYSTEM': system, 'TS': 2, 'TGID': bytes_3(9),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [bytes_3(4000)],'ON': [],'RESET': [], 'TIMER': time()})
|
|
|
|
|
|
if SQLCONFIG[system]['DEFAULT_REFLECTOR'] > 0:
|
|
if 'OPTIONS' not in SQLCONFIG[system]:
|
|
logger.debug('(MYSQL) %s setting default reflector',system)
|
|
make_default_reflector(SQLCONFIG[system]['DEFAULT_REFLECTOR'],_tmout,system)
|
|
|
|
if SQLCONFIG[system]['TS1_STATIC']:
|
|
if 'OPTIONS' not in SQLCONFIG[system]:
|
|
logger.debug('(MYSQL) %s setting static TGs on TS1',system)
|
|
ts1 = SQLCONFIG[system]['TS1_STATIC'].split(',')
|
|
for tg in ts1:
|
|
if not tg:
|
|
continue
|
|
tg = int(tg)
|
|
make_static_tg(tg,1,_tmout,system)
|
|
|
|
if SQLCONFIG[system]['TS2_STATIC']:
|
|
logger.debug('(MYSQL) %s setting static TGs on TS2',system)
|
|
ts2 = SQLCONFIG[system]['TS2_STATIC'].split(',')
|
|
for tg in ts2:
|
|
if not tg:
|
|
continue
|
|
tg = int(tg)
|
|
make_static_tg(tg,2,_tmout,system)
|
|
|
|
if SQLCONFIG[system]['DEFAULT_UA_TIMER'] != CONFIG['SYSTEMS'][system]['DEFAULT_UA_TIMER']:
|
|
if 'OPTIONS' not in CONFIG['SYSTEMS'][system]:
|
|
logger.debug('(MYSQL) %s DEFAULT_UA_TIMER changed. Updating bridges.',system)
|
|
remove_bridge_system(system)
|
|
for _bridge in BRIDGES:
|
|
ts1 = False
|
|
ts2 = False
|
|
_tmout = CONFIG['SYSTEMS'][system][DEFAULT_UA_TIMER]
|
|
for i,e in enumerate(BRIDGES[_bridge]):
|
|
if e['SYSTEM'] == system and e['TS'] == 1:
|
|
ts1 = True
|
|
if e['SYSTEM'] == system and e['TS'] == 2:
|
|
ts2 = True
|
|
if _bridge[0:1] != '#':
|
|
if ts1 == False:
|
|
BRIDGES[_bridge].append({'SYSTEM': system, 'TS': 1, 'TGID': bytes_3(int(_bridge)),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(int(_bridge)),],'RESET': [], 'TIMER': time()})
|
|
if ts2 == False:
|
|
BRIDGES[_bridge].append({'SYSTEM': system, 'TS': 2, 'TGID': bytes_3(int(_bridge)),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [],'ON': [bytes_3(int(_bridge)),],'RESET': [], 'TIMER': time()})
|
|
else:
|
|
if ts2 == False:
|
|
BRIDGES[_bridge].append({'SYSTEM': system, 'TS': 2, 'TGID': bytes_3(9),'ACTIVE': False,'TIMEOUT': _tmout * 60,'TO_TYPE': 'ON','OFF': [bytes_3(4000)],'ON': [],'RESET': [], 'TIMER': time()})
|
|
|
|
|
|
if SQLCONFIG[system]['DEFAULT_REFLECTOR'] > 0:
|
|
# if 'OPTIONS' not in SQLCONFIG[system]:
|
|
logger.debug('(MYSQL) %s setting default reflector',system)
|
|
make_default_reflector(SQLCONFIG[system]['DEFAULT_REFLECTOR'],_tmout,system)
|
|
|
|
if SQLCONFIG[system]['TS1_STATIC']:
|
|
# if 'OPTIONS' not in SQLCONFIG[system]:
|
|
logger.debug('(MYSQL) %s setting static TGs on TS1',system)
|
|
ts1 = SQLCONFIG[system]['TS1_STATIC'].split(',')
|
|
for tg in ts1:
|
|
if not tg:
|
|
continue
|
|
tg = int(tg)
|
|
make_static_tg(tg,1,_tmout,system)
|
|
|
|
if SQLCONFIG[system]['TS2_STATIC']:
|
|
logger.debug('(MYSQL) %s setting static TGs on TS2',system)
|
|
ts2 = SQLCONFIG[system]['TS2_STATIC'].split(',')
|
|
for tg in ts2:
|
|
if not tg:
|
|
continue
|
|
tg = int(tg)
|
|
make_static_tg(tg,2,_tmout,system)
|
|
|
|
|
|
|
|
if SQLCONFIG[system]['IP'] != CONFIG['SYSTEMS'][system]['IP'] and CONFIG['SYSTEMS'][system]['ENABLED'] == True:
|
|
logger.debug('(MYSQL) %s IP binding changed on enabled system, killing HBP listener. Will restart in 1 minute',system)
|
|
systems[system].master_dereg()
|
|
if systems[system]._system_maintenance is not None and systems[system]._system_maintenance.running == True:
|
|
systems[system]._system_maintenance.stop()
|
|
systems[system]._system_maintenance = None
|
|
listeningPorts[system].stopListening()
|
|
SQLCONFIG[system]['ENABLED'] = False
|
|
|
|
if SQLCONFIG[system]['PORT'] != CONFIG['SYSTEMS'][system]['PORT'] and CONFIG['SYSTEMS'][system]['ENABLED'] == True:
|
|
logger.debug('(MYSQL) %s Port binding changed on enabled system, killing HBP listener. Will restart in 1 minute',system)
|
|
systems[system].master_dereg()
|
|
if systems[system]._system_maintenance is not None and systems[system]._system_maintenance.running == True:
|
|
systems[system]._system_maintenance.stop()
|
|
systems[system]._system_maintenance = None
|
|
listeningPorts[system].stopListening()
|
|
SQLCONFIG[system]['ENABLED'] = False
|
|
|
|
if SQLCONFIG[system]['MAX_PEERS'] != CONFIG['SYSTEMS'][system]['MAX_PEERS'] and CONFIG['SYSTEMS'][system]['ENABLED'] == True:
|
|
logger.debug('(MYSQL) %s MAX_PEERS changed on enabled system, killing HBP listener. Will restart in 1 minute',system)
|
|
systems[system].master_dereg()
|
|
if systems[system]._system_maintenance is not None and systems[system]._system_maintenance.running == True:
|
|
systems[system]._system_maintenance.stop()
|
|
systems[system]._system_maintenance = None
|
|
listeningPorts[system].stopListening()
|
|
SQLCONFIG[system]['ENABLED'] = False
|
|
|
|
if SQLCONFIG[system]['PASSPHRASE'] != CONFIG['SYSTEMS'][system]['PASSPHRASE'] and CONFIG['SYSTEMS'][system]['ENABLED'] == True:
|
|
logger.debug('(MYSQL) %s Passphrase changed on enabled system. Kicking peers',system)
|
|
systems[system].master_dereg()
|
|
|
|
if SQLCONFIG[system]['DEFAULT_REFLECTOR'] != CONFIG['SYSTEMS'][system]['DEFAULT_REFLECTOR']:
|
|
if 'OPTIONS' not in SQLCONFIG[system]:
|
|
_tmout = SQLCONFIG[system]['DEFAULT_UA_TIMER']
|
|
if SQLCONFIG[system]['DEFAULT_REFLECTOR'] > 0:
|
|
logger.debug('(MYSQL) %s default reflector changed, updating',system)
|
|
reset_default_reflector(CONFIG['SYSTEMS'][system]['DEFAULT_REFLECTOR'],_tmout,system)
|
|
make_default_reflector(SQLCONFIG[system]['DEFAULT_REFLECTOR'],_tmout,system)
|
|
else:
|
|
logger.debug('(MYSQL) %s default reflector disabled, updating',system)
|
|
reset_default_reflector(CONFIG['SYSTEMS'][system]['DEFAULT_REFLECTOR'],_tmout,system)
|
|
|
|
if SQLCONFIG[system]['TS1_STATIC'] != CONFIG['SYSTEMS'][system]['TS1_STATIC']:
|
|
if 'OPTIONS' not in CONFIG['SYSTEMS'][system]:
|
|
_tmout = SQLCONFIG[system]['DEFAULT_UA_TIMER']
|
|
logger.debug('(MYSQL) %s TS1 static TGs changed, updating',system)
|
|
ts1 = []
|
|
if CONFIG['SYSTEMS'][system]['TS1_STATIC']:
|
|
ts1 = CONFIG['SYSTEMS'][system]['TS1_STATIC'].split(',')
|
|
for tg in ts1:
|
|
if not tg:
|
|
continue
|
|
tg = int(tg)
|
|
reset_static_tg(tg,1,_tmout,system)
|
|
ts1 = []
|
|
if SQLCONFIG[system]['TS1_STATIC']:
|
|
ts1 = SQLCONFIG[system]['TS1_STATIC'].split(',')
|
|
for tg in ts1:
|
|
if not tg:
|
|
continue
|
|
tg = int(tg)
|
|
make_static_tg(tg,1,_tmout,system)
|
|
|
|
if SQLCONFIG[system]['TS2_STATIC'] != CONFIG['SYSTEMS'][system]['TS2_STATIC']:
|
|
if 'OPTIONS' not in CONFIG['SYSTEMS'][system]:
|
|
_tmout = SQLCONFIG[system]['DEFAULT_UA_TIMER']
|
|
logger.debug('(MYSQL) %s TS2 static TGs changed, updating',system)
|
|
ts2 = []
|
|
if CONFIG['SYSTEMS'][system]['TS2_STATIC']:
|
|
ts2 = CONFIG['SYSTEMS'][system]['TS2_STATIC'].split(',')
|
|
for tg in ts2:
|
|
if not tg:
|
|
continue
|
|
tg = int(tg)
|
|
reset_static_tg(tg,2,_tmout,system)
|
|
ts2 = []
|
|
if SQLCONFIG[system]['TS2_STATIC']:
|
|
ts2 = SQLCONFIG[system]['TS2_STATIC'].split(',')
|
|
for tg in ts2:
|
|
if not tg:
|
|
continue
|
|
tg = int(tg)
|
|
make_static_tg(tg,2,_tmout,system)
|
|
|
|
if SQLCONFIG[system]['ANNOUNCEMENT_LANGUAGE'] != CONFIG['SYSTEMS'][system]['ANNOUNCEMENT_LANGUAGE']:
|
|
logger.debug('(MYSQL) %s announcement language changed to %s',system, SQLCONFIG[system]['ANNOUNCEMENT_LANGUAGE'])
|
|
|
|
#Rebuild ACLs
|
|
SQLCONFIG[system]['REG_ACL'] = acl_build(SQLCONFIG[system]['REG_ACL'], PEER_MAX)
|
|
SQLCONFIG[system]['SUB_ACL'] = acl_build(SQLCONFIG[system]['SUB_ACL'], ID_MAX)
|
|
SQLCONFIG[system]['TG1_ACL'] = acl_build(SQLCONFIG[system]['TG1_ACL'], ID_MAX)
|
|
SQLCONFIG[system]['TG2_ACL'] = acl_build(SQLCONFIG[system]['TG2_ACL'], ID_MAX)
|
|
|
|
if SQLCONFIG[system]['REG_ACL'] != CONFIG['SYSTEMS'][system]['REG_ACL']:
|
|
logger.debug('(MYSQL) registration ACL changed')
|
|
if SQLCONFIG[system]['SUB_ACL'] != CONFIG['SYSTEMS'][system]['SUB_ACL']:
|
|
logger.debug('(MYSQL) subscriber ACL changed')
|
|
if SQLCONFIG[system]['TG1_ACL'] != CONFIG['SYSTEMS'][system]['TG1_ACL']:
|
|
logger.debug('(MYSQL) TG1 ACL changed')
|
|
if SQLCONFIG[system]['TG2_ACL'] != CONFIG['SYSTEMS'][system]['TG2_ACL']:
|
|
logger.debug('(MYSQL) TG2 ACL changed')
|
|
|
|
#Preserve peers list
|
|
if system in CONFIG['SYSTEMS'] and CONFIG['SYSTEMS'][system]['ENABLED'] and 'PEERS' in CONFIG['SYSTEMS'][system] :
|
|
SQLCONFIG[system]['PEERS'] = CONFIG['SYSTEMS'][system]['PEERS']
|
|
CONFIG['SYSTEMS'][system].update(SQLCONFIG[system])
|
|
else:
|
|
CONFIG['SYSTEMS'][system].update(SQLCONFIG[system])
|
|
|
|
|
|
#Add MySQL config data to config dict
|
|
#CONFIG['SYSTEMS'].update(SQLCONFIG)
|
|
|
|
SQLCONFIG = {}
|
|
|
|
class routerOBP(OPENBRIDGE):
|
|
|
|
def __init__(self, _name, _config, _report):
|
|
OPENBRIDGE.__init__(self, _name, _config, _report)
|
|
self.STATUS = {}
|
|
|
|
def to_target(self, _peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data, pkt_time, dmrpkt, _bits,_bridge,_system,_noOBP,sysIgnore):
|
|
_sysIgnore = sysIgnore
|
|
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'],_target['TS']) in _sysIgnore:
|
|
#logger.debug("(DEDUP) OBP Source Skipping system %s TS: %s",_target['SYSTEM'],_target['TS'])
|
|
continue
|
|
if _target_system['MODE'] == 'OPENBRIDGE':
|
|
if _noOBP == True:
|
|
continue
|
|
#We want to ignore this system and TS combination if it's called again for this packet
|
|
_sysIgnore.append((_target['SYSTEM'],_target['TS']))
|
|
|
|
#If target has quenched us, don't send
|
|
if ('_bcsq' in _target_system) and (_dst_id in _target_system['_bcsq']) and (_target_system['_bcsq'][_dst_id] == _stream_id):
|
|
#logger.info('(%s) Conference Bridge: %s, is Source Quenched for Stream ID: %s, skipping system: %s TS: %s, TGID: %s', self._system, _bridge, int_id(_stream_id), _target['SYSTEM'], _target['TS'], int_id(_target['TGID']))
|
|
continue
|
|
|
|
#If target has missed 6 (on 1 min) of keepalives, don't send
|
|
if _target_system['ENHANCED_OBP'] and ('_bcka' not in _target_system or _target_system['_bcka'] < pkt_time - 60):
|
|
continue
|
|
|
|
#If talkgroup is prohibited by ACL
|
|
if self._CONFIG['GLOBAL']['USE_ACL']:
|
|
if not acl_check(_target['TGID'], self._CONFIG['GLOBAL']['TG1_ACL']):
|
|
#logger.info('(%s) TGID prohibited by ACL, not sending', _target['SYSTEM'], int_id(_dst_id))
|
|
continue
|
|
|
|
if not acl_check(_target['TGID'],_target_system['TG1_ACL']):
|
|
#logger.info('(%s) TGID prohibited by ACL, not sending', _target['SYSTEM'])
|
|
continue
|
|
|
|
|
|
# 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,
|
|
'RX_PEER': _peer_id,
|
|
|
|
}
|
|
# Generate LCs (full and EMB) for the TX stream
|
|
try:
|
|
dst_lc = b''.join([self.STATUS[_stream_id]['LC'][0:3], _target['TGID'], _rf_src])
|
|
except Exception:
|
|
logger.exception('(to_target) caught exception')
|
|
return
|
|
_target_status[_stream_id]['H_LC'] = bptc.encode_header_lc(dst_lc)
|
|
_target_status[_stream_id]['T_LC'] = bptc.encode_terminator_lc(dst_lc)
|
|
_target_status[_stream_id]['EMB_LC'] = bptc.encode_emblc(dst_lc)
|
|
|
|
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']))
|
|
#Ignore this system and TS pair if it's called again on this packet
|
|
return(_sysIgnore)
|
|
|
|
|
|
def dmrd_received(self, _peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data):
|
|
pkt_time = time()
|
|
dmrpkt = _data[20:53]
|
|
_bits = _data[15]
|
|
|
|
if _call_type == 'group' or _call_type == 'vcsbk':
|
|
# 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,
|
|
'1ST': perf_counter(),
|
|
'lastSeq': False,
|
|
'lastData': False,
|
|
'RX_PEER': _peer_id
|
|
|
|
}
|
|
|
|
# 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'))
|
|
|
|
|
|
else:
|
|
#Finished stream handling#
|
|
if '_fin' in self.STATUS[_stream_id]:
|
|
if '_finlog' not in self.STATUS[_stream_id]:
|
|
logger.warning("(%s) OBP *LoopControl* STREAM ID: %s ALREADY FINISHED FROM THIS SOURCE, IGNORING",self._system, int_id(_stream_id))
|
|
self.STATUS[_stream_id]['_finlog'] = True
|
|
return
|
|
|
|
#LoopControl
|
|
hr_times = {}
|
|
for system in systems:
|
|
# if system == self._system:
|
|
# continue
|
|
if system != self._system and CONFIG['SYSTEMS'][system]['MODE'] != 'OPENBRIDGE':
|
|
for _sysslot in systems[system].STATUS:
|
|
if 'RX_STREAM_ID' in systems[system].STATUS[_sysslot] and _stream_id == systems[system].STATUS[_sysslot]['RX_STREAM_ID']:
|
|
if 'LOOPLOG' not in self.STATUS[_stream_id] or not self.STATUS[_stream_id]['LOOPLOG']:
|
|
logger.warning("(%s) OBP *LoopControl* FIRST HBP: %s, STREAM ID: %s, TG: %s, TS: %s, IGNORE THIS SOURCE",self._system, system, int_id(_stream_id), int_id(_dst_id),_sysslot)
|
|
self.STATUS[_stream_id]['LOOPLOG'] = True
|
|
self.STATUS[_stream_id]['LAST'] = pkt_time
|
|
return
|
|
else:
|
|
#if _stream_id in systems[system].STATUS and systems[system].STATUS[_stream_id]['START'] <= self.STATUS[_stream_id]['START']:
|
|
if _stream_id in systems[system].STATUS and '1ST' in systems[system].STATUS[_stream_id] and systems[system].STATUS[_stream_id]['TGID'] == _dst_id:
|
|
hr_times[system] = systems[system].STATUS[_stream_id]['1ST']
|
|
|
|
#use the minimum perf_counter to ensure
|
|
#We always use only the earliest packet
|
|
fi = min(hr_times, key=hr_times.get, default = False)
|
|
|
|
hr_times = None
|
|
|
|
if not fi:
|
|
logger.warning("(%s) OBP *LoopControl* fi is empty for some reason : %s, STREAM ID: %s, TG: %s, TS: %s",self._system, int_id(_stream_id), int_id(_dst_id),_sysslot)
|
|
return
|
|
|
|
if self._system != fi:
|
|
if 'LOOPLOG' not in self.STATUS[_stream_id] or not self.STATUS[_stream_id]['LOOPLOG']:
|
|
logger.warning("(%s) OBP *LoopControl* FIRST OBP %s, STREAM ID: %s, TG %s, IGNORE THIS SOURCE",self._system, fi, int_id(_stream_id), int_id(_dst_id))
|
|
self.STATUS[_stream_id]['LOOPLOG'] = True
|
|
self.STATUS[_stream_id]['LAST'] = pkt_time
|
|
|
|
if CONFIG['SYSTEMS'][self._system]['ENHANCED_OBP'] and '_bcsq' not in self.STATUS[_stream_id]:
|
|
systems[self._system].send_bcsq(_dst_id,_stream_id)
|
|
#logger.warning("(%s) OBP *BridgeControl* Sent BCSQ , STREAM ID: %s, TG %s",self._system, int_id(_stream_id), int_id(_dst_id))
|
|
self.STATUS[_stream_id]['_bcsq'] = True
|
|
return
|
|
|
|
#Duplicate handling#
|
|
#Duplicate complete packet
|
|
if self.STATUS[_stream_id]['lastData'] and self.STATUS[_stream_id]['lastData'] == _data and _seq > 1:
|
|
logger.warning("(%s) *PacketControl* last packet is a complete duplicate of the previous one, disgarding. Stream ID:, %s TGID: %s",self._system,int_id(_stream_id),int_id(_dst_id))
|
|
return
|
|
#Handle inbound duplicates
|
|
if _seq and _seq == self.STATUS[_stream_id]['lastSeq']:
|
|
logger.warning("(%s) *PacketControl* Duplicate sequence number %s, disgarding. Stream ID:, %s TGID: %s",self._system,_seq,int_id(_stream_id),int_id(_dst_id))
|
|
return
|
|
#Inbound out-of-order packets
|
|
if _seq and self.STATUS[_stream_id]['lastSeq'] and (_seq != 1) and (_seq < self.STATUS[_stream_id]['lastSeq']):
|
|
logger.warning("%s) *PacketControl* Out of order packet - last SEQ: %s, this SEQ: %s, disgarding. Stream ID:, %s TGID: %s ",self._system,self.STATUS[_stream_id]['lastSeq'],_seq,int_id(_stream_id),int_id(_dst_id))
|
|
return
|
|
#Inbound missed packets
|
|
if _seq and self.STATUS[_stream_id]['lastSeq'] and _seq > (self.STATUS[_stream_id]['lastSeq']+1):
|
|
logger.warning("(%s) *PacketControl* Missed packet(s) - last SEQ: %s, this SEQ: %s. Stream ID:, %s TGID: %s ",self._system,self.STATUS[_stream_id]['lastSeq'],_seq,int_id(_stream_id),int_id(_dst_id))
|
|
|
|
#Save this sequence number
|
|
self.STATUS[_stream_id]['lastSeq'] = _seq
|
|
#Save this packet
|
|
self.STATUS[_stream_id]['lastData'] = _data
|
|
|
|
|
|
|
|
self.STATUS[_stream_id]['LAST'] = pkt_time
|
|
|
|
|
|
#Create STAT bridge for unknown TG
|
|
if CONFIG['GLOBAL']['GEN_STAT_BRIDGES']:
|
|
if int_id(_dst_id) >= 5 and int_id(_dst_id) != 9 and (str(int_id(_dst_id)) not in BRIDGES):
|
|
logger.info('(%s) Bridge for STAT TG %s does not exist. Creating',self._system, int_id(_dst_id))
|
|
make_stat_bridge(_dst_id)
|
|
|
|
_sysIgnore = []
|
|
for _bridge in BRIDGES:
|
|
#if _bridge[0:1] != '#':
|
|
#if True:
|
|
for _system in BRIDGES[_bridge]:
|
|
|
|
if _system['SYSTEM'] == self._system and _system['TGID'] == _dst_id and _system['TS'] == _slot and _system['ACTIVE'] == True:
|
|
_sysIgnore = self.to_target(_peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data, pkt_time, dmrpkt, _bits,_bridge,_system,False,_sysIgnore)
|
|
|
|
|
|
# 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'))
|
|
self.STATUS[_stream_id]['_fin'] = True
|
|
#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))
|
|
|
|
#Reset sequence number
|
|
self.STATUS[_stream_id]['lastSeq'] = False
|
|
|
|
class routerHBP(HBSYSTEM):
|
|
|
|
def __init__(self, _name, _config, _report):
|
|
HBSYSTEM.__init__(self, _name, _config, _report)
|
|
# Status information for the system, TS1 & TS2
|
|
# 1 & 2 are "timeslot"
|
|
# In TX_EMB_LC, 2-5 are burst B-E
|
|
self.STATUS = {
|
|
1: {
|
|
'RX_START': time(),
|
|
'TX_START': time(),
|
|
'RX_SEQ': 0,
|
|
'RX_RFS': b'\x00',
|
|
'TX_RFS': b'\x00',
|
|
'RX_PEER': b'\x00',
|
|
'TX_PEER': b'\x00',
|
|
'RX_STREAM_ID': b'\x00',
|
|
'TX_STREAM_ID': b'\x00',
|
|
'RX_TGID': b'\x00\x00\x00',
|
|
'TX_TGID': b'\x00\x00\x00',
|
|
'RX_TIME': time(),
|
|
'TX_TIME': time(),
|
|
'RX_TYPE': HBPF_SLT_VTERM,
|
|
'TX_TYPE': HBPF_SLT_VTERM,
|
|
'RX_LC': b'\x00',
|
|
'TX_H_LC': b'\x00',
|
|
'TX_T_LC': b'\x00',
|
|
'TX_EMB_LC': {
|
|
1: b'\x00',
|
|
2: b'\x00',
|
|
3: b'\x00',
|
|
4: b'\x00',
|
|
},
|
|
'lastSeq': False,
|
|
'lastData': False
|
|
},
|
|
2: {
|
|
'RX_START': time(),
|
|
'TX_START': time(),
|
|
'RX_SEQ': 0,
|
|
'RX_RFS': b'\x00',
|
|
'TX_RFS': b'\x00',
|
|
'RX_PEER': b'\x00',
|
|
'TX_PEER': b'\x00',
|
|
'RX_STREAM_ID': b'\x00',
|
|
'TX_STREAM_ID': b'\x00',
|
|
'RX_TGID': b'\x00\x00\x00',
|
|
'TX_TGID': b'\x00\x00\x00',
|
|
'RX_TIME': time(),
|
|
'TX_TIME': time(),
|
|
'RX_TYPE': HBPF_SLT_VTERM,
|
|
'TX_TYPE': HBPF_SLT_VTERM,
|
|
'RX_LC': b'\x00',
|
|
'TX_H_LC': b'\x00',
|
|
'TX_T_LC': b'\x00',
|
|
'TX_EMB_LC': {
|
|
1: b'\x00',
|
|
2: b'\x00',
|
|
3: b'\x00',
|
|
4: b'\x00',
|
|
},
|
|
'lastSeq': False,
|
|
'lastData': False
|
|
}
|
|
}
|
|
|
|
def to_target(self, _peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data, pkt_time, dmrpkt, _bits,_bridge,_system,_noOBP,sysIgnore):
|
|
_sysIgnore = sysIgnore
|
|
for _target in BRIDGES[_bridge]:
|
|
#if _target['SYSTEM'] != self._system or (_target['SYSTEM'] == self._system and _target['TS'] != _slot):
|
|
if _target['SYSTEM'] != self._system and _target['ACTIVE']:
|
|
#if _target['ACTIVE']:
|
|
_target_status = systems[_target['SYSTEM']].STATUS
|
|
_target_system = self._CONFIG['SYSTEMS'][_target['SYSTEM']]
|
|
|
|
if (_target['SYSTEM'],_target['TS']) in _sysIgnore:
|
|
#logger.debug("(DEDUP) HBP Source - Skipping system %s TS: %s",_target['SYSTEM'],_target['TS'])
|
|
continue
|
|
if _target_system['MODE'] == 'OPENBRIDGE':
|
|
if _noOBP == True:
|
|
continue
|
|
#We want to ignore this system and TS combination if it's called again for this packet
|
|
_sysIgnore.append((_target['SYSTEM'],_target['TS']))
|
|
|
|
#If target has quenched us, don't send
|
|
if ('_bcsq' in _target_system) and (_dst_id in _target_system['_bcsq']) and (_target_system['_bcsq'][_target['TGID']] == _stream_id):
|
|
#logger.info('(%s) Conference Bridge: %s, is Source Quenched for Stream ID: %s, skipping system: %s TS: %s, TGID: %s', self._system, _bridge, int_id(_stream_id), _target['SYSTEM'], _target['TS'], int_id(_target['TGID']))
|
|
continue
|
|
|
|
#If target has missed 6 (on 1 min) of keepalives, don't send
|
|
if _target_system['ENHANCED_OBP'] and '_bcka' in _target_system and _target_system['_bcka'] < pkt_time - 60:
|
|
continue
|
|
|
|
#If talkgroup is prohibited by ACL
|
|
if self._CONFIG['GLOBAL']['USE_ACL']:
|
|
if not acl_check(_target['TGID'],self._CONFIG['GLOBAL']['TG1_ACL']):
|
|
#logger.info('(%s) TGID prohibited by ACL, not sending', _target['SYSTEM'])
|
|
continue
|
|
|
|
if _target_system['USE_ACL']:
|
|
if not acl_check(_target['TGID'],_target_system['TG1_ACL']):
|
|
#logger.info('(%s) TGID prohibited by ACL, not sending', _target['SYSTEM'])
|
|
continue
|
|
|
|
|
|
# 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,
|
|
'RX_PEER': _peer_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]
|
|
try:
|
|
dmrpkt = dmrbits.tobytes()
|
|
except AttributeError:
|
|
logger.exception('(%s) Non-fatal AttributeError - dmrbits.tobytes()',self._system)
|
|
|
|
_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']))
|
|
return _sysIgnore
|
|
|
|
def dmrd_received(self, _peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data):
|
|
pkt_time = time()
|
|
dmrpkt = _data[20:53]
|
|
_bits = _data[15]
|
|
|
|
_nine = bytes_3(9)
|
|
|
|
_lang = CONFIG['SYSTEMS'][self._system]['ANNOUNCEMENT_LANGUAGE']
|
|
|
|
_int_dst_id = int_id(_dst_id)
|
|
|
|
#Handle private calls (for reflectors)
|
|
if _call_type == 'unit' and _slot == 2:
|
|
if (_stream_id != self.STATUS[_slot]['RX_STREAM_ID']):
|
|
|
|
self.STATUS[_slot]['_stopTgAnnounce'] = False
|
|
|
|
logger.warning('(%s) Reflector: Private call from %s to %s',self._system, int_id(_rf_src), _int_dst_id)
|
|
#if _int_dst_id >= 4000 and _int_dst_id <= 5000:
|
|
if _int_dst_id >= 5 and _int_dst_id != 9 and _int_dst_id <= 999999:
|
|
_bridgename = '#'+ str(_int_dst_id)
|
|
if _bridgename not in BRIDGES and not (_int_dst_id >= 4000 and _int_dst_id <= 5000) and not (_int_dst_id >=9991 and _int_dst_id <= 9999):
|
|
logger.info('(%s) [A] Reflector for TG %s does not exist. Creating as User Activated. Timeout: %s',self._system, _int_dst_id,CONFIG['SYSTEMS'][self._system]['DEFAULT_UA_TIMER'])
|
|
make_single_reflector(_dst_id,CONFIG['SYSTEMS'][self._system]['DEFAULT_UA_TIMER'],self._system)
|
|
|
|
if _int_dst_id > 5 and _int_dst_id != 9 and _int_dst_id != 5000 and not (_int_dst_id >=9991 and _int_dst_id <= 9999):
|
|
for _bridge in BRIDGES:
|
|
if _bridge[0:1] != '#':
|
|
continue
|
|
for _system in BRIDGES[_bridge]:
|
|
_dehash_bridge = _bridge[1:]
|
|
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)):
|
|
_system['TIMER'] = pkt_time + _system['TIMEOUT']
|
|
logger.info('(%s) [B] Transmission match for Reflector: %s. Reset timeout to %s', self._system, _bridge, _system['TIMER'])
|
|
|
|
# TGID matches an ACTIVATION trigger
|
|
if _int_dst_id == int(_dehash_bridge) and _system['SYSTEM'] == self._system and _slot == _system['TS']:
|
|
# Set the matching rule as ACTIVE
|
|
if _system['ACTIVE'] == False:
|
|
_system['ACTIVE'] = True
|
|
_system['TIMER'] = pkt_time + _system['TIMEOUT']
|
|
logger.info('(%s) [C] Reflector: %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) [D] Reflector: %s has an "OFF" timer and set to "ON": 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) [E] Reflector: %s, timeout timer reset to: %s', self._system, _bridge, _system['TIMER'] - pkt_time)
|
|
|
|
# TGID matches an DE-ACTIVATION trigger
|
|
#Single TG mode
|
|
if (_dst_id in _system['OFF'] or _dst_id in _system['RESET'] or (_int_dst_id != int(_dehash_bridge)) and _system['SYSTEM'] == self._system and _slot == _system['TS']):
|
|
# Set the matching rule as ACTIVE
|
|
#Single TG mode
|
|
if _dst_id in _system['OFF'] or _int_dst_id != int(_dehash_bridge) :
|
|
#if _dst_id in _system['OFF']:
|
|
if _system['ACTIVE'] == True:
|
|
_system['ACTIVE'] = False
|
|
logger.info('(%s) [F] Reflector: %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) [G] Reflector: %s has ON timer and set to "OFF": 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) [H] Reflector: %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_id in _system['OFF']:
|
|
_system['TIMER'] = pkt_time
|
|
logger.info('(%s) [I] Reflector: %s has ON timer and set to "OFF": timeout timer cancelled', self._system, _bridge)
|
|
|
|
|
|
if (_frame_type == HBPF_DATA_SYNC) and (_dtype_vseq == HBPF_SLT_VTERM) and (self.STATUS[_slot]['RX_TYPE'] != HBPF_SLT_VTERM):
|
|
|
|
#Speak callsign before message
|
|
_say = [words[_lang]['silence']]
|
|
# _systemcs = re.sub(r'\W+', '', self._system)
|
|
# _systemcs.upper()
|
|
# for character in _systemcs:
|
|
# _say.append(words[character])
|
|
# _say.append(words['silence'])
|
|
|
|
if _int_dst_id <= 5 or _int_dst_id == 9:
|
|
logger.info('(%s) Reflector: voice called - TG < 5 or 9 - "busy""', self._system)
|
|
_say.append(words[_lang]['busy'])
|
|
_say.append(words[_lang]['silence'])
|
|
|
|
#If disconnection called
|
|
if _int_dst_id == 4000:
|
|
logger.info('(%s) Reflector: voice called - 4000 "not linked"', self._system)
|
|
_say.append(words[_lang]['notlinked'])
|
|
_say.append(words[_lang]['silence'])
|
|
|
|
#If status called
|
|
elif _int_dst_id == 5000:
|
|
_active = False
|
|
for _bridge in BRIDGES:
|
|
if _bridge[0:1] != '#':
|
|
continue
|
|
for _system in BRIDGES[_bridge]:
|
|
_dehash_bridge = _bridge[1:]
|
|
if _system['SYSTEM'] == self._system and _slot == _system['TS']:
|
|
if _system['ACTIVE'] == True:
|
|
logger.info('(%s) Reflector: voice called - 5000 status - "linked to %s"', self._system,_dehash_bridge)
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['linkedto'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['to'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['silence'])
|
|
|
|
for num in str(_dehash_bridge):
|
|
_say.append(words[_lang][num])
|
|
|
|
_active = True
|
|
break
|
|
|
|
if _active == False:
|
|
logger.info('(%s) Reflector: voice called - 5000 status - "not linked"', self._system)
|
|
_say.append(words[_lang]['notlinked'])
|
|
|
|
#Information services
|
|
elif _int_dst_id >= 9991 and _int_dst_id <= 9999:
|
|
self.STATUS[_slot]['_stopTgAnnounce'] = True
|
|
reactor.callInThread(playFileOnRequest,self,_int_dst_id)
|
|
#playFileOnRequest(self,_int_dst_id)
|
|
|
|
|
|
#Speak what TG was requested to link
|
|
elif not self.STATUS[_slot]['_stopTgAnnounce']:
|
|
logger.info('(%s) Reflector: voice called (linking) "linked to %s"', self._system,_int_dst_id)
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['linkedto'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['to'])
|
|
_say.append(words[_lang]['silence'])
|
|
_say.append(words[_lang]['silence'])
|
|
|
|
for num in str(_int_dst_id):
|
|
_say.append(words[_lang][num])
|
|
|
|
if not self.STATUS[_slot]['_stopTgAnnounce']:
|
|
speech = pkt_gen(bytes_3(5000), _nine, bytes_4(9), 1, _say)
|
|
#call speech in a thread as it contains sleep() and hence could block the reactor
|
|
reactor.callInThread(sendSpeech,self,speech)
|
|
|
|
# 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
|
|
|
|
|
|
#Handle group calls
|
|
if _call_type == 'group' or _call_type == 'vcsbk':
|
|
|
|
# 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 <FROM> SUB: %s PEER: %s <TO> 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) *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 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
|
|
|
|
#Create default bridge for unknown TG
|
|
if int_id(_dst_id) >= 5 and int_id(_dst_id) != 9 and int_id(_dst_id) != 4000 and int_id(_dst_id) != 5000 and (str(int_id(_dst_id)) not in BRIDGES):
|
|
logger.info('(%s) Bridge for TG %s does not exist. Creating as User Activated. Timeout %s',self._system, int_id(_dst_id),CONFIG['SYSTEMS'][self._system]['DEFAULT_UA_TIMER'])
|
|
make_single_bridge(_dst_id,self._system,_slot,CONFIG['SYSTEMS'][self._system]['DEFAULT_UA_TIMER'])
|
|
|
|
#LoopControl#
|
|
for system in systems:
|
|
if system == self._system:
|
|
continue
|
|
if CONFIG['SYSTEMS'][system]['MODE'] != 'OPENBRIDGE':
|
|
for _sysslot in systems[system].STATUS:
|
|
if 'RX_STREAM_ID' in systems[system].STATUS[_sysslot] and _stream_id == systems[system].STATUS[_sysslot]['RX_STREAM_ID']:
|
|
if 'LOOPLOG' not in self.STATUS[_slot] or not self.STATUS[_slot]['LOOPLOG']:
|
|
logger.info("(%s) OBP *LoopControl* FIRST HBP: %s, STREAM ID: %s, TG: %s, TS: %s, IGNORE THIS SOURCE",self._system, system, int_id(_stream_id), int_id(_dst_id),_sysslot)
|
|
self.STATUS[_slot]['LOOPLOG'] = True
|
|
self.STATUS[_slot]['LAST'] = pkt_time
|
|
return
|
|
else:
|
|
#if _stream_id in systems[system].STATUS and systems[system].STATUS[_stream_id]['START'] <= self.STATUS[_stream_id]['START']:
|
|
if _stream_id in systems[system].STATUS and '1ST' in systems[system].STATUS[_stream_id] and systems[system].STATUS[_stream_id]['TGID'] == _dst_id:
|
|
if 'LOOPLOG' not in self.STATUS[_slot] or not self.STATUS[_slot]['LOOPLOG']:
|
|
logger.info("(%s) OBP *LoopControl* FIRST OBP %s, STREAM ID: %s, TG %s, IGNORE THIS SOURCE",self._system, system, int_id(_stream_id), int_id(_dst_id))
|
|
self.STATUS[_slot]['LOOPLOG'] = True
|
|
self.STATUS[_slot]['LAST'] = pkt_time
|
|
|
|
if CONFIG['SYSTEMS'][self._system]['ENHANCED_OBP'] and '_bcsq' not in self.STATUS[_slot]:
|
|
systems[self._system].send_bcsq(_dst_id,_stream_id)
|
|
#logger.warning("(%s) OBP *BridgeControl* Sent BCSQ , STREAM ID: %s, TG %s",self._system, int_id(_stream_id), int_id(_dst_id))
|
|
self.STATUS[_slot]['_bcsq'] = True
|
|
return
|
|
|
|
#Duplicate handling#
|
|
#Duplicate complete packet
|
|
if self.STATUS[_slot]['lastData'] and self.STATUS[_slot]['lastData'] == _data and _seq > 1:
|
|
logger.warning("(%s) *PacketControl* last packet is a complete duplicate of the previous one, disgarding. Stream ID:, %s TGID: %s",self._system,int_id(_stream_id),int_id(_dst_id))
|
|
return
|
|
#Handle inbound duplicates
|
|
if _seq and _seq == self.STATUS[_slot]['lastSeq']:
|
|
logger.warning("(%s) *PacketControl* Duplicate sequence number %s, disgarding. Stream ID:, %s TGID: %s",self._system,_seq,int_id(_stream_id),int_id(_dst_id))
|
|
return
|
|
#Inbound out-of-order packets
|
|
if _seq and self.STATUS[_slot]['lastSeq'] and (_seq != 1) and (_seq < self.STATUS[_slot]['lastSeq']):
|
|
logger.warning("%s) *PacketControl* Out of order packet - last SEQ: %s, this SEQ: %s, disgarding. Stream ID:, %s TGID: %s ",self._system,self.STATUS[_slot]['lastSeq'],_seq,int_id(_stream_id),int_id(_dst_id))
|
|
return
|
|
#Inbound missed packets
|
|
if _seq and self.STATUS[_slot]['lastSeq'] and _seq > (self.STATUS[_slot]['lastSeq']+1):
|
|
logger.warning("(%s) *PacketControl* Missed packet(s) - last SEQ: %s, this SEQ: %s. Stream ID:, %s TGID: %s ",self._system,self.STATUS[_slot]['lastSeq'],_seq,int_id(_stream_id),int_id(_dst_id))
|
|
|
|
#Save this sequence number
|
|
self.STATUS[_slot]['lastSeq'] = _seq
|
|
#Save this packet
|
|
self.STATUS[_slot]['lastData'] = _data
|
|
|
|
_sysIgnore = []
|
|
for _bridge in BRIDGES:
|
|
#if _bridge[0:1] != '#':
|
|
if True:
|
|
for _system in BRIDGES[_bridge]:
|
|
if _system['SYSTEM'] == self._system and _system['TGID'] == _dst_id and _system['TS'] == _slot and _system['ACTIVE'] == True:
|
|
_sysIgnore = self.to_target(_peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data, pkt_time, dmrpkt, _bits,_bridge,_system,False,_sysIgnore)
|
|
|
|
#Send to reflector or TG too, if it exists
|
|
if _bridge[0:1] == '#':
|
|
_bridge = _bridge[1:]
|
|
else:
|
|
_bridge = '#'+_bridge
|
|
if _bridge in BRIDGES:
|
|
_sysIgnore = self.to_target(_peer_id, _rf_src, _dst_id, _seq, _slot, _call_type, _frame_type, _dtype_vseq, _stream_id, _data, pkt_time, dmrpkt, _bits,_bridge,_system,False,_sysIgnore)
|
|
|
|
# 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'))
|
|
|
|
#Reset back to False
|
|
self.STATUS[_slot]['lastSeq'] = False
|
|
self.STATUS[_slot]['lastData'] = False
|
|
|
|
#
|
|
# Begin in-band signalling for call end. This has nothign to do with routing traffic directly.
|
|
#
|
|
|
|
# Iterate the rules dictionary
|
|
for _bridge in BRIDGES:
|
|
if (_bridge[0:1] == '#') and (_int_dst_id != 9):
|
|
continue
|
|
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)):
|
|
_system['TIMER'] = pkt_time + _system['TIMEOUT']
|
|
logger.info('(%s) [1] Transmission match for Bridge: %s. Reset timeout to %s', self._system, _bridge, _system['TIMER'])
|
|
|
|
# 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) [2] 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) [3] 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) [4] Bridge: %s, timeout timer reset to: %s', self._system, _bridge, _system['TIMER'] - pkt_time)
|
|
|
|
# TGID matches an DE-ACTIVATION trigger
|
|
#Single TG mode
|
|
if (CONFIG['SYSTEMS'][self._system]['MODE'] == 'MASTER' and CONFIG['SYSTEMS'][self._system]['SINGLE_MODE']) == True:
|
|
if (_dst_id in _system['OFF'] or _dst_id in _system['RESET'] or _dst_id != _system['TGID']) and _slot == _system['TS']:
|
|
#if (_dst_id in _system['OFF'] or _dst_id in _system['RESET']) and _slot == _system['TS']:
|
|
# Set the matching rule as ACTIVE
|
|
#Single TG mode
|
|
if _dst_id in _system['OFF'] or _dst_id != _system['TGID']:
|
|
#if _dst_id in _system['OFF']:
|
|
if _system['ACTIVE'] == True:
|
|
_system['ACTIVE'] = False
|
|
logger.info('(%s) [5] 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) [6] Bridge: %s set to ON with an "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) [7] 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_id in _system['OFF']:
|
|
_system['TIMER'] = pkt_time
|
|
logger.info('(%s) [8] Bridge: %s set to ON with and "OFF" timer rule: timeout timer cancelled', self._system, _bridge)
|
|
else:
|
|
|
|
if (_dst_id in _system['OFF'] or _dst_id in _system['RESET']) and _slot == _system['TS']:
|
|
#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 _dst_id in _system['OFF']:
|
|
if _system['ACTIVE'] == True:
|
|
_system['ACTIVE'] = False
|
|
logger.info('(%s) [9] 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) [10] 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) [11] 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_id in _system['OFF']:
|
|
_system['TIMER'] = pkt_time
|
|
logger.info('(%s) [12] Bridge: %s set to ON with and "OFF" timer rule: timeout timer cancelled', self._system, _bridge)
|
|
|
|
#
|
|
# 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
|
|
|
|
#
|
|
# Socket-based reporting section
|
|
#
|
|
class bridgeReportFactory(reportFactory):
|
|
|
|
def send_bridge(self):
|
|
serialized = pickle.dumps(BRIDGES, protocol=2) #.decode("utf-8", errors='ignore')
|
|
self.send_clients(REPORT_OPCODES['BRIDGE_SND']+serialized)
|
|
|
|
def send_bridgeEvent(self, _data):
|
|
if isinstance(_data, str):
|
|
_data = _data.decode('utf-8', error='ignore')
|
|
self.send_clients(REPORT_OPCODES['BRDG_EVENT']+_data)
|
|
|
|
|
|
#************************************************
|
|
# MAIN PROGRAM LOOP STARTS HERE
|
|
#************************************************
|
|
|
|
if __name__ == '__main__':
|
|
|
|
import argparse
|
|
import sys
|
|
import os
|
|
import signal
|
|
|
|
# Higheset peer ID permitted by HBP
|
|
PEER_MAX = 4294967295
|
|
|
|
ID_MAX = 16776415
|
|
|
|
#Set process title early
|
|
setproctitle(__file__)
|
|
|
|
# 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('-r', '--rules', action='store', dest='RULES_FILE', help='/full/path/to/rules.file (usually rules.py)')
|
|
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)
|
|
|
|
# Ensure we have a path for the rules file, if one wasn't specified, then use the default (top of file)
|
|
if not cli_args.RULES_FILE:
|
|
cli_args.RULES_FILE = os.path.dirname(os.path.abspath(__file__))+'/rules.py'
|
|
|
|
# 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) 2020, 2021 Simon G7RZU simon@gb7fr.org.uk')
|
|
logger.info('Copyright (c) 2013, 2014, 2015, 2016, 2018, 2019\n\tThe Regents of the K0USY Group. All rights reserved.\n')
|
|
logger.debug('(GLOBAL) Logging system started, anything from here on gets logged')
|
|
|
|
|
|
#If MySQL is enabled, read master config from MySQL too
|
|
if CONFIG['MYSQL']['USE_MYSQL'] == True:
|
|
logger.debug('(MYSQL) MySQL config enabled')
|
|
SQLCONFIG = {}
|
|
sql = useMYSQL(CONFIG['MYSQL']['SERVER'], CONFIG['MYSQL']['USER'], CONFIG['MYSQL']['PASS'], CONFIG['MYSQL']['DB'],CONFIG['MYSQL']['TABLE'],logger)
|
|
#Run it once immediately
|
|
if sql.con():
|
|
logger.debug('(MYSQL) reading config from database')
|
|
try:
|
|
SQLCONFIG = sql.getConfig()
|
|
#Add MySQL config data to config dict
|
|
except:
|
|
logger.debug('(MYSQL) problem with SQL query, aborting')
|
|
sql.close()
|
|
logger.debug('(MYSQL) building ACLs')
|
|
# Build ACLs
|
|
for system in SQLCONFIG:
|
|
SQLCONFIG[system]['REG_ACL'] = acl_build(SQLCONFIG[system]['REG_ACL'], PEER_MAX)
|
|
for acl in ['SUB_ACL', 'TG1_ACL', 'TG2_ACL']:
|
|
SQLCONFIG[system][acl] = acl_build(SQLCONFIG[system][acl], ID_MAX)
|
|
|
|
CONFIG['SYSTEMS'].update(SQLCONFIG)
|
|
else:
|
|
logger.debug('(MYSQL) problem connecting to SQL server, aborting')
|
|
|
|
|
|
# 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)
|
|
|
|
# Import the ruiles file as a module, and create BRIDGES from it
|
|
spec = importlib.util.spec_from_file_location("module.name", cli_args.RULES_FILE)
|
|
rules_module = importlib.util.module_from_spec(spec)
|
|
try:
|
|
spec.loader.exec_module(rules_module)
|
|
logger.info('(ROUTER) Routing bridges file found and bridges imported: %s', cli_args.RULES_FILE)
|
|
except (ImportError, FileNotFoundError):
|
|
sys.exit('(ROUTER) TERMINATING: Routing bridges file not found or invalid: {}'.format(cli_args.RULES_FILE))
|
|
|
|
# Build the routing rules file
|
|
BRIDGES = make_bridges(rules_module.BRIDGES)
|
|
|
|
#Generator
|
|
generator = {}
|
|
systemdelete = []
|
|
for system in CONFIG['SYSTEMS']:
|
|
if CONFIG['SYSTEMS'][system]['ENABLED']:
|
|
if CONFIG['SYSTEMS'][system]['MODE'] == 'MASTER' and (CONFIG['SYSTEMS'][system]['GENERATOR'] > 1):
|
|
for count in range(CONFIG['SYSTEMS'][system]['GENERATOR']):
|
|
_systemname = system+'-'+str(count)
|
|
generator[_systemname] = copy.deepcopy(CONFIG['SYSTEMS'][system])
|
|
generator[_systemname]['PORT'] = generator[_systemname]['PORT'] + count
|
|
generator[_systemname]['_default_options'] = "TS1_STATIC={};TS2_STATIC={};SINGLE={};DEFAULT_UA_TIMER={};DEFAULT_REFLECTOR={};VOICE={};LANG={}".format(generator[_systemname]['TS1_STATIC'],generator[_systemname]['TS2_STATIC'],int(generator[_systemname]['SINGLE_MODE']),generator[_systemname]['DEFAULT_UA_TIMER'],generator[_systemname]['DEFAULT_REFLECTOR'],int(generator[_systemname]['VOICE_IDENT']), generator[_systemname]['ANNOUNCEMENT_LANGUAGE'])
|
|
logger.debug('(GLOBAL) Generator - generated system %s',_systemname)
|
|
generator[_systemname]['_default_options']
|
|
systemdelete.append(system)
|
|
|
|
for _system in generator:
|
|
CONFIG['SYSTEMS'][_system] = generator[_system]
|
|
for _system in systemdelete:
|
|
CONFIG['SYSTEMS'].pop(_system)
|
|
|
|
del generator
|
|
del systemdelete
|
|
|
|
# Default reflector
|
|
logger.debug('(ROUTER) Setting default reflectors')
|
|
for system in CONFIG['SYSTEMS']:
|
|
if CONFIG['SYSTEMS'][system]['MODE'] != 'MASTER':
|
|
continue
|
|
if CONFIG['SYSTEMS'][system]['DEFAULT_REFLECTOR'] > 0:
|
|
make_default_reflector(CONFIG['SYSTEMS'][system]['DEFAULT_REFLECTOR'],CONFIG['SYSTEMS'][system]['DEFAULT_UA_TIMER'],system)
|
|
|
|
#static TGs
|
|
logger.debug('(ROUTER) setting static TGs')
|
|
for system in CONFIG['SYSTEMS']:
|
|
if CONFIG['SYSTEMS'][system]['MODE'] != 'MASTER':
|
|
continue
|
|
_tmout = CONFIG['SYSTEMS'][system]['DEFAULT_UA_TIMER']
|
|
ts1 = []
|
|
ts2 = []
|
|
if CONFIG['SYSTEMS'][system]['TS1_STATIC']:
|
|
ts1 = CONFIG['SYSTEMS'][system]['TS1_STATIC'].split(',')
|
|
if CONFIG['SYSTEMS'][system]['TS2_STATIC']:
|
|
ts2 = CONFIG['SYSTEMS'][system]['TS2_STATIC'].split(',')
|
|
|
|
#if CONFIG['SYSTEMS'][system]['SINGLE_MODE'] == True:
|
|
#if ts1:
|
|
#make_static_tg(int(ts1[0]),1,system)
|
|
#if ts2:
|
|
#make_static_tg(int(ts2[0]),2,system)
|
|
#else:
|
|
for tg in ts1:
|
|
if not tg:
|
|
continue
|
|
tg = int(tg)
|
|
make_static_tg(tg,1,_tmout,system)
|
|
for tg in ts2:
|
|
if not tg:
|
|
continue
|
|
tg = int(tg)
|
|
make_static_tg(tg,2,_tmout,system)
|
|
|
|
# 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')
|
|
|
|
#Read AMBE
|
|
AMBEobj = readAMBE(CONFIG['GLOBAL']['ANNOUNCEMENT_LANGUAGES'],'./Audio/')
|
|
|
|
#global words
|
|
words = AMBEobj.readfiles()
|
|
|
|
for lang in words.keys():
|
|
logger.info('(AMBE) for language %s, read %s words into voice dict',lang,len(words[lang]) - 1)
|
|
|
|
#Remap words for internationalisation
|
|
if lang in voiceMap:
|
|
logger.info('(AMBE) i8n voice map entry for language %s',lang)
|
|
_map = voiceMap[lang]
|
|
for _mapword in _map:
|
|
logger.info('(AMBE) Mapping \"%s\" to \"%s\"',_mapword,_map[_mapword])
|
|
words[lang][_mapword] = words[lang][_map[_mapword]]
|
|
|
|
# HBlink instance creation
|
|
logger.info('(GLOBAL) FreeDMR \'bridge_master.py\' -- SYSTEM STARTING...')
|
|
|
|
|
|
listeningPorts = {}
|
|
|
|
|
|
for system in CONFIG['SYSTEMS']:
|
|
if CONFIG['SYSTEMS'][system]['ENABLED']:
|
|
# if CONFIG['SYSTEMS'][system]['MODE'] == 'XLXPEER':
|
|
# logger.warning('(GLOBAL) system %s not started - XLXPEER connections currently unsupported ', system)
|
|
# continue
|
|
if CONFIG['SYSTEMS'][system]['MODE'] == 'OPENBRIDGE':
|
|
systems[system] = routerOBP(system, CONFIG, report_server)
|
|
else:
|
|
if CONFIG['SYSTEMS'][system]['MODE'] == 'MASTER' and CONFIG['SYSTEMS'][system]['ANNOUNCEMENT_LANGUAGE'] not in CONFIG['GLOBAL']['ANNOUNCEMENT_LANGUAGES'].split(','):
|
|
logger.warning('(GLOBAL) Invalid language in ANNOUNCEMENT_LANGUAGE, skipping system %s',system)
|
|
continue
|
|
systems[system] = routerHBP(system, CONFIG, report_server)
|
|
listeningPorts[system] = 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()
|
|
|
|
# Initialize the rule timer -- this if for user activated stuff
|
|
rule_timer_task = task.LoopingCall(rule_timer_loop)
|
|
rule_timer = rule_timer_task.start(60)
|
|
rule_timer.addErrback(loopingErrHandle)
|
|
|
|
# Initialize the stream trimmer
|
|
stream_trimmer_task = task.LoopingCall(stream_trimmer_loop)
|
|
stream_trimmer = stream_trimmer_task.start(5)
|
|
stream_trimmer.addErrback(loopingErrHandle)
|
|
|
|
# Ident
|
|
#This runs in a thread so as not to block the reactor
|
|
ident_task = task.LoopingCall(threadIdent)
|
|
identa = ident_task.start(900)
|
|
identa.addErrback(loopingErrHandle)
|
|
|
|
#Options parsing
|
|
options_task = task.LoopingCall(options_config)
|
|
options = options_task.start(30)
|
|
options.addErrback(loopingErrHandle)
|
|
|
|
#Mysql config checker
|
|
#This runs in a thread so as not to block the reactor
|
|
if CONFIG['MYSQL']['USE_MYSQL'] == True:
|
|
mysql_task = task.LoopingCall(threadedMysql)
|
|
mysql = mysql_task.start(30)
|
|
mysql.addErrback(loopingErrHandle)
|
|
|
|
#STAT trimmer - once every hour
|
|
if CONFIG['GLOBAL']['GEN_STAT_BRIDGES']:
|
|
stat_trimmer_task = task.LoopingCall(statTrimmer)
|
|
stat_trimmer = stat_trimmer_task.start(3600)#3600
|
|
stat_trimmer.addErrback(loopingErrHandle)
|
|
|
|
#KA Reporting
|
|
ka_task = task.LoopingCall(kaReporting)
|
|
ka = ka_task.start(60)
|
|
ka.addErrback(loopingErrHandle)
|
|
|
|
#more threads
|
|
reactor.suggestThreadPoolSize(100)
|
|
|
|
reactor.run()
|