mirror of
https://github.com/ShaYmez/hblink3.git
synced 2024-11-25 01:08:40 -05:00
108 lines
4.2 KiB
Python
Executable File
108 lines
4.2 KiB
Python
Executable File
#!/usr/bin/env python
|
|
#
|
|
###############################################################################
|
|
# Copyright (C) 2016-2020 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
|
|
###############################################################################
|
|
|
|
from bitarray import bitarray
|
|
from dmr_utils3 import bptc, golay, qr
|
|
from dmr_utils3.utils import bytes_3, bytes_4
|
|
from dmr_utils3.const import EMB, SLOT_TYPE, BS_VOICE_SYNC, BS_DATA_SYNC, LC_OPT_G
|
|
from random import randint
|
|
from voice_lib import words
|
|
|
|
|
|
# Precalculated "dmrbits" (DMRD packet byte 15) -- just (slot << 7 | this value) and you're good to go!
|
|
HEADBITS = 0b00100001
|
|
BURSTBITS = [0b00010000,0b00000001,0b00000010,0b00000011,0b00000100,0b00000101]
|
|
TERMBITS = 0b00100010
|
|
|
|
# Need a bitstring of 4-bytes of zero for burst F
|
|
NULL_EMB_LC = bitarray(endian='big')
|
|
NULL_EMB_LC.frombytes(b'\x00\x00\x00\x00')
|
|
|
|
# This is where HBP encodes RSSI, it will need to be null
|
|
TAIL = b'\x00\x00'
|
|
|
|
# WARNING this funciton uses yeild to return a generator that will pass the next HBP packet for a phrase
|
|
# each time that it is called. Do NOT try to use it like a normal function.
|
|
def pkt_gen(_rf_src, _dst_id, _peer, _slot, _phrase):
|
|
|
|
# Calculate all of the static components up-front
|
|
STREAM_ID = bytes_4(randint(0x00, 0xFFFFFFFF))
|
|
SDP = _rf_src + _dst_id + _peer
|
|
LC = LC_OPT_G + _dst_id + _rf_src
|
|
|
|
HEAD_LC = bptc.encode_header_lc(LC)
|
|
HEAD_LC = [HEAD_LC[:98], HEAD_LC[-98:]]
|
|
|
|
TERM_LC = bptc.encode_terminator_lc(LC)
|
|
TERM_LC = [TERM_LC[:98], TERM_LC[-98:]]
|
|
|
|
EMB_LC = bptc.encode_emblc(LC)
|
|
|
|
EMBED = []
|
|
EMBED.append( BS_VOICE_SYNC )
|
|
EMBED.append(EMB['BURST_B'][:8] + EMB_LC[1] + EMB['BURST_B'][-8:])
|
|
EMBED.append(EMB['BURST_C'][:8] + EMB_LC[2] + EMB['BURST_C'][-8:])
|
|
EMBED.append(EMB['BURST_D'][:8] + EMB_LC[3] + EMB['BURST_D'][-8:])
|
|
EMBED.append(EMB['BURST_E'][:8] + EMB_LC[4] + EMB['BURST_E'][-8:])
|
|
EMBED.append(EMB['BURST_F'][:8] + NULL_EMB_LC + EMB['BURST_F'][-8:])
|
|
|
|
|
|
|
|
#initialize the HBP calls stream sequence to 0
|
|
SEQ = 0
|
|
|
|
# Send the Call Stream
|
|
|
|
# Send 3 Voice Header Frames
|
|
for i in range(3):
|
|
pkt = b'DMRD' + bytes([SEQ]) + SDP + bytes([_slot << 7 | HEADBITS]) + STREAM_ID + (HEAD_LC[0] + SLOT_TYPE['VOICE_LC_HEAD'][:10] + BS_DATA_SYNC + SLOT_TYPE['VOICE_LC_HEAD'][-10:] + HEAD_LC[1]).tobytes() + TAIL
|
|
SEQ = (SEQ + 1) % 0x100
|
|
yield pkt
|
|
|
|
# Send each burst, six bursts per Superframe rotating through with the proper EMBED value per burst A-F
|
|
for word in _phrase:
|
|
for burst in range(0, len(word)):
|
|
pkt = b'DMRD' + bytes([SEQ]) + SDP + bytes([_slot << 7 | BURSTBITS[burst % 6]]) + STREAM_ID + (word[burst][0] + EMBED[burst % 6] + word[burst][1]).tobytes() + TAIL
|
|
SEQ = (SEQ + 1) % 0x100
|
|
yield pkt
|
|
|
|
# Send 2 Voice Terminator Frames
|
|
for i in range(2):
|
|
pkt = b'DMRD' + bytes([SEQ]) + SDP + bytes([_slot << 7 | TERMBITS]) + STREAM_ID + (TERM_LC[0] + SLOT_TYPE['VOICE_LC_TERM'][:10] + BS_DATA_SYNC + SLOT_TYPE['VOICE_LC_TERM'][-10:] + TERM_LC[1]).tobytes() + TAIL
|
|
SEQ = (SEQ + 1) % 0x100
|
|
yield pkt
|
|
|
|
# Return False to indicate we're done.
|
|
return False
|
|
|
|
|
|
if __name__ == '__main__':
|
|
from time import time
|
|
|
|
speech = pkt_gen(bytes_3(3120101), bytes_3(3120), bytes_4(312000), 0, [words['connected']])
|
|
|
|
|
|
while True:
|
|
try:
|
|
pkt = next(speech)
|
|
except StopIteration:
|
|
break
|
|
print(len(pkt), pkt[4], pkt)
|