mirror of
https://github.com/DJ2LS/FreeDATA
synced 2024-05-14 08:04:33 +00:00
938 lines
36 KiB
Python
938 lines
36 KiB
Python
|
|
import sys
|
|
import threading
|
|
import time
|
|
import lzma
|
|
from random import randrange
|
|
import hmac
|
|
import hashlib
|
|
import helpers
|
|
import modem
|
|
import numpy as np
|
|
from codec2 import FREEDV_MODE
|
|
from modem_frametypes import FRAME_TYPE as FR_TYPE
|
|
import event_manager
|
|
|
|
from deprecated_protocol_arq_session import ARQ
|
|
|
|
class ISS(ARQ):
|
|
def __init__(self, config, event_queue, states):
|
|
super().__init__(config, event_queue, states)
|
|
|
|
self.tx_n_max_retries_per_burst = 40
|
|
self.datachannel_opening_interval = self.duration_sig1_frame + self.channel_busy_timeout + 1 # time between attempts when opening data channel
|
|
self.irs_buffer_position = 0
|
|
# actual n retries of burst
|
|
self.tx_n_retry_of_burst = 0
|
|
self.burst_ack_snr = 0 # SNR from received burst ack frames
|
|
|
|
|
|
|
|
def arq_transmit(self, data_out: bytes, hmac_salt: bytes):
|
|
"""
|
|
Transmit ARQ frame
|
|
|
|
Args:
|
|
data_out:bytes:
|
|
|
|
|
|
"""
|
|
|
|
# set signalling modes we want to listen to
|
|
# we are in an ongoing arq transmission, so we don't need sig0 actually
|
|
modem.demodulator.RECEIVE_SIG0 = False
|
|
modem.demodulator.RECEIVE_SIG1 = True
|
|
|
|
self.tx_n_retry_of_burst = 0 # retries we already sent data
|
|
# Maximum number of retries to send before declaring a frame is lost
|
|
|
|
# save len of data_out to TOTAL_BYTES for our statistics
|
|
self.states.set("arq_total_bytes", len(data_out))
|
|
|
|
self.arq_file_transfer = True
|
|
frame_total_size = len(data_out).to_bytes(4, byteorder="big")
|
|
|
|
# Compress data frame
|
|
data_frame_compressed = lzma.compress(data_out)
|
|
compression_factor = len(data_out) / len(data_frame_compressed)
|
|
self.arq_compression_factor = np.clip(compression_factor, 0, 255)
|
|
compression_factor = bytes([int(self.arq_compression_factor * 10)])
|
|
|
|
self.event_manager.send_custom_event(
|
|
freedata="modem-message",
|
|
arq="transmission",
|
|
status="transmitting",
|
|
uuid=self.transmission_uuid,
|
|
percent=self.states.arq_transmission_percent,
|
|
bytesperminute=self.states.arq_bytes_per_minute,
|
|
compression=self.arq_compression_factor,
|
|
finished=self.states.arq_seconds_until_finish,
|
|
mycallsign=str(self.mycallsign, 'UTF-8'),
|
|
dxcallsign=str(self.dxcallsign, 'UTF-8'),
|
|
irs=helpers.bool_to_string(self.is_IRS)
|
|
)
|
|
|
|
self.log.info(
|
|
"[Modem] | TX | DATACHANNEL",
|
|
Bytes=self.states.arq_total_bytes,
|
|
)
|
|
|
|
data_out = data_frame_compressed
|
|
|
|
# Reset data transfer statistics
|
|
tx_start_of_transmission = time.time()
|
|
self.calculate_transfer_rate_tx(tx_start_of_transmission, 0, len(data_out))
|
|
|
|
# check if hmac signature is available
|
|
if hmac_salt not in ['', False]:
|
|
print(data_out)
|
|
# create hmac digest
|
|
hmac_digest = hmac.new(hmac_salt, data_out, hashlib.sha256).digest()
|
|
# truncate to 32bit
|
|
frame_payload_crc = hmac_digest[:4]
|
|
self.log.debug("[Modem] frame payload HMAC:", crc=frame_payload_crc.hex())
|
|
|
|
else:
|
|
# Append a crc at the beginning and end of file indicators
|
|
frame_payload_crc = helpers.get_crc_32(data_out)
|
|
self.log.debug("[Modem] frame payload CRC:", crc=frame_payload_crc.hex())
|
|
|
|
# Assemble the data frame
|
|
data_out = (
|
|
self.data_frame_bof
|
|
+ frame_payload_crc
|
|
+ frame_total_size
|
|
+ compression_factor
|
|
+ data_out
|
|
+ self.data_frame_eof
|
|
)
|
|
self.log.debug("[Modem] frame raw data:", data=data_out)
|
|
# Initial bufferposition is 0
|
|
bufferposition = 0
|
|
bufferposition_end = 0
|
|
bufferposition_burst_start = 0
|
|
|
|
# Iterate through data_out buffer
|
|
while not self.data_frame_ack_received and self.states.is_arq_state:
|
|
# we have self.tx_n_max_retries_per_burst attempts for sending a burst
|
|
for self.tx_n_retry_of_burst in range(self.tx_n_max_retries_per_burst):
|
|
# Bound speed level to:
|
|
# - minimum of either the speed or the length of mode list - 1
|
|
# - maximum of either the speed or zero
|
|
self.speed_level = min(self.speed_level, len(self.mode_list) - 1)
|
|
self.speed_level = max(self.speed_level, 0)
|
|
|
|
self.states.set("arq_speed_level", self.speed_level)
|
|
data_mode = self.mode_list[self.speed_level]
|
|
|
|
self.log.debug(
|
|
"[Modem] Speed-level:",
|
|
level=self.speed_level,
|
|
retry=self.tx_n_retry_of_burst,
|
|
mode=FREEDV_MODE(data_mode).name,
|
|
)
|
|
|
|
# Payload information
|
|
payload_per_frame = modem.get_bytes_per_frame(data_mode) - 2
|
|
|
|
self.log.info("[Modem] early buffer info",
|
|
bufferposition=bufferposition,
|
|
bufferposition_end=bufferposition_end,
|
|
bufferposition_burst_start=bufferposition_burst_start
|
|
)
|
|
|
|
# check for maximum frames per burst for remaining data
|
|
n_frames_per_burst = 1
|
|
if self.max_n_frames_per_burst > 1:
|
|
while (payload_per_frame * n_frames_per_burst) % len(data_out[bufferposition_burst_start:]) == (
|
|
payload_per_frame * n_frames_per_burst):
|
|
threading.Event().wait(0.01)
|
|
# print((payload_per_frame * n_frames_per_burst) % len(data_out))
|
|
n_frames_per_burst += 1
|
|
if n_frames_per_burst == self.max_n_frames_per_burst:
|
|
break
|
|
else:
|
|
n_frames_per_burst = 1
|
|
self.log.info("[Modem] calculated frames_per_burst:", n=n_frames_per_burst)
|
|
|
|
tempbuffer = []
|
|
self.rpt_request_buffer = []
|
|
# Append data frames with n_frames_per_burst to tempbuffer
|
|
for n_frame in range(n_frames_per_burst):
|
|
arqheader = bytearray()
|
|
arqheader[:1] = bytes([FR_TYPE.BURST_01.value + n_frame])
|
|
#####arqheader[:1] = bytes([FR_TYPE.BURST_01.value])
|
|
arqheader[1:2] = bytes([n_frames_per_burst])
|
|
arqheader[2:3] = self.session_id
|
|
|
|
# only check for buffer position if at least one NACK received
|
|
self.log.info("[Modem] ----- data buffer position:", iss_buffer_pos=bufferposition,
|
|
irs_bufferposition=self.irs_buffer_position)
|
|
if self.frame_nack_counter > 0 and self.irs_buffer_position != bufferposition:
|
|
self.log.error("[Modem] ----- data buffer offset:", iss_buffer_pos=bufferposition,
|
|
irs_bufferposition=self.irs_buffer_position)
|
|
# only adjust buffer position for experimental versions
|
|
if self.enable_experimental_features:
|
|
self.log.warning("[Modem] ----- data adjustment enabled!")
|
|
bufferposition = self.irs_buffer_position
|
|
|
|
bufferposition_end = bufferposition + payload_per_frame - len(arqheader)
|
|
|
|
# Normal condition
|
|
if bufferposition_end <= len(data_out):
|
|
frame = data_out[bufferposition:bufferposition_end]
|
|
frame = arqheader + frame
|
|
|
|
# Pad the last bytes of a frame
|
|
else:
|
|
extended_data_out = data_out[bufferposition:]
|
|
extended_data_out += bytes([0]) * (
|
|
payload_per_frame - len(extended_data_out) - len(arqheader)
|
|
)
|
|
frame = arqheader + extended_data_out
|
|
|
|
######tempbuffer = frame # [frame]
|
|
tempbuffer.append(frame)
|
|
# add data to our repeat request buffer for easy access if we received a request
|
|
self.rpt_request_buffer.append(frame)
|
|
# set new buffer position
|
|
bufferposition = bufferposition_end
|
|
|
|
self.log.debug("[Modem] tempbuffer:", tempbuffer=tempbuffer)
|
|
self.log.info(
|
|
"[Modem] ARQ | TX | FRAMES",
|
|
mode=FREEDV_MODE(data_mode).name,
|
|
fpb=n_frames_per_burst,
|
|
retry=self.tx_n_retry_of_burst,
|
|
)
|
|
|
|
self.enqueue_frame_for_tx(tempbuffer, c2_mode=data_mode)
|
|
|
|
# After transmission finished, wait for an ACK or RPT frame
|
|
while (
|
|
self.states.is_arq_state
|
|
and not self.burst_ack
|
|
and not self.burst_nack
|
|
and not self.rpt_request_received
|
|
and not self.data_frame_ack_received
|
|
):
|
|
threading.Event().wait(0.01)
|
|
|
|
# Once we receive a burst ack, reset its state and break the RETRIES loop
|
|
if self.burst_ack:
|
|
self.burst_ack = False # reset ack state
|
|
self.tx_n_retry_of_burst = 0 # reset retries
|
|
self.log.debug(
|
|
"[Modem] arq_transmit: Received BURST ACK. Sending next chunk."
|
|
, irs_snr=self.burst_ack_snr)
|
|
# update temp bufferposition for n frames per burst early calculation
|
|
bufferposition_burst_start = bufferposition_end
|
|
break # break retry loop
|
|
|
|
if self.data_frame_ack_received:
|
|
self.log.debug(
|
|
"[Modem] arq_transmit: Received FRAME ACK. Braking retry loop."
|
|
)
|
|
break # break retry loop
|
|
|
|
if self.burst_nack:
|
|
self.tx_n_retry_of_burst += 1
|
|
|
|
self.log.warning(
|
|
"[Modem] arq_transmit: Received BURST NACK. Resending data",
|
|
bufferposition_burst_start=bufferposition_burst_start,
|
|
bufferposition=bufferposition
|
|
)
|
|
|
|
bufferposition = bufferposition_burst_start
|
|
self.burst_nack = False # reset nack state
|
|
|
|
# We need this part for leaving the repeat loop
|
|
# self.states.is_arq_state == "DATA" --> when stopping transmission manually
|
|
if not self.states.is_arq_state:
|
|
self.log.debug(
|
|
"[Modem] arq_transmit: ARQ State changed to FALSE. Breaking retry loop."
|
|
)
|
|
break
|
|
|
|
self.calculate_transfer_rate_tx(
|
|
tx_start_of_transmission, bufferposition_end, len(data_out)
|
|
)
|
|
# NEXT ATTEMPT
|
|
self.log.debug(
|
|
"[Modem] ATTEMPT:",
|
|
retry=self.tx_n_retry_of_burst,
|
|
maxretries=self.tx_n_max_retries_per_burst,
|
|
)
|
|
|
|
# update buffer position
|
|
bufferposition = bufferposition_end
|
|
|
|
# update stats
|
|
self.calculate_transfer_rate_tx(
|
|
tx_start_of_transmission, bufferposition_end, len(data_out)
|
|
)
|
|
|
|
self.event_manager.send_custom_event(
|
|
freedata="modem-message",
|
|
arq="transmission",
|
|
status="transmitting",
|
|
uuid=self.transmission_uuid,
|
|
percent=self.states.arq_transmission_percent,
|
|
bytesperminute=self.states.arq_bytes_per_minute,
|
|
compression=self.arq_compression_factor,
|
|
finished=self.states.arq_seconds_until_finish,
|
|
irs_snr=self.burst_ack_snr,
|
|
mycallsign=str(self.mycallsign, 'UTF-8'),
|
|
dxcallsign=str(self.dxcallsign, 'UTF-8'),
|
|
irs=helpers.bool_to_string(self.is_IRS)
|
|
)
|
|
|
|
# Stay in the while loop until we receive a data_frame_ack. Otherwise,
|
|
# the loop exits after sending the last frame only once and doesn't
|
|
# wait for an acknowledgement.
|
|
if self.data_frame_ack_received and bufferposition > len(data_out):
|
|
self.log.debug("[Modem] arq_tx: Last fragment sent and acknowledged.")
|
|
break
|
|
# GOING TO NEXT ITERATION
|
|
|
|
if self.data_frame_ack_received:
|
|
self.arq_transmit_success()
|
|
else:
|
|
self.arq_transmit_failed()
|
|
|
|
if TESTMODE:
|
|
# Quit after transmission
|
|
self.log.debug("[Modem] TESTMODE: arq_transmit exiting.")
|
|
sys.exit(0)
|
|
|
|
def arq_transmit_success(self):
|
|
"""
|
|
will be called if we successfully transmitted all of queued data
|
|
|
|
"""
|
|
# we need to wait until sending "transmitted" state
|
|
# gui database is too slow for handling this within 0.001 seconds
|
|
# so let's sleep a little
|
|
threading.Event().wait(0.2)
|
|
self.event_manager.send_custom_event(
|
|
freedata="modem-message",
|
|
arq="transmission",
|
|
status="transmitted",
|
|
uuid=self.transmission_uuid,
|
|
percent=self.states.arq_transmission_percent,
|
|
bytesperminute=self.states.arq_bytes_per_minute,
|
|
compression=self.arq_compression_factor,
|
|
finished=self.states.arq_seconds_until_finish,
|
|
mycallsign=str(self.mycallsign, 'UTF-8'),
|
|
dxcallsign=str(self.dxcallsign, 'UTF-8'),
|
|
irs=helpers.bool_to_string(self.is_IRS),
|
|
nacks=self.frame_nack_counter,
|
|
speed_list=self.states.arq_speed_list
|
|
)
|
|
|
|
self.log.info(
|
|
"[Modem] ARQ | TX | DATA TRANSMITTED!",
|
|
BytesPerMinute=self.states.arq_bytes_per_minute,
|
|
total_bytes=self.states.arq_total_bytes,
|
|
BitsPerSecond=self.states.arq_bits_per_second,
|
|
)
|
|
|
|
# finally do an arq cleanup
|
|
self.arq_cleanup()
|
|
|
|
def arq_transmit_failed(self):
|
|
"""
|
|
will be called if we not successfully transmitted all of queued data
|
|
"""
|
|
self.event_manager.send_custom_event(
|
|
freedata="modem-message",
|
|
arq="transmission",
|
|
status="failed",
|
|
uuid=self.transmission_uuid,
|
|
percent=self.states.arq_transmission_percent,
|
|
bytesperminute=self.states.arq_bytes_per_minute,
|
|
compression=self.arq_compression_factor,
|
|
mycallsign=str(self.mycallsign, 'UTF-8'),
|
|
dxcallsign=str(self.dxcallsign, 'UTF-8'),
|
|
irs=helpers.bool_to_string(self.is_IRS),
|
|
nacks=self.frame_nack_counter,
|
|
speed_list=self.states.arq_speed_list
|
|
)
|
|
|
|
self.log.info(
|
|
"[Modem] ARQ | TX | TRANSMISSION FAILED OR TIME OUT!")
|
|
|
|
self.stop_transmission()
|
|
|
|
def burst_ack_nack_received(self, data_in: bytes, snr) -> None:
|
|
"""
|
|
Received an ACK/NACK for a transmitted frame, keep track and
|
|
make adjustments to speed level if needed.
|
|
|
|
Args:
|
|
data_in:bytes:
|
|
|
|
Returns:
|
|
|
|
"""
|
|
# Process data only if we are in ARQ and BUSY state
|
|
if self.states.is_arq_state:
|
|
self.dxgrid = b'------'
|
|
helpers.add_to_heard_stations(
|
|
self.dxcallsign,
|
|
self.dxgrid,
|
|
"DATA",
|
|
snr,
|
|
self.modem_frequency_offset,
|
|
self.states.radio_frequency,
|
|
self.states.heard_stations
|
|
)
|
|
|
|
frametype = int.from_bytes(bytes(data_in[:1]), "big")
|
|
if frametype == FR_TYPE.BURST_ACK.value:
|
|
# Increase speed level if we received a burst ack
|
|
# self.speed_level = min(self.speed_level + 1, len(self.mode_list) - 1)
|
|
# Force data retry loops of TX Modem to stop and continue with next frame
|
|
self.burst_ack = True
|
|
# Reset burst nack counter
|
|
self.burst_nack_counter = 0
|
|
# Reset n retries per burst counter
|
|
self.n_retries_per_burst = 0
|
|
self.irs_buffer_position = int.from_bytes(data_in[4:8], "big")
|
|
|
|
self.burst_ack_snr = helpers.snr_from_bytes(data_in[2:3])
|
|
else:
|
|
|
|
# Decrease speed level if we received a burst nack
|
|
# self.speed_level = max(self.speed_level - 1, 0)
|
|
# Set flag to retry frame again.
|
|
self.burst_nack = True
|
|
# Increment burst nack counter
|
|
self.burst_nack_counter += 1
|
|
self.burst_ack_snr = 'NaN'
|
|
self.irs_buffer_position = int.from_bytes(data_in[5:9], "big")
|
|
|
|
self.log.warning(
|
|
"[Modem] ARQ | TX | Burst NACK received",
|
|
burst_nack_counter=self.burst_nack_counter,
|
|
irs_buffer_position=self.irs_buffer_position,
|
|
)
|
|
|
|
# Update data_channel timestamp
|
|
self.data_channel_last_received = int(time.time())
|
|
# self.burst_ack_snr = int.from_bytes(bytes(data_in[2:3]), "big")
|
|
self.burst_ack_snr = helpers.snr_from_bytes(data_in[2:3])
|
|
# self.log.info("SNR ON IRS", snr=self.burst_ack_snr)
|
|
|
|
self.speed_level = int.from_bytes(bytes(data_in[3:4]), "big")
|
|
self.states.set("arq_speed_level", self.speed_level)
|
|
|
|
def frame_ack_received(
|
|
self, data_in: bytes, snr # pylint: disable=unused-argument,
|
|
) -> None:
|
|
"""Received an ACK for a transmitted frame"""
|
|
# Process data only if we are in ARQ and BUSY state
|
|
if self.states.is_arq_state:
|
|
self.dxgrid = b'------'
|
|
helpers.add_to_heard_stations(
|
|
self.dxcallsign,
|
|
self.dxgrid,
|
|
"DATA",
|
|
snr,
|
|
self.modem_frequency_offset,
|
|
self.states.radio_frequency,
|
|
self.states.heard_stations
|
|
)
|
|
# Force data loops of Modem to stop and continue with next frame
|
|
self.data_frame_ack_received = True
|
|
# Update arq_session and data_channel timestamp
|
|
self.data_channel_last_received = int(time.time())
|
|
self.arq_session_last_received = int(time.time())
|
|
|
|
def frame_nack_received(
|
|
self, data_in: bytes, snr # pylint: disable=unused-argument
|
|
) -> None:
|
|
"""
|
|
Received a NACK for a transmitted frame
|
|
|
|
Args:
|
|
data_in:bytes:
|
|
|
|
"""
|
|
self.log.warning("[Modem] ARQ FRAME NACK RECEIVED - cleanup!",
|
|
arq="transmission",
|
|
status="failed",
|
|
uuid=self.transmission_uuid,
|
|
percent=self.states.arq_transmission_percent,
|
|
bytesperminute=self.states.arq_bytes_per_minute,
|
|
mycallsign=str(self.mycallsign, 'UTF-8'),
|
|
dxcallsign=str(self.dxcallsign, 'UTF-8'),
|
|
irs=helpers.bool_to_string(self.is_IRS),
|
|
nacks=self.frame_nack_counter,
|
|
speed_list=self.states.arq_speed_list
|
|
)
|
|
|
|
self.dxgrid = b'------'
|
|
helpers.add_to_heard_stations(
|
|
self.dxcallsign,
|
|
self.dxgrid,
|
|
"DATA",
|
|
snr,
|
|
self.modem_frequency_offset,
|
|
self.states.radio_frequency,
|
|
self.states.heard_stations
|
|
)
|
|
self.event_manager.send_custom_event(
|
|
freedata="modem-message",
|
|
arq="transmission",
|
|
status="failed",
|
|
uuid=self.transmission_uuid,
|
|
percent=self.states.arq_transmission_percent,
|
|
bytesperminute=self.states.arq_bytes_per_minute,
|
|
compression=self.arq_compression_factor,
|
|
mycallsign=str(self.mycallsign, 'UTF-8'),
|
|
dxcallsign=str(self.dxcallsign, 'UTF-8'),
|
|
irs=helpers.bool_to_string(self.is_IRS),
|
|
nacks=self.frame_nack_counter,
|
|
speed_list=self.states.arq_speed_list
|
|
)
|
|
# Update data_channel timestamp
|
|
self.arq_session_last_received = int(time.time())
|
|
|
|
self.arq_cleanup()
|
|
|
|
def burst_rpt_received(self, data_in: bytes, snr):
|
|
"""
|
|
Repeat request frame received for transmitted frame
|
|
|
|
Args:
|
|
data_in:bytes:
|
|
|
|
"""
|
|
# Only process data if we are in ARQ and BUSY state
|
|
if not self.states.is_arq_state or not self.states.is_modem_busy:
|
|
return
|
|
self.dxgrid = b'------'
|
|
helpers.add_to_heard_stations(
|
|
self.dxcallsign,
|
|
self.dxgrid,
|
|
"DATA",
|
|
snr,
|
|
self.modem_frequency_offset,
|
|
self.states.radio_frequency,
|
|
self.states.heard_stations
|
|
)
|
|
|
|
self.log.info("[Modem] ARQ REPEAT RECEIVED")
|
|
|
|
# self.rpt_request_received = True
|
|
# Update data_channel timestamp
|
|
self.data_channel_last_received = int(time.time())
|
|
# self.rpt_request_buffer = []
|
|
|
|
missing_area = bytes(data_in[2:12]) # 1:9
|
|
missing_area = missing_area.strip(b"\x00")
|
|
print(missing_area)
|
|
print(self.rpt_request_buffer)
|
|
|
|
tempbuffer_rptframes = []
|
|
for i in range(len(missing_area)):
|
|
print(missing_area[i])
|
|
missing_frames_buffer_position = missing_area[i] - 1
|
|
tempbuffer_rptframes.append(self.rpt_request_buffer[missing_frames_buffer_position])
|
|
|
|
self.log.info("[Modem] SENDING REPEAT....")
|
|
data_mode = self.mode_list[self.speed_level]
|
|
self.enqueue_frame_for_tx(tempbuffer_rptframes, c2_mode=data_mode)
|
|
|
|
|
|
############################################################################################################
|
|
# ARQ SESSION HANDLER
|
|
############################################################################################################
|
|
def arq_session_handler(self, mycallsign, dxcallsign) -> bool:
|
|
"""
|
|
Create a session with `self.dxcallsign` and wait until the session is open.
|
|
|
|
Returns:
|
|
True if the session was opened successfully
|
|
False if the session open request failed
|
|
"""
|
|
|
|
encoded_call = helpers.callsign_to_bytes(mycallsign)
|
|
mycallsign = helpers.bytes_to_callsign(encoded_call)
|
|
|
|
encoded_call = helpers.callsign_to_bytes(dxcallsign)
|
|
dxcallsign = helpers.bytes_to_callsign(encoded_call)
|
|
|
|
self.states.set("dxcallsign", dxcallsign)
|
|
dxcallsign_crc = helpers.get_crc_24(dxcallsign)
|
|
|
|
self.log.info(
|
|
"[Modem] SESSION ["
|
|
+ str(mycallsign, "UTF-8")
|
|
+ "]>> <<["
|
|
+ str(dxcallsign, "UTF-8")
|
|
+ "]",
|
|
self.states.arq_session_state,
|
|
)
|
|
|
|
# wait if we have a channel busy condition
|
|
if self.states.channel_busy:
|
|
self.channel_busy_handler()
|
|
|
|
self.open_session()
|
|
|
|
# wait until data channel is open
|
|
while not self.states.is_arq_session and not self.arq_session_timeout:
|
|
threading.Event().wait(0.01)
|
|
self.states.set("arq_session_state", "connecting")
|
|
self.event_manager.send_custom_event(
|
|
freedata="modem-message",
|
|
arq="session",
|
|
status="connecting",
|
|
mycallsign=str(mycallsign, 'UTF-8'),
|
|
dxcallsign=str(dxcallsign, 'UTF-8'),
|
|
)
|
|
if self.states.is_arq_session and self.states.arq_session_state == "connected":
|
|
# self.states.set("arq_session_state", "connected")
|
|
self.event_manager.send_custom_event(
|
|
freedata="modem-message",
|
|
arq="session",
|
|
status="connected",
|
|
mycallsign=str(mycallsign, 'UTF-8'),
|
|
dxcallsign=str(dxcallsign, 'UTF-8'),
|
|
)
|
|
return True
|
|
|
|
self.log.warning(
|
|
"[Modem] SESSION FAILED ["
|
|
+ str(mycallsign, "UTF-8")
|
|
+ "]>>X<<["
|
|
+ str(dxcallsign, "UTF-8")
|
|
+ "]",
|
|
attempts=self.session_connect_max_retries, # Adjust for 0-based for user display
|
|
reason="maximum connection attempts reached",
|
|
state=self.states.arq_session_state,
|
|
)
|
|
self.states.set("arq_session_state", "failed")
|
|
self.event_manager.send_custom_event(
|
|
freedata="modem-message",
|
|
arq="session",
|
|
status="failed",
|
|
reason="timeout",
|
|
mycallsign=str(mycallsign, 'UTF-8'),
|
|
dxcallsign=str(dxcallsign, 'UTF-8'),
|
|
)
|
|
return False
|
|
|
|
|
|
def open_dc_and_transmit(
|
|
self,
|
|
data_out: bytes,
|
|
transmission_uuid: str,
|
|
mycallsign,
|
|
dxcallsign,
|
|
) -> bool:
|
|
"""
|
|
Open data channel and transmit data
|
|
|
|
Args:
|
|
data_out:bytes:
|
|
transmission_uuid:str:
|
|
mycallsign:bytes:
|
|
|
|
Returns:
|
|
True if the data session was opened and the data was sent
|
|
False if the data session was not opened
|
|
"""
|
|
self.mycallsign = mycallsign
|
|
|
|
# additional step for being sure our callsign is correctly
|
|
# in case we are not getting a station ssid
|
|
# then we are forcing a station ssid = 0
|
|
if not self.states.is_arq_session:
|
|
dxcallsign = helpers.callsign_to_bytes(dxcallsign)
|
|
dxcallsign = helpers.bytes_to_callsign(dxcallsign)
|
|
self.dxcallsign = dxcallsign
|
|
|
|
self.dxcallsign_crc = helpers.get_crc_24(self.dxcallsign)
|
|
|
|
# check if hmac hash is provided
|
|
try:
|
|
self.log.info("[SCK] [HMAC] Looking for salt/token", local=mycallsign, remote=dxcallsign)
|
|
hmac_salt = helpers.get_hmac_salt(dxcallsign, mycallsign)
|
|
self.log.info("[SCK] [HMAC] Salt info", local=mycallsign, remote=dxcallsign, salt=hmac_salt)
|
|
except Exception:
|
|
self.log.warning("[SCK] [HMAC] No salt/token found")
|
|
hmac_salt = ''
|
|
|
|
self.states.set("is_modem_busy", True)
|
|
self.arq_file_transfer = True
|
|
self.beacon_paused = True
|
|
|
|
self.transmission_uuid = transmission_uuid
|
|
|
|
# wait a moment for the case, a heartbeat is already on the way back to us
|
|
# this makes channel establishment more clean
|
|
if self.states.is_arq_session:
|
|
threading.Event().wait(2.5)
|
|
|
|
# init arq state event
|
|
self.arq_state_event = threading.Event()
|
|
|
|
# finally start the channel opening procedure
|
|
self.arq_open_data_channel(mycallsign)
|
|
|
|
# if data channel is open, return true else false
|
|
if self.arq_state_event.is_set():
|
|
# start arq transmission
|
|
self.arq_transmit(data_out, hmac_salt)
|
|
return True
|
|
|
|
else:
|
|
self.log.debug(
|
|
"[Modem] arq_open_data_channel:", transmission_uuid=self.transmission_uuid
|
|
)
|
|
|
|
self.event_manager.send_custom_event(
|
|
freedata="modem-message",
|
|
arq="transmission",
|
|
status="failed",
|
|
reason="unknown",
|
|
uuid=self.transmission_uuid,
|
|
percent=self.states.arq_transmission_percent,
|
|
bytesperminute=self.states.arq_bytes_per_minute,
|
|
compression=self.arq_compression_factor,
|
|
mycallsign=str(self.mycallsign, 'UTF-8'),
|
|
dxcallsign=str(self.dxcallsign, 'UTF-8'),
|
|
irs=helpers.bool_to_string(self.is_IRS),
|
|
nacks=self.frame_nack_counter,
|
|
speed_list=self.states.arq_speed_list
|
|
)
|
|
|
|
self.log.warning(
|
|
"[Modem] ARQ | TX | DATA ["
|
|
+ str(mycallsign, "UTF-8")
|
|
+ "]>>X<<["
|
|
+ str(self.dxcallsign, "UTF-8")
|
|
+ "]"
|
|
)
|
|
|
|
# Attempt to clean up the far-side, if it received the
|
|
# open_session frame and can still hear us.
|
|
self.close_session()
|
|
|
|
# release beacon pause
|
|
self.beacon_paused = False
|
|
|
|
# otherwise return false
|
|
return False
|
|
|
|
def arq_open_data_channel(
|
|
self, mycallsign
|
|
) -> bool:
|
|
"""
|
|
Open an ARQ data channel.
|
|
|
|
Args:
|
|
mycallsign:bytes:
|
|
|
|
Returns:
|
|
True if the data channel was opened successfully
|
|
False if the data channel failed to open
|
|
"""
|
|
# set IRS indicator to false, because we are IRS
|
|
self.is_IRS = False
|
|
|
|
# init a new random session id if we are not in an arq session
|
|
if not self.states.is_arq_session:
|
|
self.session_id = np.random.bytes(1)
|
|
|
|
# Update data_channel timestamp
|
|
self.data_channel_last_received = int(time.time())
|
|
|
|
# check if the Modem is running in low bandwidth mode
|
|
# then set the corresponding frametype and build frame
|
|
if self.low_bandwidth_mode:
|
|
frametype = bytes([FR_TYPE.ARQ_DC_OPEN_N.value])
|
|
self.log.debug("[Modem] Requesting low bandwidth mode")
|
|
else:
|
|
frametype = bytes([FR_TYPE.ARQ_DC_OPEN_W.value])
|
|
self.log.debug("[Modem] Requesting high bandwidth mode")
|
|
|
|
# build connection frame
|
|
connection_frame = self.frame_factory.build_arq_connect(
|
|
session_id=self.session_id,
|
|
destination_crc=self.dxcallsign_crc,
|
|
)
|
|
for attempt in range(self.data_channel_max_retries):
|
|
|
|
self.event_manager.send_custom_event(
|
|
freedata="modem-message",
|
|
arq="transmission",
|
|
status="opening",
|
|
mycallsign=mycallsign,
|
|
dxcallsign=str(self.dxcallsign, 'UTF-8'),
|
|
irs=helpers.bool_to_string(self.is_IRS)
|
|
)
|
|
|
|
self.log.info(
|
|
"[Modem] ARQ | DATA | TX | ["
|
|
+ mycallsign
|
|
+ "]>> <<["
|
|
+ str(self.dxcallsign, "UTF-8")
|
|
+ "]",
|
|
attempt=f"{str(attempt + 1)}/{str(self.data_channel_max_retries)}",
|
|
)
|
|
|
|
# Let's check if we have a busy channel and if we are not in a running arq session.
|
|
if self.states.channel_busy and not self.arq_state_event.is_set() or self.states.is_codec2_traffic:
|
|
self.channel_busy_handler()
|
|
|
|
# if channel free, enqueue frame for tx
|
|
if not self.arq_state_event.is_set():
|
|
self.enqueue_frame_for_tx([connection_frame], c2_mode=FREEDV_MODE.sig0.value, copies=1, repeat_delay=0)
|
|
|
|
# wait until timeout or event set
|
|
|
|
random_wait_time = randrange(int(self.duration_sig1_frame * 10),
|
|
int(self.datachannel_opening_interval * 10), 1) / 10
|
|
self.arq_state_event.wait(timeout=random_wait_time)
|
|
|
|
if self.arq_state_event.is_set():
|
|
return True
|
|
if not self.states.is_modem_busy:
|
|
return False
|
|
|
|
# `data_channel_max_retries` attempts have been sent. Aborting attempt & cleaning up
|
|
return False
|
|
|
|
def arq_received_channel_is_open(self, data_in: bytes, snr) -> None:
|
|
"""
|
|
Called if we received a data channel opener
|
|
Args:
|
|
data_in:bytes:
|
|
|
|
"""
|
|
protocol_version = int.from_bytes(bytes(data_in[13:14]), "big")
|
|
if protocol_version == self.arq_protocol_version:
|
|
self.event_manager.send_custom_event(
|
|
freedata="modem-message",
|
|
arq="transmission",
|
|
status="opened",
|
|
mycallsign=str(self.mycallsign, 'UTF-8'),
|
|
dxcallsign=str(self.dxcallsign, 'UTF-8'),
|
|
irs=helpers.bool_to_string(self.is_IRS)
|
|
)
|
|
frametype = int.from_bytes(bytes(data_in[:1]), "big")
|
|
|
|
if frametype == FR_TYPE.ARQ_DC_OPEN_ACK_N.value:
|
|
self.received_LOW_BANDWIDTH_MODE = True
|
|
self.mode_list = self.mode_list_low_bw
|
|
self.time_list = self.time_list_low_bw
|
|
self.log.debug("[Modem] low bandwidth mode", modes=self.mode_list)
|
|
else:
|
|
self.received_LOW_BANDWIDTH_MODE = False
|
|
self.mode_list = self.mode_list_high_bw
|
|
self.time_list = self.time_list_high_bw
|
|
self.log.debug("[Modem] high bandwidth mode", modes=self.mode_list)
|
|
|
|
# set speed level from session opener frame delegation
|
|
self.speed_level = int.from_bytes(bytes(data_in[8:9]), "big")
|
|
self.log.debug("[Modem] speed level selected for given SNR", speed_level=self.speed_level)
|
|
|
|
self.dxgrid = b'------'
|
|
helpers.add_to_heard_stations(
|
|
self.dxcallsign,
|
|
self.dxgrid,
|
|
"DATA",
|
|
snr,
|
|
self.modem_frequency_offset,
|
|
self.states.radio_frequency,
|
|
self.states.heard_stations
|
|
)
|
|
|
|
self.log.info(
|
|
"[Modem] ARQ | DATA | TX | ["
|
|
+ str(self.mycallsign, "UTF-8")
|
|
+ "]>>|<<["
|
|
+ str(self.dxcallsign, "UTF-8")
|
|
+ "]",
|
|
snr=snr,
|
|
)
|
|
|
|
# as soon as we set ARQ_STATE to True, transmission starts
|
|
self.states.set("is_arq_state", True)
|
|
# also set the ARQ event
|
|
self.arq_state_event.set()
|
|
|
|
# Update data_channel timestamp
|
|
self.data_channel_last_received = int(time.time())
|
|
|
|
else:
|
|
self.event_manager.send_custom_event(
|
|
freedata="modem-message",
|
|
arq="transmission",
|
|
status="failed",
|
|
reason="protocol version missmatch",
|
|
mycallsign=str(self.mycallsign, 'UTF-8'),
|
|
dxcallsign=str(self.dxcallsign, 'UTF-8'),
|
|
irs=helpers.bool_to_string(self.is_IRS)
|
|
)
|
|
self.log.warning(
|
|
"[Modem] protocol version mismatch:",
|
|
received=protocol_version,
|
|
own=self.arq_protocol_version,
|
|
)
|
|
self.stop_transmission()
|
|
|
|
def calculate_transfer_rate_tx(
|
|
self, tx_start_of_transmission: float, sentbytes: int, tx_buffer_length: int
|
|
) -> list:
|
|
"""
|
|
Calculate transfer rate for transmission
|
|
Args:
|
|
tx_start_of_transmission:float:
|
|
sentbytes:int:
|
|
tx_buffer_length:int:
|
|
|
|
Returns: List of:
|
|
bits_per_second: float,
|
|
bytes_per_minute: float,
|
|
transmission_percent: float
|
|
"""
|
|
try:
|
|
arq_transmission_percent = min(
|
|
int((sentbytes / tx_buffer_length) * 100), 100
|
|
)
|
|
|
|
transmissiontime = time.time() - tx_start_of_transmission
|
|
|
|
if sentbytes > 0:
|
|
arq_bits_per_second = int((sentbytes * 8) / transmissiontime)
|
|
bytes_per_minute = int(sentbytes / (transmissiontime / 60))
|
|
arq_seconds_until_finish = int(((tx_buffer_length - sentbytes) / (
|
|
bytes_per_minute * self.arq_compression_factor)) * 60)
|
|
|
|
speed_chart = {"snr": self.burst_ack_snr, "bpm": bytes_per_minute,
|
|
"timestamp": int(time.time())}
|
|
# check if data already in list
|
|
if speed_chart not in self.states.arq_speed_list:
|
|
self.states.arq_speed_list.append(speed_chart)
|
|
|
|
else:
|
|
arq_bits_per_second = 0
|
|
bytes_per_minute = 0
|
|
arq_seconds_until_finish = 0
|
|
|
|
except Exception as err:
|
|
self.log.error(f"[Modem] calculate_transfer_rate_tx: Exception: {err}")
|
|
arq_transmission_percent = 0.0
|
|
arq_bits_per_second = 0
|
|
bytes_per_minute = 0
|
|
|
|
self.states.set("arq_bits_per_second", arq_bits_per_second)
|
|
self.states.set("bytes_per_minute", bytes_per_minute)
|
|
self.states.set("arq_transmission_percent", arq_transmission_percent)
|
|
self.states.set("arq_compression_factor", self.arq_compression_factor)
|