mirror of
https://github.com/DJ2LS/FreeDATA
synced 2024-05-14 08:04:33 +00:00
Short and long data transfer test.
These appear to work reasonably well with an occasional transient failure.
This commit is contained in:
Paul Kronenwetter
parent
0ff2b69895
commit
79a1ba2302
0
test/__init__.py
Normal file
0
test/__init__.py
Normal file
|
|
@ -1,166 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Created on Wed Dec 23 07:04:24 2020
|
||||
|
||||
@author: DJ2LS
|
||||
"""
|
||||
|
||||
import multiprocessing
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from pprint import pformat
|
||||
|
||||
import pytest
|
||||
import structlog
|
||||
|
||||
# pylint: disable=wrong-import-position
|
||||
sys.path.insert(0, "..")
|
||||
sys.path.insert(0, "../tnc")
|
||||
sys.path.insert(0, "test")
|
||||
import helpers
|
||||
import log_handler
|
||||
import util_arq_chat_file_1 as util1
|
||||
import util_arq_chat_file_2 as util2
|
||||
|
||||
log_handler.setup_logging(filename="", level="INFO")
|
||||
|
||||
STATIONS = ["AA2BB", "ZZ9YY"]
|
||||
|
||||
bytes_out = b'{"dt":"f","fn":"zeit.txt","ft":"text\\/plain","d":"data:text\\/plain;base64,MyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5Cg=MyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5Cg=","crc":"123123123"}'
|
||||
|
||||
messages = ["This is a test chat."]
|
||||
PIPE_THREAD_RUNNING = True
|
||||
|
||||
|
||||
def analyze_results(station1: list, station2: list):
|
||||
"""Examine the information retrieved from the sub-processes."""
|
||||
# Data in these lists is either a series of bytes of received data,
|
||||
# or a bytearray of transmitted data from the station.
|
||||
log = structlog.get_logger("analyze_results")
|
||||
|
||||
for s1, s2, text in [(station1, station2, "S1"), (station2, station1, "S2")]:
|
||||
for item in s1:
|
||||
if not isinstance(item, list):
|
||||
continue
|
||||
data = bytes(item[0])
|
||||
frametype = int.from_bytes(data[:1], "big")
|
||||
call1 = helpers.decode_call(helpers.bytes_to_callsign(data[1:4]))
|
||||
call2 = helpers.decode_call(helpers.bytes_to_callsign(data[2:5]))
|
||||
log.debug("analyze_results: callsigns:", call1=call1, call2=call2)
|
||||
|
||||
if data in s2:
|
||||
log.debug(
|
||||
f"analyze_results: {text} no CRC", _frame=frametype, data=data
|
||||
)
|
||||
elif data + helpers.get_crc_16(data) in s2:
|
||||
log.debug(f"analyze_results: {text} CRC16", _frame=frametype, data=data)
|
||||
elif data + helpers.get_crc_24(data) in s2:
|
||||
log.debug(f"analyze_results: {text} CRC24", _frame=frametype, data=data)
|
||||
else:
|
||||
log.debug(
|
||||
f"analyze_results: {text} not received:",
|
||||
_frame=frametype,
|
||||
data=data,
|
||||
)
|
||||
|
||||
# log.debug("Everything")
|
||||
# log.debug("S1:", s1=pformat(s1))
|
||||
# log.debug("S2:", s2=pformat(s2))
|
||||
|
||||
|
||||
# @pytest.mark.parametrize("freedv_mode", ["datac0", "datac1", "datac3"])
|
||||
# @pytest.mark.parametrize("n_frames_per_burst", [1, 2, 3])
|
||||
@pytest.mark.parametrize("freedv_mode", ["datac3"])
|
||||
@pytest.mark.parametrize("n_frames_per_burst", [2])
|
||||
@pytest.mark.parametrize("lowbwmode", [False, True])
|
||||
def test_arq_short(freedv_mode: str, n_frames_per_burst: int, lowbwmode: bool):
|
||||
log = structlog.get_logger("test_arq_short")
|
||||
|
||||
s1_data = []
|
||||
s2_data = []
|
||||
|
||||
def recv_data(buffer: list, pipe):
|
||||
while PIPE_THREAD_RUNNING:
|
||||
if pipe.poll(0.1):
|
||||
buffer.append(pipe.recv())
|
||||
else:
|
||||
time.sleep(0.1)
|
||||
|
||||
def recv_from_pipes(s1_rx, s1_pipe, s2_rx, s2_pipe) -> list:
|
||||
processes = [
|
||||
threading.Thread(target=recv_data, args=(s1_rx, s1_pipe)),
|
||||
threading.Thread(target=recv_data, args=(s2_rx, s2_pipe)),
|
||||
]
|
||||
for item in processes:
|
||||
item.start()
|
||||
|
||||
return processes
|
||||
|
||||
# This sufficiently separates the two halves of the test. This is needed
|
||||
# because both scripts change global state. They would conflict if running in
|
||||
# the same process.
|
||||
from_s1, s1_send = multiprocessing.Pipe()
|
||||
from_s2, s2_send = multiprocessing.Pipe()
|
||||
proc = [
|
||||
multiprocessing.Process(
|
||||
target=util1.t_highsnr_arq_short_station1,
|
||||
args=(
|
||||
s1_send,
|
||||
freedv_mode,
|
||||
n_frames_per_burst,
|
||||
STATIONS[0],
|
||||
STATIONS[1],
|
||||
messages[0],
|
||||
lowbwmode,
|
||||
),
|
||||
daemon=True,
|
||||
),
|
||||
multiprocessing.Process(
|
||||
target=util2.t_highsnr_arq_short_station2,
|
||||
args=(
|
||||
s2_send,
|
||||
freedv_mode,
|
||||
n_frames_per_burst,
|
||||
STATIONS[1],
|
||||
STATIONS[0],
|
||||
messages[0],
|
||||
lowbwmode,
|
||||
),
|
||||
daemon=True,
|
||||
),
|
||||
]
|
||||
|
||||
pipe_receivers = recv_from_pipes(s1_data, from_s1, s2_data, from_s2)
|
||||
# log.debug("Creating ")
|
||||
# print("Starting threads.")
|
||||
for p_item in proc:
|
||||
p_item.start()
|
||||
|
||||
# This relies on each process exiting when its job is complete!
|
||||
|
||||
# print("Waiting for threads to exit.")
|
||||
for p_item in proc:
|
||||
p_item.join()
|
||||
|
||||
global PIPE_THREAD_RUNNING # pylint: disable=global-statement
|
||||
PIPE_THREAD_RUNNING = False
|
||||
for pipe_recv in pipe_receivers:
|
||||
pipe_recv.join()
|
||||
|
||||
# print(f"\n{proc.exitcode=}")
|
||||
for p_item in proc:
|
||||
assert p_item.exitcode == 0
|
||||
p_item.close()
|
||||
|
||||
analyze_results(s1_data, s2_data)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Run pytest with the current script as the filename.
|
||||
ecode = pytest.main(["-s", "-v", sys.argv[0]])
|
||||
if ecode == 0:
|
||||
print("errors: 0")
|
||||
else:
|
||||
print(ecode)
|
||||
198
test/test_data_channel.py
Normal file
198
test/test_data_channel.py
Normal file
|
|
@ -0,0 +1,198 @@
|
|||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Created on Wed Dec 23 07:04:24 2020
|
||||
|
||||
@author: DJ2LS
|
||||
"""
|
||||
|
||||
import contextlib
|
||||
import multiprocessing
|
||||
import sys
|
||||
import test.util_data_channel_1 as util1
|
||||
import test.util_data_channel_2 as util2
|
||||
import threading
|
||||
import time
|
||||
import zlib
|
||||
|
||||
import helpers
|
||||
import log_handler
|
||||
import pytest
|
||||
import static
|
||||
import structlog
|
||||
|
||||
# pylint: disable=wrong-import-position
|
||||
# sys.path.insert(0, "..")
|
||||
# sys.path.insert(0, "../tnc")
|
||||
# sys.path.insert(0, "test")
|
||||
|
||||
log_handler.setup_logging(filename="", level="INFO")
|
||||
|
||||
STATIONS = ["AA2BB", "ZZ9YY"]
|
||||
|
||||
bytes_out = b'{"dt":"f","fn":"zeit.txt","ft":"text\\/plain","d":"data:text\\/plain;base64,MyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5Cg=MyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5Cg=","crc":"123123123"}'
|
||||
|
||||
messages = [
|
||||
"This is a test chat...",
|
||||
"This is a much longer message, hopefully longer than each of the datac1 and datac3 frames available to use in this modem. This should be long enought, but to err on the side of completeness this will string on for many more words before coming to the long awaited conclusion. We are not at the concluding point just yet because there is still more space to be taken up in the datac3 frame. Perhaps now would be a good place to terminate this test message, but perhaps not because we need a few more bytes. Here then we stop. This compresses so well that I need more data, even more stuff than is already here and included in the unreadable diatribe below, or is it a soliloquy? MyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5Cg=MyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5Cg=",
|
||||
]
|
||||
PIPE_THREAD_RUNNING = True
|
||||
|
||||
|
||||
def locate_data_with_crc(source_list: list, text: str, data: bytes, frametype: str):
|
||||
"""Try to locate data in source_list."""
|
||||
log = structlog.get_logger("locate_data_with_crc")
|
||||
|
||||
if data in source_list:
|
||||
with contextlib.suppress():
|
||||
data = zlib.decompress(data[2:])
|
||||
log.info(f"analyze_results: {text} no CRC", _frametype=frametype, data=data)
|
||||
elif data + helpers.get_crc_8(data) in source_list:
|
||||
with contextlib.suppress(zlib.error):
|
||||
data = zlib.decompress(data[2:-1])
|
||||
log.info(f"analyze_results: {text} CRC 8", _frametype=frametype, data=data)
|
||||
elif data + helpers.get_crc_16(data) in source_list:
|
||||
with contextlib.suppress(zlib.error):
|
||||
data = zlib.decompress(data[2:-2])
|
||||
log.info(f"analyze_results: {text} CRC16", _frametype=frametype, data=data)
|
||||
elif data + helpers.get_crc_24(data) in source_list:
|
||||
with contextlib.suppress(zlib.error):
|
||||
data = zlib.decompress(data[2:-3])
|
||||
log.info(f"analyze_results: {text} CRC24", _frametype=frametype, data=data)
|
||||
elif data + helpers.get_crc_32(data) in source_list:
|
||||
with contextlib.suppress(zlib.error):
|
||||
data = zlib.decompress(data[2:-4])
|
||||
log.info(f"analyze_results: {text} CRC32", _frametype=frametype, data=data)
|
||||
else:
|
||||
log.info(
|
||||
f"analyze_results: {text} not received:",
|
||||
_frame=frametype,
|
||||
data=data,
|
||||
)
|
||||
|
||||
|
||||
def analyze_results(station1: list, station2: list, call_list: list):
|
||||
"""Examine the information retrieved from the sub-processes."""
|
||||
# Data in these lists is either a series of bytes of received data,
|
||||
# or a bytearray of transmitted data from the station.
|
||||
log = structlog.get_logger("analyze_results")
|
||||
|
||||
# Check that each station's transmitted data was received by the other.
|
||||
for s1, s2, text in [
|
||||
(station1, station2, call_list[0]),
|
||||
(station2, station1, call_list[1]),
|
||||
]:
|
||||
for s1_item in s1:
|
||||
if not isinstance(s1_item, list):
|
||||
continue
|
||||
data = bytes(s1_item[0])
|
||||
frametypeno = int.from_bytes(data[:1], "big")
|
||||
frametype = static.FRAME_TYPE(frametypeno).name
|
||||
s1_crc = helpers.decode_call(helpers.bytes_to_callsign(data[1:4]))
|
||||
s2_crc = helpers.decode_call(helpers.bytes_to_callsign(data[2:5]))
|
||||
log.info(
|
||||
"analyze_results: callsign CRCs:",
|
||||
tx_station=text,
|
||||
s1_crc=s1_crc,
|
||||
s2_crc=s2_crc,
|
||||
)
|
||||
|
||||
locate_data_with_crc(s2, text, data, frametype)
|
||||
|
||||
# log.debug("Everything")
|
||||
# log.debug("S1:", s1=pformat(s1))
|
||||
# log.debug("S2:", s2=pformat(s2))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("freedv_mode", ["datac1", "datac3"])
|
||||
@pytest.mark.parametrize("n_frames_per_burst", [1]) # Higher fpb is broken.
|
||||
@pytest.mark.parametrize("message_no", range(len(messages)))
|
||||
@pytest.mark.flaky(reruns=1)
|
||||
def test_data_channel(freedv_mode: str, n_frames_per_burst: int, message_no: int):
|
||||
log = structlog.get_logger("test_data_channel")
|
||||
|
||||
s1_data = []
|
||||
s2_data = []
|
||||
|
||||
def recv_data(buffer: list, pipe):
|
||||
while PIPE_THREAD_RUNNING:
|
||||
if pipe.poll(0.1):
|
||||
buffer.append(pipe.recv())
|
||||
else:
|
||||
time.sleep(0.1)
|
||||
|
||||
def recv_from_pipes(s1_rx, s1_pipe, s2_rx, s2_pipe) -> list:
|
||||
processes = [
|
||||
threading.Thread(target=recv_data, args=(s1_rx, s1_pipe)),
|
||||
threading.Thread(target=recv_data, args=(s2_rx, s2_pipe)),
|
||||
]
|
||||
for item in processes:
|
||||
item.start()
|
||||
|
||||
return processes
|
||||
|
||||
# This sufficiently separates the two halves of the test. This is needed
|
||||
# because both scripts change global state. They would conflict if running in
|
||||
# the same process.
|
||||
from_s1, s1_send = multiprocessing.Pipe()
|
||||
from_s2, s2_send = multiprocessing.Pipe()
|
||||
proc = [
|
||||
multiprocessing.Process(
|
||||
target=util1.t_highsnr_arq_short_station1,
|
||||
args=(
|
||||
s1_send,
|
||||
freedv_mode,
|
||||
n_frames_per_burst,
|
||||
STATIONS[0],
|
||||
STATIONS[1],
|
||||
messages[message_no],
|
||||
True, # low bandwidth mode
|
||||
),
|
||||
daemon=True,
|
||||
),
|
||||
multiprocessing.Process(
|
||||
target=util2.t_highsnr_arq_short_station2,
|
||||
args=(
|
||||
s2_send,
|
||||
freedv_mode,
|
||||
n_frames_per_burst,
|
||||
STATIONS[1],
|
||||
STATIONS[0],
|
||||
messages[message_no],
|
||||
True, # low bandwidth mode
|
||||
),
|
||||
daemon=True,
|
||||
),
|
||||
]
|
||||
|
||||
pipe_receivers = recv_from_pipes(s1_data, from_s1, s2_data, from_s2)
|
||||
# log.debug("Creating ")
|
||||
# print("Starting threads.")
|
||||
for p_item in proc:
|
||||
p_item.start()
|
||||
|
||||
# This relies on each process exiting when its job is complete!
|
||||
|
||||
# print("Waiting for threads to exit.")
|
||||
for p_item in proc:
|
||||
p_item.join()
|
||||
|
||||
global PIPE_THREAD_RUNNING # pylint: disable=global-statement
|
||||
PIPE_THREAD_RUNNING = False
|
||||
for pipe_recv in pipe_receivers:
|
||||
pipe_recv.join()
|
||||
|
||||
for p_item in proc:
|
||||
assert p_item.exitcode == 0
|
||||
p_item.close()
|
||||
|
||||
analyze_results(s1_data, s2_data, STATIONS)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Run pytest with the current script as the filename.
|
||||
ecode = pytest.main(["-s", "-v", sys.argv[0]])
|
||||
if ecode == 0:
|
||||
print("errors: 0")
|
||||
else:
|
||||
print(ecode)
|
||||
|
|
@ -6,7 +6,6 @@ Created on Wed Dec 23 07:04:24 2020
|
|||
@author: DJ2LS
|
||||
"""
|
||||
|
||||
import array
|
||||
import base64
|
||||
import json
|
||||
import sys
|
||||
|
|
@ -25,53 +24,15 @@ import sock
|
|||
import static
|
||||
|
||||
|
||||
def t_highsnr_arq_short_station1(
|
||||
parent_pipe,
|
||||
freedv_mode: str,
|
||||
n_frames_per_burst: int,
|
||||
def t_setup(
|
||||
mycall: str,
|
||||
dxcall: str,
|
||||
message: str,
|
||||
lowbwmode: bool,
|
||||
t_transmit,
|
||||
t_process_data,
|
||||
):
|
||||
log = structlog.get_logger("station1")
|
||||
orig_tx_func: object
|
||||
orig_rx_func: object
|
||||
|
||||
def t_transmit(self, mode, repeats: int, repeat_delay: int, frames: bytearray):
|
||||
"""'Wrap' RF.transmit function to extract the arguments."""
|
||||
nonlocal orig_tx_func, parent_pipe
|
||||
|
||||
t_frames = frames
|
||||
parent_pipe.send(t_frames)
|
||||
# log.info("S1 TX: ", frames=t_frames)
|
||||
for item in t_frames:
|
||||
frametype = int.from_bytes(item[:1], "big")
|
||||
log.info("S1 TX: ", frametype=static.FRAME_TYPE(frametype).name)
|
||||
|
||||
# Apologies for the Python "magic." "orig_func" is a pointer to the
|
||||
# original function captured before this one was put in place.
|
||||
orig_tx_func(self, mode, repeats, repeat_delay, frames) # type: ignore
|
||||
|
||||
def t_process_data(self, bytes_out, freedv, bytes_per_frame: int):
|
||||
"""'Wrap' DATA.process_data function to extract the arguments."""
|
||||
nonlocal orig_rx_func, parent_pipe
|
||||
t_bytes_out = bytes(bytes_out)
|
||||
parent_pipe.send(t_bytes_out)
|
||||
log.debug(
|
||||
"S1 RX: ",
|
||||
bytes_out=t_bytes_out,
|
||||
bytes_per_frame=bytes_per_frame,
|
||||
)
|
||||
frametype = int.from_bytes(t_bytes_out[:1], "big")
|
||||
log.info("S1 RX: ", frametype=static.FRAME_TYPE(frametype).name)
|
||||
|
||||
# Apologies for the Python "magic." "orig_func" is a pointer to the
|
||||
# original function captured before this one was put in place.
|
||||
orig_rx_func(self, bytes_out, freedv, bytes_per_frame) # type: ignore
|
||||
|
||||
# enable testmode
|
||||
data_handler.TESTMODE = True
|
||||
# Disable data_handler testmode - This is required to test a conversation.
|
||||
data_handler.TESTMODE = False
|
||||
modem.RXCHANNEL = "/tmp/hfchannel1"
|
||||
modem.TESTMODE = True
|
||||
modem.TXCHANNEL = "/tmp/hfchannel2"
|
||||
|
|
@ -96,6 +57,10 @@ def t_highsnr_arq_short_station1(
|
|||
orig_rx_func = data_handler.DATA.process_data
|
||||
data_handler.DATA.process_data = t_process_data
|
||||
tnc.log = structlog.get_logger("station1_DATA")
|
||||
# Limit the frame-ack timeout
|
||||
tnc.time_list_low_bw = [3, 1, 1]
|
||||
tnc.time_list_high_bw = [3, 1, 1]
|
||||
tnc.time_list = [3, 1, 1]
|
||||
|
||||
# Create the modem
|
||||
t_modem = modem.RF()
|
||||
|
|
@ -103,13 +68,70 @@ def t_highsnr_arq_short_station1(
|
|||
modem.RF.transmit = t_transmit
|
||||
t_modem.log = structlog.get_logger("station1_RF")
|
||||
|
||||
time.sleep(0.5)
|
||||
return tnc, orig_rx_func, orig_tx_func
|
||||
|
||||
|
||||
def t_highsnr_arq_short_station1(
|
||||
parent_pipe,
|
||||
freedv_mode: str,
|
||||
n_frames_per_burst: int,
|
||||
mycall: str,
|
||||
dxcall: str,
|
||||
message: str,
|
||||
lowbwmode: bool,
|
||||
):
|
||||
log = structlog.get_logger("station1")
|
||||
orig_tx_func: object
|
||||
orig_rx_func: object
|
||||
|
||||
def t_transmit(self, mode, repeats: int, repeat_delay: int, frames: bytearray):
|
||||
"""'Wrap' RF.transmit function to extract the arguments."""
|
||||
nonlocal orig_tx_func, parent_pipe
|
||||
|
||||
t_frames = frames
|
||||
parent_pipe.send(t_frames)
|
||||
# log.info("S1 TX: ", frames=t_frames)
|
||||
for item in t_frames:
|
||||
frametype = int.from_bytes(item[:1], "big")
|
||||
log.info(
|
||||
"S1 TX: ", TX=static.FRAME_TYPE(frametype).name, frametypeno=frametype
|
||||
)
|
||||
|
||||
# Apologies for the Python "magic." "orig_func" is a pointer to the
|
||||
# original function captured before this one was put in place.
|
||||
orig_tx_func(self, mode, repeats, repeat_delay, frames) # type: ignore
|
||||
|
||||
def t_process_data(self, bytes_out, freedv, bytes_per_frame: int):
|
||||
"""'Wrap' DATA.process_data function to extract the arguments."""
|
||||
nonlocal orig_rx_func, parent_pipe
|
||||
t_bytes_out = bytes(bytes_out)
|
||||
parent_pipe.send(t_bytes_out)
|
||||
log.debug(
|
||||
"S1 RX: ",
|
||||
bytes_out=t_bytes_out,
|
||||
bytes_per_frame=bytes_per_frame,
|
||||
)
|
||||
frametype = int.from_bytes(t_bytes_out[:1], "big")
|
||||
log.info("S1 RX: ", RX=static.FRAME_TYPE(frametype).name, frametypeno=frametype)
|
||||
|
||||
# Apologies for the Python "magic." "orig_func" is a pointer to the
|
||||
# original function captured before this one was put in place.
|
||||
orig_rx_func(self, bytes_out, freedv, bytes_per_frame) # type: ignore
|
||||
|
||||
tnc, orig_rx_func, orig_tx_func = t_setup(
|
||||
mycall, dxcall, lowbwmode, t_transmit, t_process_data
|
||||
)
|
||||
|
||||
# Try to wait for the far-end to start up.
|
||||
# time.sleep(0.5)
|
||||
|
||||
# Construct message to dxstation.
|
||||
# data = {"type": "ping", "command": "ping", "dxcallsign": dxcall}
|
||||
# data = {"type": "", "command": "cqcqcq", "dxcallsign": ""}
|
||||
# sock.process_tnc_commands(ujson.dumps(data))
|
||||
b64_str = str(base64.encodebytes(bytes(message, "UTF-8")), "UTF-8").strip()
|
||||
# Tests seem to work better when this side sends a 'dummy' frame first.
|
||||
data = {"type": "ping", "command": "ping", "dxcallsign": 'BB4NN'}
|
||||
sock.process_tnc_commands(json.dumps(data, indent=None))
|
||||
time.sleep(0.5)
|
||||
|
||||
b64_str = str(base64.b64encode(bytes(message, "UTF-8")), "UTF-8").strip()
|
||||
data = {
|
||||
"type": "arq",
|
||||
"command": "send_raw",
|
||||
|
|
@ -118,50 +140,37 @@ def t_highsnr_arq_short_station1(
|
|||
"data": b64_str,
|
||||
"dxcallsign": dxcall,
|
||||
"mode": codec2.FREEDV_MODE[freedv_mode].value,
|
||||
"n_frames": 1,
|
||||
"n_frames": n_frames_per_burst,
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
sock.process_tnc_commands(json.dumps(data))
|
||||
sock.process_tnc_commands(json.dumps(data, indent=None))
|
||||
time.sleep(0.5)
|
||||
|
||||
# sock.process_tnc_commands(json.dumps(data))
|
||||
# time.sleep(0.5)
|
||||
|
||||
# sock.process_tnc_commands(json.dumps(data))
|
||||
timeout = time.time() + 45
|
||||
# This transaction should take less than 14 sec.
|
||||
timeout = time.time() + 14
|
||||
while "ARQ;TRANSMITTING;SUCCESS" not in static.INFO:
|
||||
if time.time() > timeout:
|
||||
log.warning("station1", first=True)
|
||||
log.warning("station1 TIMEOUT", first=True)
|
||||
break
|
||||
time.sleep(0.1)
|
||||
log.info("station1, first", arq_state=pformat(static.ARQ_STATE))
|
||||
|
||||
# Construct disconnect message to dxstation.
|
||||
# data = {"type": "ping", "command": "ping", "dxcallsign": dxcall}
|
||||
data = {"type": "arq", "command": "disconnect", "dxcallsign": dxcall}
|
||||
sock.process_tnc_commands(json.dumps(data, indent=None))
|
||||
|
||||
# data = {"type": "arq", "command": "disconnect", "dxcallsign": dxcall}
|
||||
# sock.process_tnc_commands(json.dumps(data))
|
||||
# time.sleep(0.5)
|
||||
|
||||
# sock.process_tnc_commands(json.dumps(data))
|
||||
# time.sleep(0.5)
|
||||
|
||||
# sock.process_tnc_commands(json.dumps(data))
|
||||
# time.sleep(1)
|
||||
# sock.process_tnc_commands(json.dumps(data))
|
||||
# time.sleep(0.5)
|
||||
|
||||
timeout = time.time() + 40
|
||||
while static.ARQ_STATE and tnc.data_queue_transmit.queue:
|
||||
# Allow enough time for this side to process the disconnect frame.
|
||||
timeout = time.time() + 10
|
||||
while static.ARQ_STATE or tnc.data_queue_transmit.queue:
|
||||
if time.time() > timeout:
|
||||
log.error("station1", TIMEOUT=True)
|
||||
break
|
||||
time.sleep(0.5)
|
||||
log.info("station1", arq_state=pformat(static.ARQ_STATE))
|
||||
|
||||
log.info("S1 DQT: ", info=pformat(tnc.data_queue_transmit.queue))
|
||||
log.info("S1 DQR: ", info=pformat(tnc.data_queue_received.queue))
|
||||
log.info("S1 DQT: ", DQ_Tx=pformat(tnc.data_queue_transmit.queue))
|
||||
log.info("S1 DQR: ", DQ_Rx=pformat(tnc.data_queue_received.queue))
|
||||
log.info("S1 Info: ", info=static.INFO)
|
||||
assert "DATACHANNEL;OPENING" in static.INFO
|
||||
assert "DATACHANNEL;OPEN" in static.INFO
|
||||
|
|
@ -6,8 +6,6 @@ Created on Wed Dec 23 07:04:24 2020
|
|||
@author: DJ2LS
|
||||
"""
|
||||
|
||||
import array
|
||||
import base64
|
||||
import json
|
||||
import sys
|
||||
import time
|
||||
|
|
@ -17,61 +15,24 @@ import structlog
|
|||
|
||||
sys.path.insert(0, "..")
|
||||
sys.path.insert(0, "../tnc")
|
||||
import codec2
|
||||
import data_handler
|
||||
import helpers
|
||||
import modem
|
||||
import sock
|
||||
import static
|
||||
|
||||
FIRST_RX = True
|
||||
|
||||
def t_highsnr_arq_short_station2(
|
||||
parent_pipe,
|
||||
freedv_mode: str,
|
||||
n_frames_per_burst: int,
|
||||
|
||||
def t_setup(
|
||||
mycall: str,
|
||||
dxcall: str,
|
||||
message: str,
|
||||
lowbwmode: bool,
|
||||
t_transmit,
|
||||
t_process_data,
|
||||
):
|
||||
log = structlog.get_logger("station2")
|
||||
orig_tx_func: object
|
||||
orig_rx_func: object
|
||||
|
||||
def t_transmit(self, mode, repeats: int, repeat_delay: int, frames: bytearray):
|
||||
"""'Wrap' RF.transmit function to extract the arguments."""
|
||||
nonlocal orig_tx_func, parent_pipe
|
||||
|
||||
t_frames = frames
|
||||
parent_pipe.send(t_frames)
|
||||
# log.info("S2 TX: ", frames=t_frames)
|
||||
for item in t_frames:
|
||||
frametype = int.from_bytes(item[:1], "big")
|
||||
log.info("S2 TX: ", frametype=static.FRAME_TYPE(frametype).name)
|
||||
|
||||
# Apologies for the Python "magic." "orig_func" is a pointer to the
|
||||
# original function captured before this one was put in place.
|
||||
orig_tx_func(self, mode, repeats, repeat_delay, frames) # type: ignore
|
||||
|
||||
def t_process_data(self, bytes_out, freedv, bytes_per_frame: int):
|
||||
"""'Wrap' DATA.process_data function to extract the arguments."""
|
||||
nonlocal orig_rx_func, parent_pipe
|
||||
t_bytes_out = bytes(bytes_out)
|
||||
parent_pipe.send(t_bytes_out)
|
||||
log.debug(
|
||||
"S2 RX: ",
|
||||
bytes_out=t_bytes_out,
|
||||
bytes_per_frame=bytes_per_frame,
|
||||
)
|
||||
frametype = int.from_bytes(t_bytes_out[:1], "big")
|
||||
log.info("S2 RX: ", frametype=static.FRAME_TYPE(frametype).name)
|
||||
|
||||
# Apologies for the Python "magic." "orig_func" is a pointer to the
|
||||
# original function captured before this one was put in place.
|
||||
orig_rx_func(self, bytes_out, freedv, bytes_per_frame) # type: ignore
|
||||
|
||||
# enable testmode
|
||||
data_handler.TESTMODE = True
|
||||
# Disable data_handler testmode - This is required to test a conversation.
|
||||
data_handler.TESTMODE = False
|
||||
modem.RXCHANNEL = "/tmp/hfchannel2"
|
||||
modem.TESTMODE = True
|
||||
modem.TXCHANNEL = "/tmp/hfchannel1"
|
||||
|
|
@ -96,6 +57,10 @@ def t_highsnr_arq_short_station2(
|
|||
orig_rx_func = data_handler.DATA.process_data
|
||||
data_handler.DATA.process_data = t_process_data
|
||||
tnc.log = structlog.get_logger("station2_DATA")
|
||||
# Limit the frame-ack timeout
|
||||
tnc.time_list_low_bw = [1, 1, 1]
|
||||
tnc.time_list_high_bw = [1, 1, 1]
|
||||
tnc.time_list = [1, 1, 1]
|
||||
|
||||
# Create the modem
|
||||
t_modem = modem.RF()
|
||||
|
|
@ -103,25 +68,92 @@ def t_highsnr_arq_short_station2(
|
|||
modem.RF.transmit = t_transmit
|
||||
t_modem.log = structlog.get_logger("station2_RF")
|
||||
|
||||
timeout = time.time() + 45
|
||||
return tnc, orig_rx_func, orig_tx_func
|
||||
|
||||
|
||||
def t_highsnr_arq_short_station2(
|
||||
parent_pipe,
|
||||
freedv_mode: str,
|
||||
n_frames_per_burst: int,
|
||||
mycall: str,
|
||||
dxcall: str,
|
||||
message: str,
|
||||
lowbwmode: bool,
|
||||
):
|
||||
log = structlog.get_logger("station2")
|
||||
orig_tx_func: object
|
||||
orig_rx_func: object
|
||||
|
||||
def t_transmit(self, mode, repeats: int, repeat_delay: int, frames: bytearray):
|
||||
"""'Wrap' RF.transmit function to extract the arguments."""
|
||||
nonlocal orig_tx_func, parent_pipe
|
||||
|
||||
t_frames = frames
|
||||
parent_pipe.send(t_frames)
|
||||
# log.info("S2 TX: ", frames=t_frames)
|
||||
for item in t_frames:
|
||||
frametype = int.from_bytes(item[:1], "big")
|
||||
log.info("S2 TX: ", TX=static.FRAME_TYPE(frametype).name, frametypeno=frametype)
|
||||
|
||||
# Apologies for the Python "magic." "orig_func" is a pointer to the
|
||||
# original function captured before this one was put in place.
|
||||
orig_tx_func(self, mode, repeats, repeat_delay, frames) # type: ignore
|
||||
|
||||
def t_process_data(self, bytes_out, freedv, bytes_per_frame: int):
|
||||
"""'Wrap' DATA.process_data function to extract the arguments."""
|
||||
nonlocal orig_rx_func, parent_pipe
|
||||
# global FIRST_RX
|
||||
# if FIRST_RX:
|
||||
# FIRST_RX = False
|
||||
# # Swallow the first frame
|
||||
# orig_rx_func(self, bytes_out, freedv, bytes_per_frame) # type: ignore
|
||||
|
||||
t_bytes_out = bytes(bytes_out)
|
||||
parent_pipe.send(t_bytes_out)
|
||||
log.debug(
|
||||
"S2 RX: ",
|
||||
bytes_out=t_bytes_out,
|
||||
bytes_per_frame=bytes_per_frame,
|
||||
)
|
||||
frametype = int.from_bytes(t_bytes_out[:1], "big")
|
||||
log.info("S2 RX: ", RX=static.FRAME_TYPE(frametype).name, frametypeno=frametype)
|
||||
|
||||
# Apologies for the Python "magic." "orig_func" is a pointer to the
|
||||
# original function captured before this one was put in place.
|
||||
orig_rx_func(self, bytes_out, freedv, bytes_per_frame) # type: ignore
|
||||
|
||||
tnc, orig_rx_func, orig_tx_func = t_setup(
|
||||
mycall, dxcall, lowbwmode, t_transmit, t_process_data
|
||||
)
|
||||
|
||||
# data = {"type": "set", "command": "send_test_frame"}
|
||||
# sock.process_tnc_commands(json.dumps(data, indent=None))
|
||||
# time.sleep(0.5)
|
||||
# time.sleep(2)
|
||||
|
||||
# This transaction should take less than 14 sec.
|
||||
timeout = time.time() + 14
|
||||
while "ARQ;RECEIVING;SUCCESS" not in static.INFO or static.ARQ_STATE:
|
||||
if time.time() > timeout:
|
||||
log.error("station2", first=True)
|
||||
log.warning("station2 TIMEOUT", first=True)
|
||||
break
|
||||
time.sleep(0.5)
|
||||
log.info("station2", arq_state=pformat(static.ARQ_STATE))
|
||||
log.info("station2, first", arq_state=pformat(static.ARQ_STATE))
|
||||
|
||||
# Allow enough time for this side to transmit the final ACK.
|
||||
timeout = time.time() + 40
|
||||
while static.ARQ_STATE:
|
||||
# Allow enough time for this side to receive the disconnect frame.
|
||||
timeout = time.time() + 10
|
||||
while "ARQ;SESSION;CLOSE" not in static.INFO:
|
||||
if time.time() > timeout:
|
||||
log.error("station2", TIMEOUT=True)
|
||||
break
|
||||
time.sleep(0.5)
|
||||
log.info("station2", arq_state=pformat(static.ARQ_STATE))
|
||||
|
||||
log.info("S2 DQT: ", info=pformat(tnc.data_queue_transmit.queue))
|
||||
log.info("S2 DQR: ", info=pformat(tnc.data_queue_received.queue))
|
||||
# data = {"type": "arq", "command": "disconnect", "dxcallsign": dxcall}
|
||||
# sock.process_tnc_commands(json.dumps(data, indent=None))
|
||||
|
||||
log.info("S2 DQT: ", DQ_Tx=pformat(tnc.data_queue_transmit.queue))
|
||||
log.info("S2 DQR: ", DQ_Rx=pformat(tnc.data_queue_received.queue))
|
||||
log.info("S2 Info: ", info=static.INFO)
|
||||
assert "DATACHANNEL;RECEIVEDOPENER" in static.INFO
|
||||
# assert "QRV;SENDING" in static.INFO
|
||||
Loading…
Reference in a new issue