Crash_Override/FreeDATA
1
0
Fork 0
mirror of https://github.com/DJ2LS/FreeDATA synced 2024-05-14 08:04:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

moved simple test from pyaudio to sounddevice

Browse source 
thisis just a test as I'm not happy with the overall way we are dong tests. This has been great during first steps with the tnc ( virtual audio devices ) but now we should to a more reliable way with named pipes for example
This commit is contained in:
dj2ls 2022-04-30 12:27:14 +02:00
parent e4b901adad
commit ea133f054d
3 changed files with 80 additions and 65 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
gui/main.js
View file

@ -510,7 +510,7 @@ ipcMain.on('save-file-to-folder',(event,data)=>{
//restart and install udpate
ipcMain.on('request-restart-and-install',(event,data)=>{
close_sub_binaries()
close_sub_processes()
autoUpdater.quitAndInstall();
});
@ -581,8 +581,8 @@ autoUpdater.on('error', (error) => {
function close_sub_binaries(){
mainLog.warn('closing sub binaries');
function close_sub_processes(){
mainLog.warn('closing sub processes');
// closing the tnc binary if not closed when closing application and also our daemon which has been started by the gui
try {
@ -632,7 +632,7 @@ function close_all() {
// function for closing the application with closing all used processes
close_sub_binaries();
close_sub_processes();
mainLog.warn('quitting app');

76
test/test_rx.py
View file

@ -9,7 +9,7 @@ Created on Wed Dec 23 07:04:24 2020
import ctypes
from ctypes import *
import pathlib
import pyaudio
import sounddevice as sd
import sys
import logging
import time
@ -35,9 +35,13 @@ parser.add_argument('--list', dest="LIST", action="store_true", help="list audio
args = parser.parse_args()
if args.LIST:
p = pyaudio.PyAudio()
for dev in range(0,p.get_device_count()):
print("audiodev: ", dev, p.get_device_info_by_index(dev)["name"])
devices = sd.query_devices(device=None, kind=None)
index = 0
for device in devices:
print(f"{index} {device['name']}")
index += 1
sd._terminate()
quit()
N_BURSTS = args.N_BURSTS
@ -57,30 +61,34 @@ assert (AUDIO_SAMPLE_RATE_RX / MODEM_SAMPLE_RATE) == codec2.api.FDMDV_OS_48
# check if we want to use an audio device then do an pyaudio init
if AUDIO_INPUT_DEVICE != -1:
p = pyaudio.PyAudio()
# auto search for loopback devices
if AUDIO_INPUT_DEVICE == -2:
loopback_list = []
for dev in range(0,p.get_device_count()):
if 'Loopback: PCM' in p.get_device_info_by_index(dev)["name"]:
loopback_list.append(dev)
if len(loopback_list) >= 2:
devices = sd.query_devices(device=None, kind=None)
index = 0
for device in devices:
if 'Loopback: PCM' in device['name']:
print(index)
loopback_list.append(index)
index += 1
if len(loopback_list) >= 1:
AUDIO_INPUT_DEVICE = loopback_list[0] #0 = RX 1 = TX
print(f"loopback_list rx: {loopback_list}", file=sys.stderr)
print(f"loopback_list tx: {loopback_list}", file=sys.stderr)
else:
quit()
print(f"AUDIO INPUT DEVICE: {AUDIO_INPUT_DEVICE} DEVICE: {p.get_device_info_by_index(AUDIO_INPUT_DEVICE)['name']} \
AUDIO SAMPLE RATE: {AUDIO_SAMPLE_RATE_RX}", file=sys.stderr)
stream_rx = p.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
)
print("not enough audio loopback devices ready...")
print("you should wait about 30 seconds...")
sd._terminate()
quit()
print(f"AUDIO INPUT DEVICE: {AUDIO_INPUT_DEVICE}", file=sys.stderr)
# audio stream init
stream_rx = sd.RawStream(channels=1, dtype='int16', device=AUDIO_INPUT_DEVICE, samplerate = AUDIO_SAMPLE_RATE_RX, blocksize=4800)
stream_rx.start()
# ----------------------------------------------------------------
# DATA CHANNEL INITIALISATION
@ -122,19 +130,22 @@ nin = codec2.api.freedv_nin(freedv)
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)
#data_in48k = stream_rx.read(AUDIO_FRAMES_PER_BUFFER, exception_on_overflow = True)
data_in48k, overflowed = stream_rx.read(AUDIO_FRAMES_PER_BUFFER)
except OSError as err:
print(err, file=sys.stderr)
if str(err).find("Input overflowed") != -1:
nread_exceptions += 1
if str(err).find("Stream closed") != -1:
print("Ending...")
receive = False
#if str(err).find("Input overflowed") != -1:
# nread_exceptions += 1
#if str(err).find("Stream closed") != -1:
# print("Ending...")
# receive = False
else:
data_in48k = sys.stdin.buffer.read(AUDIO_FRAMES_PER_BUFFER*2)
# insert samples in buffer
x = np.frombuffer(data_in48k, dtype=np.int16)
#print(x)
#x = data_in48k
x.tofile(frx)
if len(x) != AUDIO_FRAMES_PER_BUFFER:
receive = False
@ -187,8 +198,9 @@ if nread_exceptions:
print(f"RECEIVED BURSTS: {rx_bursts} RECEIVED FRAMES: {rx_total_frames} RX_ERRORS: {rx_errors}", file=sys.stderr)
frx.close()
# and at last check if we had an opened pyaudio instance and close it
if AUDIO_INPUT_DEVICE != -1:
stream_rx.close()
p.terminate()
# and at last check if we had an opened audio instance and close it
if AUDIO_INPUT_DEVICE != -1:
sd._terminate()

61
test/test_tx.py
View file

@ -5,7 +5,7 @@
import ctypes
from ctypes import *
import pathlib
import pyaudio
import sounddevice as sd
import time
import argparse
import sys
@ -29,9 +29,13 @@ parser.add_argument('--testframes', dest="TESTFRAMES", action="store_true", defa
args = parser.parse_args()
if args.LIST:
p = pyaudio.PyAudio()
for dev in range(0,p.get_device_count()):
print("audiodev: ", dev, p.get_device_info_by_index(dev)["name"])
devices = sd.query_devices(device=None, kind=None)
index = 0
for device in devices:
print(f"{index} {device['name']}")
index += 1
sd._terminate()
quit()
N_BURSTS = args.N_BURSTS
@ -49,31 +53,31 @@ assert (AUDIO_SAMPLE_RATE_TX % MODEM_SAMPLE_RATE) == 0
# check if we want to use an audio device then do an pyaudio init
if AUDIO_OUTPUT_DEVICE != -1:
p = pyaudio.PyAudio()
# auto search for loopback devices
if AUDIO_OUTPUT_DEVICE == -2:
loopback_list = []
for dev in range(0,p.get_device_count()):
if 'Loopback: PCM' in p.get_device_info_by_index(dev)["name"]:
loopback_list.append(dev)
if len(loopback_list) >= 2:
AUDIO_OUTPUT_DEVICE = loopback_list[1] #0 = RX 1 = TX
devices = sd.query_devices(device=None, kind=None)
index = 0
for device in devices:
if 'Loopback: PCM' in device['name']:
print(index)
loopback_list.append(index)
index += 1
if len(loopback_list) >= 1:
AUDIO_OUTPUT_DEVICE = loopback_list[len(loopback_list)-1] #0 = RX 1 = TX
print(f"loopback_list tx: {loopback_list}", file=sys.stderr)
else:
print("not enough audio loopback devices ready...")
print("you should wait about 30 seconds...")
sd._terminate()
quit()
print(f"AUDIO OUTPUT DEVICE: {AUDIO_OUTPUT_DEVICE} DEVICE: {p.get_device_info_by_index(AUDIO_OUTPUT_DEVICE)['name']} \
AUDIO SAMPLE RATE: {AUDIO_SAMPLE_RATE_TX}", file=sys.stderr)
print(f"AUDIO OUTPUT DEVICE: {AUDIO_OUTPUT_DEVICE}", file=sys.stderr)
# pyaudio init
stream_tx = p.open(format=pyaudio.paInt16,
channels=1,
rate=AUDIO_SAMPLE_RATE_TX,
frames_per_buffer=AUDIO_FRAMES_PER_BUFFER, #n_nom_modem_samples
output=True,
output_device_index=AUDIO_OUTPUT_DEVICE
)
# audio stream init
stream_tx = sd.RawStream(channels=1, dtype='int16', device=(0, AUDIO_OUTPUT_DEVICE), samplerate = AUDIO_SAMPLE_RATE_TX, blocksize=4800)
resampler = codec2.resampler()
# data binary string
@ -155,18 +159,17 @@ for i in range(1,N_BURSTS+1):
# check if we want to use an audio device or stdout
if AUDIO_OUTPUT_DEVICE != -1:
# Gotcha: we have to convert from np.int16 to Python "bytes"
stream_tx.write(txbuffer_48k.tobytes())
stream_tx.start()
stream_tx.write(txbuffer_48k)
else:
# print data to terminal for piping the output to other programs
sys.stdout.buffer.write(txbuffer_48k)
sys.stdout.flush()
# and at last check if we had an opened pyaudio instance and close it
# and at last check if we had an opened audio instance and close it
if AUDIO_OUTPUT_DEVICE != -1:
time.sleep(stream_tx.get_output_latency())
stream_tx.stop_stream()
stream_tx.close()
p.terminate()
sd._terminate()