FreeDATA/test/test_multimode_rx.py
Paul Kronenwetter 5c6cee1c21
Enhance tests (#185)
* Initial attempt to create unit tests for DATA class (tnc).

* Completed initial set of tests.

* Adding pytest to install packages.

* Corrects issue #173 fix

I didn't look carefully enough at `helpers.check_callsign` to see that it returns a list. The first element of the list is `True`/`False`.

* Making check_callsign evaluation more consistent.

* Update .gitignore

this is more a test commit to see if GitHub Client for MacOS is working,

* mkfifo test prototype

First partially working prototype for testing the full tnc with mkfifo named pipes.

* single tnc test file

moved to a single file for running tnc tests

* fixed typo

* Added parameters to tests.

Make other minor tweaks and documentation.

* Clean up two existing tests.

Adapted both tests to pytest and maintained
compatibility with existing ctest method.
Tweaked CMakeLists.txt .

* Adding pure python  highsnr_stdio_P_P_multi test.

Intended to replace highsnr_stdio_P_P_multi which uses POSIX shell.

* Adding pure python  highsnr_stdio_P_P_datac0 test.

Intended to replace highsnr_stdio_P_P_datac0 which uses POSIX shell.

* Parameterize recent tests.

Renamed datac0 to datacx after including all data codecs in test.

* Parameterize mode as well.

Add ability to run tests from main directory as well as within test/.

* Add list of tests and brief descriptions.

* Add more native python tests conversions.

* Update README with new tests.

* Tweak README again.

* Rename test to be findable by pytest.

* Rename test for ctest.

* Update correct file this time.

* Minor test tweaks.

* Add modem test proof-of-concept.

* Adjustment to ARQ short test.

* Various refactorings.

Type hints, trailing backslash, range usage, etc.

* Ignore unknown arguments in argparse.

* Minor cleanups.

* Update test/README.md.

* Update test_pa to quiet pylint.

* Give up trying to suppress structlog output.

* Correct module comments.

* Remove excess trailing spaces.

* Remove excess newlines.

* Various refactorings.

Type hints, trailing backslash, range usage, etc.

* mkfifo test prototype

First partially working prototype for testing the full tnc with mkfifo named pipes.

* Update test_tnc and tweak IRS/ISS.

* Correct test_modem to detect failures.

* Trying to be less dependent on env variables.

* Add IRS/ISS tests to ctests

* Pin codec2 revision to v1.0.3.

* Correcting git mistake.

* Pin codec2 revision to master.

This should be a specific release, that implements freedv_set_tuning_range.

Co-authored-by: DJ2LS <75909252+DJ2LS@users.noreply.github.com>
2022-05-21 23:04:17 +00:00

242 lines
8 KiB
Python
Executable file

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import argparse
import ctypes
import sys
import time
from typing import List
import numpy as np
import pyaudio
sys.path.insert(0, "..")
from tnc import codec2
def test_mm_rx():
# AUDIO PARAMETERS
AUDIO_FRAMES_PER_BUFFER = 2400 * 2
MODEM_SAMPLE_RATE = codec2.api.FREEDV_FS_8000
AUDIO_SAMPLE_RATE_RX = 48000
# make sure our resampler will work
assert (AUDIO_SAMPLE_RATE_RX / MODEM_SAMPLE_RATE) == codec2.api.FDMDV_OS_48
# SET COUNTERS
rx_bursts_datac = [0, 0, 0]
rx_frames_datac = [0, 0, 0]
rx_total_frames_datac = [0, 0, 0]
# time meassurement
time_end_datac = [0.0, 0.0, 0.0]
time_needed_datac = [0.0, 0.0, 0.0]
time_start_datac = [0.0, 0.0, 0.0]
datac_buffer: List[codec2.audio_buffer] = []
datac_bytes_out: List[ctypes.Array] = []
datac_bytes_per_frame = []
datac_freedv: List[ctypes.c_void_p] = []
args = parse_arguments()
if args.LIST:
p_audio = pyaudio.PyAudio()
for dev in range(p_audio.get_device_count()):
print("audiodev: ", dev, p_audio.get_device_info_by_index(dev)["name"])
sys.exit()
N_BURSTS = args.N_BURSTS
N_FRAMES_PER_BURST = args.N_FRAMES_PER_BURST
AUDIO_INPUT_DEVICE = args.AUDIO_INPUT_DEVICE
DEBUGGING_MODE = args.DEBUGGING_MODE
MAX_TIME = args.TIMEOUT
# open codec2 instances
for idx in range(3):
datac_freedv.append(
ctypes.cast(
codec2.api.freedv_open(codec2.api.FREEDV_MODE_DATAC0), ctypes.c_void_p
)
)
datac_bytes_per_frame.append(
int(codec2.api.freedv_get_bits_per_modem_frame(datac_freedv[idx]) / 8)
)
datac_bytes_out.append(ctypes.create_string_buffer(datac_bytes_per_frame[idx]))
codec2.api.freedv_set_frames_per_burst(datac_freedv[idx], N_FRAMES_PER_BURST)
datac_buffer.append(codec2.audio_buffer(2 * AUDIO_FRAMES_PER_BUFFER))
resampler = codec2.resampler()
# check if we want to use an audio device then do an pyaudio init
if AUDIO_INPUT_DEVICE != -1:
p_audio = pyaudio.PyAudio()
# auto search for loopback devices
if AUDIO_INPUT_DEVICE == -2:
loopback_list = [
dev
for dev in range(p_audio.get_device_count())
if "Loopback: PCM" in p_audio.get_device_info_by_index(dev)["name"]
]
if len(loopback_list) >= 2:
AUDIO_INPUT_DEVICE = loopback_list[0] # 0 = RX 1 = TX
print(f"loopback_list rx: {loopback_list}", file=sys.stderr)
else:
sys.exit()
print(
f"AUDIO INPUT DEVICE: {AUDIO_INPUT_DEVICE} "
f"DEVICE: {p_audio.get_device_info_by_index(AUDIO_INPUT_DEVICE)['name']} "
f"AUDIO SAMPLE RATE: {AUDIO_SAMPLE_RATE_RX}",
file=sys.stderr,
)
stream_rx = p_audio.open(
format=pyaudio.paInt16,
channels=1,
rate=AUDIO_SAMPLE_RATE_RX,
frames_per_buffer=AUDIO_FRAMES_PER_BUFFER,
input=True,
input_device_index=AUDIO_INPUT_DEVICE,
)
timeout = time.time() + MAX_TIME
print(time.time(), MAX_TIME, timeout)
receive = True
nread_exceptions = 0
# initial nin values
datac_nin = [0, 0, 0]
for idx in range(3):
datac_nin[idx] = codec2.api.freedv_nin(datac_freedv[idx])
def print_stats(time_datac0, time_datac1, time_datac3):
if not DEBUGGING_MODE:
return
time_datac = [time_datac0, time_datac1, time_datac3]
datac_rxstatus = ["", "", ""]
for idx in range(3):
datac_rxstatus[idx] = codec2.api.rx_sync_flags_to_text[
codec2.api.freedv_get_rx_status(datac_freedv[idx])
]
text_out = ""
for idx in range(3):
text_out += f"NIN{idx}: {datac_nin[idx]:5d} "
text_out += f"RX_STATUS{idx}: {datac_rxstatus[idx]:4s} "
text_out += f"TIME: {round(time_datac[idx], 4):.4f} | "
text_out = text_out.rstrip(" ").rstrip("|").rstrip(" ")
print(text_out, file=sys.stderr)
while receive and time.time() < timeout:
if AUDIO_INPUT_DEVICE != -1:
try:
data_in48k = stream_rx.read(
AUDIO_FRAMES_PER_BUFFER, exception_on_overflow=True
)
except OSError as err:
print(err, file=sys.stderr)
if "Input overflowed" in str(err):
nread_exceptions += 1
if "Stream closed" in str(err):
print("Ending....")
receive = False
else:
data_in48k = sys.stdin.buffer.read(AUDIO_FRAMES_PER_BUFFER * 2)
# insert samples in buffer
audio_buffer = np.frombuffer(data_in48k, dtype=np.int16)
if len(audio_buffer) != AUDIO_FRAMES_PER_BUFFER:
print("len(x)", len(audio_buffer))
receive = False
audio_buffer = resampler.resample48_to_8(audio_buffer)
for idx in range(3):
datac_buffer[idx].push(audio_buffer)
while datac_buffer[idx].nbuffer >= datac_nin[idx]:
# demodulate audio
time_start_datac[idx] = time.time()
nbytes = codec2.api.freedv_rawdatarx(
datac_freedv[idx],
datac_bytes_out[idx],
datac_buffer[idx].buffer.ctypes,
)
time_end_datac[idx] = time.time()
datac_buffer[idx].pop(datac_nin[idx])
datac_nin[idx] = codec2.api.freedv_nin(datac_freedv[idx])
if nbytes == datac_bytes_per_frame[idx]:
rx_total_frames_datac[idx] += 1
rx_frames_datac[idx] += 1
if rx_frames_datac[idx] == N_FRAMES_PER_BURST:
rx_frames_datac[idx] = 0
rx_bursts_datac[idx] += 1
time_needed_datac[idx] = time_end_datac[idx] - time_start_datac[idx]
print_stats(
time_needed_datac[0], time_needed_datac[1], time_needed_datac[2]
)
if (
rx_bursts_datac[0] == N_BURSTS
and rx_bursts_datac[1] == N_BURSTS
and rx_bursts_datac[2] == N_BURSTS
):
receive = False
if nread_exceptions:
print(
f"nread_exceptions {nread_exceptions:d} - receive audio lost! "
"Consider increasing Pyaudio frames_per_buffer...",
file=sys.stderr,
)
# INFO IF WE REACHED TIMEOUT
if time.time() > timeout:
print("TIMEOUT REACHED", file=sys.stderr)
print(
f"DATAC0: {rx_bursts_datac[0]}/{rx_total_frames_datac[0]} "
f"DATAC1: {rx_bursts_datac[1]}/{rx_total_frames_datac[1]} "
f"DATAC3: {rx_bursts_datac[2]}/{rx_total_frames_datac[2]}",
file=sys.stderr,
)
if AUDIO_INPUT_DEVICE != -1:
stream_rx.close()
p_audio.terminate()
def parse_arguments():
# --------------------------------------------GET PARAMETER INPUTS
parser = argparse.ArgumentParser(description="Simons TEST TNC")
parser.add_argument("--bursts", dest="N_BURSTS", default=1, type=int)
parser.add_argument(
"--framesperburst", dest="N_FRAMES_PER_BURST", default=1, type=int
)
parser.add_argument(
"--audiodev",
dest="AUDIO_INPUT_DEVICE",
default=-1,
type=int,
help="audio device number to use",
)
parser.add_argument("--debug", dest="DEBUGGING_MODE", action="store_true")
parser.add_argument(
"--list",
dest="LIST",
action="store_true",
help="list audio devices by number and exit",
)
parser.add_argument(
"--timeout",
dest="TIMEOUT",
default=10,
type=int,
help="Timeout (seconds) before test ends",
)
args, _ = parser.parse_known_args()
return args
if __name__ == "__main__":
test_mm_rx()