ShaYmez/HBLink3
1
0
Fork 0
mirror of https://github.com/ShaYmez/hblink3.git synced 2026-01-20 12:35:32 -05:00
Code Issues Projects Releases Wiki Activity

Fix typos, remove commented code, and simplify boolean comparisons

Browse Source 
Co-authored-by: ShaYmez <76499782+ShaYmez@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot] 2025-12-13 00:58:15 +00:00
parent 1481b09b79
commit db045ad582
7 changed files with 33 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -5,7 +5,7 @@ Please join the DVSwitch group at groups.io for online forum support, discussion
DVSwitch@groups.io
A voluntary registrty for HBlink systems with public access has been created at http://hblink-register.com.es Please consider listing your system if you allow open access.
A voluntary registry for HBlink systems with public access has been created at http://hblink-register.com.es Please consider listing your system if you allow open access.
---

43
bridge.py
View File

@ -25,7 +25,7 @@ reflector. You define conference bridges and any system joined to that conferenc
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.
you create in the configuration file) to pass traffic.
This program currently only works with group voice calls.
'''
@ -51,7 +51,6 @@ from const import *
# Stuff for socket reporting
import pickle
# REMOVE LATER from datetime import datetime
# The module needs logging, but handlers, etc. are controlled by the parent
import logging
logger = logging.getLogger(__name__)
@ -65,7 +64,7 @@ __license__ = 'GNU GPLv3'
__maintainer__ = 'Cort Buffington, N0MJS'
__email__ = 'n0mjs@me.com'
# Module gobal varaibles
# Module global variables
# Dictionary for dynamically mapping unit (subscriber) to a system.
# This is for pruning unit-to-uint calls to not broadcast once the
@ -114,7 +113,7 @@ def make_bridges(_rules):
for i, e in enumerate(_system['OFF']):
_system['OFF'][i] = bytes_3(_system['OFF'][i])
_system['TIMEOUT'] = _system['TIMEOUT']*60
if _system['ACTIVE'] == True:
if _system['ACTIVE']:
_system['TIMER'] = time() + _system['TIMEOUT']
else:
_system['TIMER'] = time()
@ -130,24 +129,24 @@ def rule_timer_loop():
for _bridge in BRIDGES:
for _system in BRIDGES[_bridge]:
if _system['TO_TYPE'] == 'ON':
if _system['ACTIVE'] == True:
if _system['ACTIVE']:
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']))
else:
timeout_in = _system['TIMER'] - _now
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:
elif not _system['ACTIVE']:
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 not _system['ACTIVE']:
if _system['TIMER'] < _now:
_system['ACTIVE'] = 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
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:
elif _system['ACTIVE']:
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:
logger.debug('(ROUTER) Conference Bridge NO ACTION: System: %s, Bridge: %s, TS: %s, TGID: %s', _system['SYSTEM'], _bridge, _system['TS'], int_id(_system['TGID']))
@ -335,22 +334,22 @@ class routerOBP(OPENBRIDGE):
# 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:
if not self.STATUS[_stream_id]['CONTENTION']:
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:
if not self.STATUS[_stream_id]['CONTENTION']:
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:
if not self.STATUS[_stream_id]['CONTENTION']:
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:
if not self.STATUS[_stream_id]['CONTENTION']:
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
@ -512,23 +511,23 @@ class routerOBP(OPENBRIDGE):
#
'''
if ((_dst_id != _target_status[_slot]['RX_TGID']) and ((pkt_time - _target_status[_slot]['RX_TIME']) < _target_system['GROUP_HANGTIME'])):
if self.STATUS[_stream_id]['CONTENTION'] == False:
if not self.STATUS[_stream_id]['CONTENTION']:
self.STATUS[_stream_id]['CONTENTION'] = True
logger.info('(%s) Call not routed to TGID %s, target active or in group hangtime: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_dst_id), _target, _slot, int_id(_target_status[_slot]['RX_TGID']))
continue
if ((_dst_id != _target_status[_slot]['TX_TGID']) and ((pkt_time - _target_status[_slot]['TX_TIME']) < _target_system['GROUP_HANGTIME'])):
if self.STATUS[_stream_id]['CONTENTION'] == False:
if not self.STATUS[_stream_id]['CONTENTION']:
self.STATUS[_stream_id]['CONTENTION'] = True
logger.info('(%s) Call not routed to TGID%s, target in group hangtime: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_dst_id), _target, _slot, int_id(_target_status[_slot]['TX_TGID']))
continue
'''
if (_dst_id == _target_status[_slot]['RX_TGID']) and ((pkt_time - _target_status[_slot]['RX_TIME']) < STREAM_TO):
if self.STATUS[_stream_id]['CONTENTION'] == False:
if not self.STATUS[_stream_id]['CONTENTION']:
self.STATUS[_stream_id]['CONTENTION'] = True
logger.info('(%s) Call not routed to TGID%s, matching call already active on target: HBSystem: %s, TS: %s, TGID: %s', self._system, int_id(_dst_id), _target, _slot, int_id(_target_status[_slot]['RX_TGID']))
continue
if (_dst_id == _target_status[_slot]['TX_TGID']) and (_rf_src != _target_status[_slot]['TX_RFS']) and ((pkt_time - _target_status[_slot]['TX_TIME']) < STREAM_TO):
if self.STATUS[_stream_id]['CONTENTION'] == False:
if not self.STATUS[_stream_id]['CONTENTION']:
self.STATUS[_stream_id]['CONTENTION'] = True
logger.info('(%s) Call not routed for subscriber %s, call route in progress on target: HBSystem: %s, TS: %s, TGID: %s, SUB: %s', self._system, int_id(_rf_src), _target, _slot, int_id(_target_status[_slot]['TX_TGID']), int_id(_target_status[_slot]['TX_RFS']))
continue
@ -847,7 +846,7 @@ class routerHBP(HBSYSTEM):
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)):
if _slot == _system['TS'] and _dst_id == _system['TGID'] and ((_system['TO_TYPE'] == 'ON' and _system['ACTIVE']) or (_system['TO_TYPE'] == 'OFF' and not _system['ACTIVE'])):
_system['TIMER'] = pkt_time + _system['TIMEOUT']
logger.info('(%s) Transmission match for Bridge: %s. Reset timeout to %s', self._system, _bridge, _system['TIMER'])
@ -855,7 +854,7 @@ class routerHBP(HBSYSTEM):
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:
if not _system['ACTIVE']:
_system['ACTIVE'] = True
_system['TIMER'] = pkt_time + _system['TIMEOUT']
logger.info('(%s) Bridge: %s, connection changed to state: %s', self._system, _bridge, _system['ACTIVE'])
@ -864,7 +863,7 @@ class routerHBP(HBSYSTEM):
_system['TIMER'] = pkt_time
logger.info('(%s) Bridge: %s set to "OFF" with an on timer rule: timeout timer cancelled', self._system, _bridge)
# Reset the timer for the rule
if _system['ACTIVE'] == True and _system['TO_TYPE'] == 'ON':
if _system['ACTIVE'] and _system['TO_TYPE'] == 'ON':
_system['TIMER'] = pkt_time + _system['TIMEOUT']
logger.info('(%s) Bridge: %s, timeout timer reset to: %s', self._system, _bridge, _system['TIMER'] - pkt_time)
@ -872,7 +871,7 @@ class routerHBP(HBSYSTEM):
if (_dst_id in _system['OFF'] or _dst_id in _system['RESET']) and _slot == _system['TS']:
# Set the matching rule as ACTIVE
if _dst_id in _system['OFF']:
if _system['ACTIVE'] == True:
if _system['ACTIVE']:
_system['ACTIVE'] = False
logger.info('(%s) Bridge: %s, connection changed to state: %s', self._system, _bridge, _system['ACTIVE'])
# Cancel the timer if we've enabled an "ON" type timeout
@ -880,11 +879,11 @@ class routerHBP(HBSYSTEM):
_system['TIMER'] = pkt_time
logger.info('(%s) Bridge: %s set to ON with and "OFF" timer rule: timeout timer cancelled', self._system, _bridge)
# Reset the timer for the rule
if _system['ACTIVE'] == False and _system['TO_TYPE'] == 'OFF':
if not _system['ACTIVE'] and _system['TO_TYPE'] == 'OFF':
_system['TIMER'] = pkt_time + _system['TIMEOUT']
logger.info('(%s) Bridge: %s, timeout timer reset to: %s', self._system, _bridge, _system['TIMER'] - pkt_time)
# Cancel the timer if we've enabled an "ON" type timeout
if _system['ACTIVE'] == True and _system['TO_TYPE'] == 'ON' and _dst_group in _system['OFF']:
if _system['ACTIVE'] and _system['TO_TYPE'] == 'ON' and _dst_group in _system['OFF']:
_system['TIMER'] = pkt_time
logger.info('(%s) Bridge: %s set to ON with and "OFF" timer rule: timeout timer cancelled', self._system, _bridge)

2
bridge_all.py
View File

@ -67,7 +67,7 @@ __maintainer__ = 'Cort Buffington, N0MJS'
__email__ = 'n0mjs@me.com'
__status__ = 'pre-alpha'
# Module gobal varaibles
# Module global variables
class bridgeallSYSTEM(HBSYSTEM):

6
config.py
View File

@ -20,9 +20,9 @@
'''
This module generates the configuration data structure for hblink.py and
assoicated programs that use it. It has been seaparated into a different
module so as to keep hblink.py easeier to navigate. This file only needs
updated if the items in the main configuraiton file (usually hblink.cfg)
associated programs that use it. It has been separated into a different
module so as to keep hblink.py easier to navigate. This file only needs
updated if the items in the main configuration file (usually hblink.cfg)
change.
'''

8
hblink.py
View File

@ -27,7 +27,7 @@ works stand-alone before troubleshooting any applications that use it. It has
sufficient logging to be used standalone as a troubleshooting application.
'''
# Specifig functions from modules we need
# Specific functions from modules we need
from binascii import b2a_hex as ahex
from binascii import a2b_hex as bhex
from random import randint
@ -388,7 +388,7 @@ class HBSYSTEM(DatagramProtocol):
return
# The basic purpose of a master is to repeat to the peers
if self._config['REPEAT'] == True:
if self._config['REPEAT']:
pkt = [_data[:11], '', _data[15:]]
for _peer in self._peers:
if _peer != _peer_id:
@ -466,7 +466,7 @@ class HBSYSTEM(DatagramProtocol):
self.transport.write(b''.join([MSTNAK, _peer_id]), _sockaddr)
logger.warning('(%s) Login challenge from Radio ID that has not logged in: %s', self._system, int_id(_peer_id))
elif _command == RPTC: # Repeater is sending it's configuraiton OR disconnecting
elif _command == RPTC: # Repeater is sending it's configuration OR disconnecting
if _data[:5] == RPTCL: # Disconnect command
_peer_id = _data[5:9]
if _peer_id in self._peers \
@ -752,7 +752,7 @@ class reportFactory(Factory):
# ID ALIAS CREATION
# Download
def mk_aliases(_config):
if _config['ALIASES']['TRY_DOWNLOAD'] == True:
if _config['ALIASES']['TRY_DOWNLOAD']:
# Try updating peer aliases file
result = try_download(_config['ALIASES']['PATH'], _config['ALIASES']['PEER_FILE'], _config['ALIASES']['PEER_URL'], _config['ALIASES']['STALE_TIME'])
logger.info('(GLOBAL) %s', result)

3
play_ambe.py
View File

@ -42,7 +42,6 @@ from voice_lib import words
# 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__)
@ -56,7 +55,7 @@ __license__ = 'GNU GPLv3'
__maintainer__ = 'Cort Buffington, N0MJS'
__email__ = 'n0mjs@me.com'
# Module gobal varaibles
# Module global variables
class OBP(OPENBRIDGE):

4
playback.py
View File

@ -51,7 +51,7 @@ __maintainer__ = 'Cort Buffington, N0MJS'
__email__ = 'n0mjs@me.com'
__status__ = 'pre-alpha'
# Module gobal varaibles
# Module global variables
class playback(HBSYSTEM):
@ -200,7 +200,7 @@ if __name__ == '__main__':
# ID ALIAS CREATION
# Download
if CONFIG['ALIASES']['TRY_DOWNLOAD'] == True:
if CONFIG['ALIASES']['TRY_DOWNLOAD']:
# Try updating peer aliases file
result = try_download(CONFIG['ALIASES']['PATH'], CONFIG['ALIASES']['PEER_FILE'], CONFIG['ALIASES']['PEER_URL'], CONFIG['ALIASES']['STALE_TIME'])
logger.info(result)