Merge branch 'develop' into packet
This commit is contained in:
Jakob Ketterl
492
csdr.py
Executable file → Normal file
492
csdr.py
Executable file → Normal file
@ -21,33 +21,56 @@ OpenWebRX csdr plugin: do the signal processing with csdr
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
import time
|
||||
import os
|
||||
import signal
|
||||
import threading
|
||||
from functools import partial
|
||||
from owrx.wsjt import Ft8Chopper, WsprChopper, Jt9Chopper, Jt65Chopper, Ft4Chopper
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class output(object):
|
||||
def add_output(self, type, read_fn):
|
||||
pass
|
||||
def reset(self):
|
||||
def send_output(self, t, read_fn):
|
||||
if not self.supports_type(t):
|
||||
# TODO rewrite the output mechanism in a way that avoids producing unnecessary data
|
||||
logger.warning("dumping output of type %s since it is not supported.", t)
|
||||
threading.Thread(target=self.pump(read_fn, lambda x: None)).start()
|
||||
return
|
||||
self.receive_output(t, read_fn)
|
||||
|
||||
def receive_output(self, t, read_fn):
|
||||
pass
|
||||
|
||||
def pump(self, read, write):
|
||||
def copy():
|
||||
run = True
|
||||
while run:
|
||||
data = read()
|
||||
if data is None or (isinstance(data, bytes) and len(data) == 0):
|
||||
run = False
|
||||
else:
|
||||
write(data)
|
||||
|
||||
return copy
|
||||
|
||||
def supports_type(self, t):
|
||||
return True
|
||||
|
||||
|
||||
class dsp(object):
|
||||
|
||||
def __init__(self, output):
|
||||
self.samp_rate = 250000
|
||||
self.output_rate = 11025 #this is default, and cannot be set at the moment
|
||||
self.output_rate = 11025 # this is default, and cannot be set at the moment
|
||||
self.fft_size = 1024
|
||||
self.fft_fps = 5
|
||||
self.offset_freq = 0
|
||||
self.low_cut = -4000
|
||||
self.high_cut = 4000
|
||||
self.bpf_transition_bw = 320 #Hz, and this is a constant
|
||||
self.ddc_transition_bw_rate = 0.15 # of the IF sample rate
|
||||
self.bpf_transition_bw = 320 # Hz, and this is a constant
|
||||
self.ddc_transition_bw_rate = 0.15 # of the IF sample rate
|
||||
self.running = False
|
||||
self.secondary_processes_running = False
|
||||
self.audio_compression = "none"
|
||||
@ -67,105 +90,165 @@ class dsp(object):
|
||||
self.secondary_fft_size = 1024
|
||||
self.secondary_process_fft = None
|
||||
self.secondary_process_demod = None
|
||||
self.pipe_names=["bpf_pipe", "shift_pipe", "squelch_pipe", "smeter_pipe", "meta_pipe", "iqtee_pipe",
|
||||
"iqtee2_pipe", "dmr_control_pipe"]
|
||||
self.secondary_pipe_names=["secondary_shift_pipe"]
|
||||
self.pipe_names = [
|
||||
"bpf_pipe",
|
||||
"shift_pipe",
|
||||
"squelch_pipe",
|
||||
"smeter_pipe",
|
||||
"meta_pipe",
|
||||
"iqtee_pipe",
|
||||
"iqtee2_pipe",
|
||||
"dmr_control_pipe",
|
||||
]
|
||||
self.secondary_pipe_names = ["secondary_shift_pipe"]
|
||||
self.secondary_offset_freq = 1000
|
||||
self.unvoiced_quality = 1
|
||||
self.modification_lock = threading.Lock()
|
||||
self.output = output
|
||||
self.temporary_directory = "/tmp"
|
||||
|
||||
def chain(self,which):
|
||||
chain ="nc -v 127.0.0.1 {nc_port} | "
|
||||
if self.csdr_dynamic_bufsize: chain += "csdr setbuf {start_bufsize} | "
|
||||
if self.csdr_through: chain +="csdr through | "
|
||||
def set_temporary_directory(self, what):
|
||||
self.temporary_directory = what
|
||||
|
||||
def chain(self, which):
|
||||
chain = ["nc -v 127.0.0.1 {nc_port}"]
|
||||
if self.csdr_dynamic_bufsize:
|
||||
chain += ["csdr setbuf {start_bufsize}"]
|
||||
if self.csdr_through:
|
||||
chain += ["csdr through"]
|
||||
if which == "fft":
|
||||
chain += "csdr fft_cc {fft_size} {fft_block_size} | " + \
|
||||
("csdr logpower_cf -70 | " if self.fft_averages == 0 else "csdr logaveragepower_cf -70 {fft_size} {fft_averages} | ") + \
|
||||
"csdr fft_exchange_sides_ff {fft_size}"
|
||||
if self.fft_compression=="adpcm":
|
||||
chain += " | csdr compress_fft_adpcm_f_u8 {fft_size}"
|
||||
chain += [
|
||||
"csdr fft_cc {fft_size} {fft_block_size}",
|
||||
"csdr logpower_cf -70"
|
||||
if self.fft_averages == 0
|
||||
else "csdr logaveragepower_cf -70 {fft_size} {fft_averages}",
|
||||
"csdr fft_exchange_sides_ff {fft_size}",
|
||||
]
|
||||
if self.fft_compression == "adpcm":
|
||||
chain += ["csdr compress_fft_adpcm_f_u8 {fft_size}"]
|
||||
return chain
|
||||
chain += "csdr shift_addition_cc --fifo {shift_pipe} | "
|
||||
chain += "csdr fir_decimate_cc {decimation} {ddc_transition_bw} HAMMING | "
|
||||
chain += "csdr bandpass_fir_fft_cc --fifo {bpf_pipe} {bpf_transition_bw} HAMMING | csdr squelch_and_smeter_cc --fifo {squelch_pipe} --outfifo {smeter_pipe} 5 {smeter_report_every} | "
|
||||
chain += [
|
||||
"csdr shift_addition_cc --fifo {shift_pipe}",
|
||||
"csdr fir_decimate_cc {decimation} {ddc_transition_bw} HAMMING",
|
||||
"csdr bandpass_fir_fft_cc --fifo {bpf_pipe} {bpf_transition_bw} HAMMING",
|
||||
]
|
||||
if self.output.supports_type("smeter"):
|
||||
chain += [
|
||||
"csdr squelch_and_smeter_cc --fifo {squelch_pipe} --outfifo {smeter_pipe} 5 {smeter_report_every}"
|
||||
]
|
||||
if self.secondary_demodulator:
|
||||
chain += "csdr tee {iqtee_pipe} | "
|
||||
chain += "csdr tee {iqtee2_pipe} | "
|
||||
if self.output.supports_type("secondary_fft"):
|
||||
chain += ["csdr tee {iqtee_pipe}"]
|
||||
chain += ["csdr tee {iqtee2_pipe}"]
|
||||
# early exit if we don't want audio
|
||||
if not self.output.supports_type("audio"):
|
||||
return chain
|
||||
# safe some cpu cycles... no need to decimate if decimation factor is 1
|
||||
last_decimation_block = "csdr fractional_decimator_ff {last_decimation} | " if self.last_decimation != 1.0 else ""
|
||||
last_decimation_block = (
|
||||
["csdr fractional_decimator_ff {last_decimation}"] if self.last_decimation != 1.0 else []
|
||||
)
|
||||
if which == "nfm":
|
||||
chain += "csdr fmdemod_quadri_cf | csdr limit_ff | "
|
||||
chain += ["csdr fmdemod_quadri_cf", "csdr limit_ff"]
|
||||
chain += last_decimation_block
|
||||
chain += "csdr deemphasis_nfm_ff {output_rate} | csdr convert_f_s16"
|
||||
chain += ["csdr deemphasis_nfm_ff {output_rate}", "csdr convert_f_s16"]
|
||||
elif self.isDigitalVoice(which):
|
||||
chain += "csdr fmdemod_quadri_cf | dc_block | "
|
||||
chain += ["csdr fmdemod_quadri_cf", "dc_block "]
|
||||
chain += last_decimation_block
|
||||
# dsd modes
|
||||
if which in [ "dstar", "nxdn" ]:
|
||||
chain += "csdr limit_ff | csdr convert_f_s16 | "
|
||||
if which in ["dstar", "nxdn"]:
|
||||
chain += ["csdr limit_ff", "csdr convert_f_s16"]
|
||||
if which == "dstar":
|
||||
chain += "dsd -fd"
|
||||
chain += ["dsd -fd -i - -o - -u {unvoiced_quality} -g -1 "]
|
||||
elif which == "nxdn":
|
||||
chain += "dsd -fi"
|
||||
chain += " -i - -o - -u {unvoiced_quality} -g -1 | CSDR_FIXED_BUFSIZE=32 csdr convert_s16_f | "
|
||||
chain += ["dsd -fi -i - -o - -u {unvoiced_quality} -g -1 "]
|
||||
chain += ["CSDR_FIXED_BUFSIZE=32 csdr convert_s16_f"]
|
||||
max_gain = 5
|
||||
# digiham modes
|
||||
else:
|
||||
chain += "rrc_filter | gfsk_demodulator | "
|
||||
chain += ["rrc_filter", "gfsk_demodulator"]
|
||||
if which == "dmr":
|
||||
chain += "dmr_decoder --fifo {meta_pipe} --control-fifo {dmr_control_pipe} | mbe_synthesizer -f -u {unvoiced_quality} | "
|
||||
chain += [
|
||||
"dmr_decoder --fifo {meta_pipe} --control-fifo {dmr_control_pipe}",
|
||||
"mbe_synthesizer -f -u {unvoiced_quality}",
|
||||
]
|
||||
elif which == "ysf":
|
||||
chain += "ysf_decoder --fifo {meta_pipe} | mbe_synthesizer -y -f -u {unvoiced_quality} | "
|
||||
chain += ["ysf_decoder --fifo {meta_pipe}", "mbe_synthesizer -y -f -u {unvoiced_quality}"]
|
||||
max_gain = 0.0005
|
||||
chain += "digitalvoice_filter -f | "
|
||||
chain += "CSDR_FIXED_BUFSIZE=32 csdr agc_ff 160000 0.8 1 0.0000001 {max_gain} | ".format(max_gain=max_gain)
|
||||
chain += "sox -t raw -r 8000 -e floating-point -b 32 -c 1 --buffer 32 - -t raw -r {output_rate} -e signed-integer -b 16 -c 1 - "
|
||||
chain += [
|
||||
"digitalvoice_filter -f",
|
||||
"CSDR_FIXED_BUFSIZE=32 csdr agc_ff 160000 0.8 1 0.0000001 {max_gain}".format(max_gain=max_gain),
|
||||
"sox -t raw -r 8000 -e floating-point -b 32 -c 1 --buffer 32 - -t raw -r {output_rate} -e signed-integer -b 16 -c 1 - ",
|
||||
]
|
||||
elif which == "packet":
|
||||
chain += "csdr fmdemod_quadri_cf | "
|
||||
chain += ["csdr fmdemod_quadri_cf"]
|
||||
chain += last_decimation_block
|
||||
chain += "csdr convert_f_s16 | "
|
||||
chain += "direwolf -r {audio_rate} - 1>&2"
|
||||
chain += [
|
||||
"csdr convert_f_s16",
|
||||
"direwolf -r {audio_rate} - 1>&2"
|
||||
]
|
||||
elif which == "am":
|
||||
chain += "csdr amdemod_cf | csdr fastdcblock_ff | "
|
||||
chain += ["csdr amdemod_cf", "csdr fastdcblock_ff"]
|
||||
chain += last_decimation_block
|
||||
chain += "csdr agc_ff | csdr limit_ff | csdr convert_f_s16"
|
||||
chain += ["csdr agc_ff", "csdr limit_ff", "csdr convert_f_s16"]
|
||||
elif which == "ssb":
|
||||
chain += "csdr realpart_cf | "
|
||||
chain += ["csdr realpart_cf"]
|
||||
chain += last_decimation_block
|
||||
chain += "csdr agc_ff | csdr limit_ff | csdr convert_f_s16"
|
||||
chain += ["csdr agc_ff", "csdr limit_ff"]
|
||||
# fixed sample rate necessary for the wsjt-x tools. fix with sox...
|
||||
if self.isWsjtMode() and self.get_audio_rate() != self.get_output_rate():
|
||||
chain += [
|
||||
"sox -t raw -r {audio_rate} -e floating-point -b 32 -c 1 --buffer 32 - -t raw -r {output_rate} -e signed-integer -b 16 -c 1 - "
|
||||
]
|
||||
else:
|
||||
chain += ["csdr convert_f_s16"]
|
||||
|
||||
if self.audio_compression=="adpcm":
|
||||
chain += " | csdr encode_ima_adpcm_i16_u8"
|
||||
if self.audio_compression == "adpcm":
|
||||
chain += ["csdr encode_ima_adpcm_i16_u8"]
|
||||
return chain
|
||||
|
||||
def secondary_chain(self, which):
|
||||
secondary_chain_base="cat {input_pipe} | "
|
||||
secondary_chain_base = "cat {input_pipe} | "
|
||||
if which == "fft":
|
||||
return secondary_chain_base+"csdr realpart_cf | csdr fft_fc {secondary_fft_input_size} {secondary_fft_block_size} | csdr logpower_cf -70 " + (" | csdr compress_fft_adpcm_f_u8 {secondary_fft_size}" if self.fft_compression=="adpcm" else "")
|
||||
return (
|
||||
secondary_chain_base
|
||||
+ "csdr realpart_cf | csdr fft_fc {secondary_fft_input_size} {secondary_fft_block_size} | csdr logpower_cf -70 "
|
||||
+ (" | csdr compress_fft_adpcm_f_u8 {secondary_fft_size}" if self.fft_compression == "adpcm" else "")
|
||||
)
|
||||
elif which == "bpsk31":
|
||||
return secondary_chain_base + "csdr shift_addition_cc --fifo {secondary_shift_pipe} | " + \
|
||||
"csdr bandpass_fir_fft_cc -{secondary_bpf_cutoff} {secondary_bpf_cutoff} {secondary_bpf_cutoff} | " + \
|
||||
"csdr simple_agc_cc 0.001 0.5 | " + \
|
||||
"csdr timing_recovery_cc GARDNER {secondary_samples_per_bits} 0.5 2 --add_q | " + \
|
||||
"CSDR_FIXED_BUFSIZE=1 csdr dbpsk_decoder_c_u8 | " + \
|
||||
"CSDR_FIXED_BUFSIZE=1 csdr psk31_varicode_decoder_u8_u8"
|
||||
return (
|
||||
secondary_chain_base
|
||||
+ "csdr shift_addition_cc --fifo {secondary_shift_pipe} | "
|
||||
+ "csdr bandpass_fir_fft_cc -{secondary_bpf_cutoff} {secondary_bpf_cutoff} {secondary_bpf_cutoff} | "
|
||||
+ "csdr simple_agc_cc 0.001 0.5 | "
|
||||
+ "csdr timing_recovery_cc GARDNER {secondary_samples_per_bits} 0.5 2 --add_q | "
|
||||
+ "CSDR_FIXED_BUFSIZE=1 csdr dbpsk_decoder_c_u8 | "
|
||||
+ "CSDR_FIXED_BUFSIZE=1 csdr psk31_varicode_decoder_u8_u8"
|
||||
)
|
||||
elif self.isWsjtMode(which):
|
||||
chain = secondary_chain_base + "csdr realpart_cf | "
|
||||
if self.last_decimation != 1.0:
|
||||
chain += "csdr fractional_decimator_ff {last_decimation} | "
|
||||
chain += "csdr agc_ff | csdr limit_ff | csdr convert_f_s16"
|
||||
return chain
|
||||
|
||||
def set_secondary_demodulator(self, what):
|
||||
if self.get_secondary_demodulator() == what:
|
||||
return
|
||||
self.secondary_demodulator = what
|
||||
self.calculate_decimation()
|
||||
self.restart()
|
||||
|
||||
def secondary_fft_block_size(self):
|
||||
return (self.samp_rate/self.decimation)/(self.fft_fps*2) #*2 is there because we do FFT on real signal here
|
||||
return (self.samp_rate / self.decimation) / (
|
||||
self.fft_fps * 2
|
||||
) # *2 is there because we do FFT on real signal here
|
||||
|
||||
def secondary_decimation(self):
|
||||
return 1 #currently unused
|
||||
return 1 # currently unused
|
||||
|
||||
def secondary_bpf_cutoff(self):
|
||||
if self.secondary_demodulator == "bpsk31":
|
||||
return 31.25 / self.if_samp_rate()
|
||||
return 31.25 / self.if_samp_rate()
|
||||
return 0
|
||||
|
||||
def secondary_bpf_transition_bw(self):
|
||||
@ -175,7 +258,7 @@ class dsp(object):
|
||||
|
||||
def secondary_samples_per_bits(self):
|
||||
if self.secondary_demodulator == "bpsk31":
|
||||
return int(round(self.if_samp_rate()/31.25))&~3
|
||||
return int(round(self.if_samp_rate() / 31.25)) & ~3
|
||||
return 0
|
||||
|
||||
def secondary_bw(self):
|
||||
@ -183,55 +266,82 @@ class dsp(object):
|
||||
return 31.25
|
||||
|
||||
def start_secondary_demodulator(self):
|
||||
if not self.secondary_demodulator: return
|
||||
logger.debug("[openwebrx] starting secondary demodulator from IF input sampled at %d"%self.if_samp_rate())
|
||||
secondary_command_fft=self.secondary_chain("fft")
|
||||
secondary_command_demod=self.secondary_chain(self.secondary_demodulator)
|
||||
self.try_create_pipes(self.secondary_pipe_names, secondary_command_demod + secondary_command_fft)
|
||||
if not self.secondary_demodulator:
|
||||
return
|
||||
logger.debug("starting secondary demodulator from IF input sampled at %d" % self.if_samp_rate())
|
||||
secondary_command_demod = self.secondary_chain(self.secondary_demodulator)
|
||||
self.try_create_pipes(self.secondary_pipe_names, secondary_command_demod)
|
||||
|
||||
secondary_command_fft=secondary_command_fft.format(
|
||||
input_pipe=self.iqtee_pipe,
|
||||
secondary_fft_input_size=self.secondary_fft_size,
|
||||
secondary_fft_size=self.secondary_fft_size,
|
||||
secondary_fft_block_size=self.secondary_fft_block_size(),
|
||||
)
|
||||
secondary_command_demod=secondary_command_demod.format(
|
||||
secondary_command_demod = secondary_command_demod.format(
|
||||
input_pipe=self.iqtee2_pipe,
|
||||
secondary_shift_pipe=self.secondary_shift_pipe,
|
||||
secondary_decimation=self.secondary_decimation(),
|
||||
secondary_samples_per_bits=self.secondary_samples_per_bits(),
|
||||
secondary_bpf_cutoff=self.secondary_bpf_cutoff(),
|
||||
secondary_bpf_transition_bw=self.secondary_bpf_transition_bw(),
|
||||
if_samp_rate=self.if_samp_rate()
|
||||
if_samp_rate=self.if_samp_rate(),
|
||||
last_decimation=self.last_decimation,
|
||||
)
|
||||
|
||||
logger.debug("secondary command (demod) = %s", secondary_command_demod)
|
||||
my_env = os.environ.copy()
|
||||
# if self.csdr_dynamic_bufsize: my_env["CSDR_DYNAMIC_BUFSIZE_ON"]="1";
|
||||
if self.csdr_print_bufsizes:
|
||||
my_env["CSDR_PRINT_BUFSIZES"] = "1"
|
||||
if self.output.supports_type("secondary_fft"):
|
||||
secondary_command_fft = self.secondary_chain("fft")
|
||||
secondary_command_fft = secondary_command_fft.format(
|
||||
input_pipe=self.iqtee_pipe,
|
||||
secondary_fft_input_size=self.secondary_fft_size,
|
||||
secondary_fft_size=self.secondary_fft_size,
|
||||
secondary_fft_block_size=self.secondary_fft_block_size(),
|
||||
)
|
||||
logger.debug("secondary command (fft) = %s", secondary_command_fft)
|
||||
|
||||
self.secondary_process_fft = subprocess.Popen(
|
||||
secondary_command_fft, stdout=subprocess.PIPE, shell=True, preexec_fn=os.setpgrp, env=my_env
|
||||
)
|
||||
self.output.send_output(
|
||||
"secondary_fft",
|
||||
partial(self.secondary_process_fft.stdout.read, int(self.get_secondary_fft_bytes_to_read())),
|
||||
)
|
||||
|
||||
logger.debug("[openwebrx-dsp-plugin:csdr] secondary command (fft) = %s", secondary_command_fft)
|
||||
logger.debug("[openwebrx-dsp-plugin:csdr] secondary command (demod) = %s", secondary_command_demod)
|
||||
my_env=os.environ.copy()
|
||||
#if self.csdr_dynamic_bufsize: my_env["CSDR_DYNAMIC_BUFSIZE_ON"]="1";
|
||||
if self.csdr_print_bufsizes: my_env["CSDR_PRINT_BUFSIZES"]="1";
|
||||
self.secondary_process_fft = subprocess.Popen(secondary_command_fft, stdout=subprocess.PIPE, shell=True, preexec_fn=os.setpgrp, env=my_env)
|
||||
logger.debug("[openwebrx-dsp-plugin:csdr] Popen on secondary command (fft)")
|
||||
self.secondary_process_demod = subprocess.Popen(secondary_command_demod, stdout=subprocess.PIPE, shell=True, preexec_fn=os.setpgrp, env=my_env) #TODO digimodes
|
||||
logger.debug("[openwebrx-dsp-plugin:csdr] Popen on secondary command (demod)") #TODO digimodes
|
||||
self.secondary_process_demod = subprocess.Popen(
|
||||
secondary_command_demod, stdout=subprocess.PIPE, shell=True, preexec_fn=os.setpgrp, env=my_env
|
||||
)
|
||||
self.secondary_processes_running = True
|
||||
|
||||
self.output.add_output("secondary_fft", partial(self.secondary_process_fft.stdout.read, int(self.get_secondary_fft_bytes_to_read())))
|
||||
self.output.add_output("secondary_demod", partial(self.secondary_process_demod.stdout.read, 1))
|
||||
if self.isWsjtMode():
|
||||
smd = self.get_secondary_demodulator()
|
||||
if smd == "ft8":
|
||||
chopper = Ft8Chopper(self.secondary_process_demod.stdout)
|
||||
elif smd == "wspr":
|
||||
chopper = WsprChopper(self.secondary_process_demod.stdout)
|
||||
elif smd == "jt65":
|
||||
chopper = Jt65Chopper(self.secondary_process_demod.stdout)
|
||||
elif smd == "jt9":
|
||||
chopper = Jt9Chopper(self.secondary_process_demod.stdout)
|
||||
elif smd == "ft4":
|
||||
chopper = Ft4Chopper(self.secondary_process_demod.stdout)
|
||||
chopper.start()
|
||||
self.output.send_output("wsjt_demod", chopper.read)
|
||||
else:
|
||||
self.output.send_output("secondary_demod", partial(self.secondary_process_demod.stdout.read, 1))
|
||||
|
||||
#open control pipes for csdr and send initialization data
|
||||
if self.secondary_shift_pipe != None: #TODO digimodes
|
||||
self.secondary_shift_pipe_file=open(self.secondary_shift_pipe,"w") #TODO digimodes
|
||||
self.set_secondary_offset_freq(self.secondary_offset_freq) #TODO digimodes
|
||||
# open control pipes for csdr and send initialization data
|
||||
if self.secondary_shift_pipe != None: # TODO digimodes
|
||||
self.secondary_shift_pipe_file = open(self.secondary_shift_pipe, "w") # TODO digimodes
|
||||
self.set_secondary_offset_freq(self.secondary_offset_freq) # TODO digimodes
|
||||
|
||||
def set_secondary_offset_freq(self, value):
|
||||
self.secondary_offset_freq=value
|
||||
if self.secondary_processes_running:
|
||||
self.secondary_shift_pipe_file.write("%g\n"%(-float(self.secondary_offset_freq)/self.if_samp_rate()))
|
||||
self.secondary_offset_freq = value
|
||||
if self.secondary_processes_running and hasattr(self, "secondary_shift_pipe_file"):
|
||||
self.secondary_shift_pipe_file.write("%g\n" % (-float(self.secondary_offset_freq) / self.if_samp_rate()))
|
||||
self.secondary_shift_pipe_file.flush()
|
||||
|
||||
def stop_secondary_demodulator(self):
|
||||
if self.secondary_processes_running == False: return
|
||||
if self.secondary_processes_running == False:
|
||||
return
|
||||
self.try_delete_pipes(self.secondary_pipe_names)
|
||||
if self.secondary_process_fft:
|
||||
try:
|
||||
@ -250,42 +360,47 @@ class dsp(object):
|
||||
def get_secondary_demodulator(self):
|
||||
return self.secondary_demodulator
|
||||
|
||||
def set_secondary_fft_size(self,secondary_fft_size):
|
||||
#to change this, restart is required
|
||||
self.secondary_fft_size=secondary_fft_size
|
||||
def set_secondary_fft_size(self, secondary_fft_size):
|
||||
# to change this, restart is required
|
||||
self.secondary_fft_size = secondary_fft_size
|
||||
|
||||
def set_audio_compression(self,what):
|
||||
def set_audio_compression(self, what):
|
||||
self.audio_compression = what
|
||||
|
||||
def set_fft_compression(self,what):
|
||||
def set_fft_compression(self, what):
|
||||
self.fft_compression = what
|
||||
|
||||
def get_fft_bytes_to_read(self):
|
||||
if self.fft_compression=="none": return self.fft_size*4
|
||||
if self.fft_compression=="adpcm": return (self.fft_size/2)+(10/2)
|
||||
if self.fft_compression == "none":
|
||||
return self.fft_size * 4
|
||||
if self.fft_compression == "adpcm":
|
||||
return (self.fft_size / 2) + (10 / 2)
|
||||
|
||||
def get_secondary_fft_bytes_to_read(self):
|
||||
if self.fft_compression=="none": return self.secondary_fft_size*4
|
||||
if self.fft_compression=="adpcm": return (self.secondary_fft_size/2)+(10/2)
|
||||
if self.fft_compression == "none":
|
||||
return self.secondary_fft_size * 4
|
||||
if self.fft_compression == "adpcm":
|
||||
return (self.secondary_fft_size / 2) + (10 / 2)
|
||||
|
||||
def set_samp_rate(self,samp_rate):
|
||||
self.samp_rate=samp_rate
|
||||
def set_samp_rate(self, samp_rate):
|
||||
self.samp_rate = samp_rate
|
||||
self.calculate_decimation()
|
||||
if self.running: self.restart()
|
||||
if self.running:
|
||||
self.restart()
|
||||
|
||||
def calculate_decimation(self):
|
||||
(self.decimation, self.last_decimation, _) = self.get_decimation(self.samp_rate, self.get_audio_rate())
|
||||
|
||||
def get_decimation(self, input_rate, output_rate):
|
||||
decimation=1
|
||||
while input_rate / (decimation+1) >= output_rate:
|
||||
decimation = 1
|
||||
while input_rate / (decimation + 1) >= output_rate:
|
||||
decimation += 1
|
||||
fraction = float(input_rate / decimation) / output_rate
|
||||
intermediate_rate = input_rate / decimation
|
||||
return (decimation, fraction, intermediate_rate)
|
||||
|
||||
def if_samp_rate(self):
|
||||
return self.samp_rate/self.decimation
|
||||
return self.samp_rate / self.decimation
|
||||
|
||||
def get_name(self):
|
||||
return self.name
|
||||
@ -296,61 +411,73 @@ class dsp(object):
|
||||
def get_audio_rate(self):
|
||||
if self.isDigitalVoice() or self.isPacket():
|
||||
return 48000
|
||||
elif self.isWsjtMode():
|
||||
return 12000
|
||||
return self.get_output_rate()
|
||||
|
||||
def isDigitalVoice(self, demodulator = None):
|
||||
def isDigitalVoice(self, demodulator=None):
|
||||
if demodulator is None:
|
||||
demodulator = self.get_demodulator()
|
||||
return demodulator in ["dmr", "dstar", "nxdn", "ysf"]
|
||||
|
||||
def isWsjtMode(self, demodulator=None):
|
||||
if demodulator is None:
|
||||
demodulator = self.get_secondary_demodulator()
|
||||
return demodulator in ["ft8", "wspr", "jt65", "jt9", "ft4"]
|
||||
|
||||
def isPacket(self, demodulator = None):
|
||||
if demodulator is None:
|
||||
demodulator = self.get_demodulator()
|
||||
return demodulator == "packet"
|
||||
|
||||
def set_output_rate(self,output_rate):
|
||||
self.output_rate=output_rate
|
||||
def set_output_rate(self, output_rate):
|
||||
self.output_rate = output_rate
|
||||
self.calculate_decimation()
|
||||
|
||||
def set_demodulator(self,demodulator):
|
||||
if (self.demodulator == demodulator): return
|
||||
self.demodulator=demodulator
|
||||
def set_demodulator(self, demodulator):
|
||||
if self.demodulator == demodulator:
|
||||
return
|
||||
self.demodulator = demodulator
|
||||
self.calculate_decimation()
|
||||
self.restart()
|
||||
|
||||
def get_demodulator(self):
|
||||
return self.demodulator
|
||||
|
||||
def set_fft_size(self,fft_size):
|
||||
self.fft_size=fft_size
|
||||
def set_fft_size(self, fft_size):
|
||||
self.fft_size = fft_size
|
||||
self.restart()
|
||||
|
||||
def set_fft_fps(self,fft_fps):
|
||||
self.fft_fps=fft_fps
|
||||
def set_fft_fps(self, fft_fps):
|
||||
self.fft_fps = fft_fps
|
||||
self.restart()
|
||||
|
||||
def set_fft_averages(self,fft_averages):
|
||||
self.fft_averages=fft_averages
|
||||
def set_fft_averages(self, fft_averages):
|
||||
self.fft_averages = fft_averages
|
||||
self.restart()
|
||||
|
||||
def fft_block_size(self):
|
||||
if self.fft_averages == 0: return self.samp_rate/self.fft_fps
|
||||
else: return self.samp_rate/self.fft_fps/self.fft_averages
|
||||
if self.fft_averages == 0:
|
||||
return self.samp_rate / self.fft_fps
|
||||
else:
|
||||
return self.samp_rate / self.fft_fps / self.fft_averages
|
||||
|
||||
def set_offset_freq(self,offset_freq):
|
||||
self.offset_freq=offset_freq
|
||||
def set_offset_freq(self, offset_freq):
|
||||
self.offset_freq = offset_freq
|
||||
if self.running:
|
||||
self.modification_lock.acquire()
|
||||
self.shift_pipe_file.write("%g\n"%(-float(self.offset_freq)/self.samp_rate))
|
||||
self.shift_pipe_file.write("%g\n" % (-float(self.offset_freq) / self.samp_rate))
|
||||
self.shift_pipe_file.flush()
|
||||
self.modification_lock.release()
|
||||
|
||||
def set_bpf(self,low_cut,high_cut):
|
||||
self.low_cut=low_cut
|
||||
self.high_cut=high_cut
|
||||
def set_bpf(self, low_cut, high_cut):
|
||||
self.low_cut = low_cut
|
||||
self.high_cut = high_cut
|
||||
if self.running:
|
||||
self.modification_lock.acquire()
|
||||
self.bpf_pipe_file.write( "%g %g\n"%(float(self.low_cut)/self.if_samp_rate(), float(self.high_cut)/self.if_samp_rate()) )
|
||||
self.bpf_pipe_file.write(
|
||||
"%g %g\n" % (float(self.low_cut) / self.if_samp_rate(), float(self.high_cut) / self.if_samp_rate())
|
||||
)
|
||||
self.bpf_pipe_file.flush()
|
||||
self.modification_lock.release()
|
||||
|
||||
@ -358,12 +485,12 @@ class dsp(object):
|
||||
return [self.low_cut, self.high_cut]
|
||||
|
||||
def set_squelch_level(self, squelch_level):
|
||||
self.squelch_level=squelch_level
|
||||
#no squelch required on digital voice modes
|
||||
self.squelch_level = squelch_level
|
||||
# no squelch required on digital voice modes
|
||||
actual_squelch = 0 if self.isDigitalVoice() or self.isPacket() else self.squelch_level
|
||||
if self.running:
|
||||
self.modification_lock.acquire()
|
||||
self.squelch_pipe_file.write("%g\n"%(float(actual_squelch)))
|
||||
self.squelch_pipe_file.write("%g\n" % (float(actual_squelch)))
|
||||
self.squelch_pipe_file.flush()
|
||||
self.modification_lock.release()
|
||||
|
||||
@ -379,7 +506,7 @@ class dsp(object):
|
||||
self.dmr_control_pipe_file.write("{0}\n".format(filter))
|
||||
self.dmr_control_pipe_file.flush()
|
||||
|
||||
def mkfifo(self,path):
|
||||
def mkfifo(self, path):
|
||||
try:
|
||||
os.unlink(path)
|
||||
except:
|
||||
@ -387,64 +514,94 @@ class dsp(object):
|
||||
os.mkfifo(path)
|
||||
|
||||
def ddc_transition_bw(self):
|
||||
return self.ddc_transition_bw_rate*(self.if_samp_rate()/float(self.samp_rate))
|
||||
return self.ddc_transition_bw_rate * (self.if_samp_rate() / float(self.samp_rate))
|
||||
|
||||
def try_create_pipes(self, pipe_names, command_base):
|
||||
for pipe_name in pipe_names:
|
||||
if "{"+pipe_name+"}" in command_base:
|
||||
setattr(self, pipe_name, self.pipe_base_path+pipe_name)
|
||||
if "{" + pipe_name + "}" in command_base:
|
||||
setattr(self, pipe_name, self.pipe_base_path + pipe_name)
|
||||
self.mkfifo(getattr(self, pipe_name))
|
||||
else:
|
||||
setattr(self, pipe_name, None)
|
||||
|
||||
def try_delete_pipes(self, pipe_names):
|
||||
for pipe_name in pipe_names:
|
||||
pipe_path = getattr(self,pipe_name,None)
|
||||
pipe_path = getattr(self, pipe_name, None)
|
||||
if pipe_path:
|
||||
try: os.unlink(pipe_path)
|
||||
try:
|
||||
os.unlink(pipe_path)
|
||||
except FileNotFoundError:
|
||||
# it seems like we keep calling this twice. no idea why, but we don't need the resulting error.
|
||||
pass
|
||||
except Exception:
|
||||
logger.exception("try_delete_pipes()")
|
||||
|
||||
def start(self):
|
||||
self.modification_lock.acquire()
|
||||
if (self.running):
|
||||
if self.running:
|
||||
self.modification_lock.release()
|
||||
return
|
||||
self.running = True
|
||||
|
||||
command_base=self.chain(self.demodulator)
|
||||
command_base = " | ".join(self.chain(self.demodulator))
|
||||
|
||||
#create control pipes for csdr
|
||||
self.pipe_base_path="/tmp/openwebrx_pipe_{myid}_".format(myid=id(self))
|
||||
# create control pipes for csdr
|
||||
self.pipe_base_path = "{tmp_dir}/openwebrx_pipe_{myid}_".format(tmp_dir=self.temporary_directory, myid=id(self))
|
||||
|
||||
self.try_create_pipes(self.pipe_names, command_base)
|
||||
|
||||
#run the command
|
||||
command=command_base.format( bpf_pipe=self.bpf_pipe, shift_pipe=self.shift_pipe, decimation=self.decimation,
|
||||
last_decimation=self.last_decimation, fft_size=self.fft_size, fft_block_size=self.fft_block_size(), fft_averages=self.fft_averages,
|
||||
bpf_transition_bw=float(self.bpf_transition_bw)/self.if_samp_rate(), ddc_transition_bw=self.ddc_transition_bw(),
|
||||
flowcontrol=int(self.samp_rate*2), start_bufsize=self.base_bufsize*self.decimation, nc_port=self.nc_port,
|
||||
squelch_pipe=self.squelch_pipe, smeter_pipe=self.smeter_pipe, meta_pipe=self.meta_pipe, iqtee_pipe=self.iqtee_pipe, iqtee2_pipe=self.iqtee2_pipe,
|
||||
output_rate = self.get_output_rate(), smeter_report_every = int(self.if_samp_rate()/6000),
|
||||
unvoiced_quality = self.get_unvoiced_quality(), audio_rate = self.get_audio_rate(),
|
||||
dmr_control_pipe = self.dmr_control_pipe)
|
||||
# run the command
|
||||
command = command_base.format(
|
||||
bpf_pipe=self.bpf_pipe,
|
||||
shift_pipe=self.shift_pipe,
|
||||
decimation=self.decimation,
|
||||
last_decimation=self.last_decimation,
|
||||
fft_size=self.fft_size,
|
||||
fft_block_size=self.fft_block_size(),
|
||||
fft_averages=self.fft_averages,
|
||||
bpf_transition_bw=float(self.bpf_transition_bw) / self.if_samp_rate(),
|
||||
ddc_transition_bw=self.ddc_transition_bw(),
|
||||
flowcontrol=int(self.samp_rate * 2),
|
||||
start_bufsize=self.base_bufsize * self.decimation,
|
||||
nc_port=self.nc_port,
|
||||
squelch_pipe=self.squelch_pipe,
|
||||
smeter_pipe=self.smeter_pipe,
|
||||
meta_pipe=self.meta_pipe,
|
||||
iqtee_pipe=self.iqtee_pipe,
|
||||
iqtee2_pipe=self.iqtee2_pipe,
|
||||
output_rate=self.get_output_rate(),
|
||||
smeter_report_every=int(self.if_samp_rate() / 6000),
|
||||
unvoiced_quality=self.get_unvoiced_quality(),
|
||||
dmr_control_pipe=self.dmr_control_pipe,
|
||||
audio_rate=self.get_audio_rate(),
|
||||
)
|
||||
|
||||
logger.debug("[openwebrx-dsp-plugin:csdr] Command = %s", command)
|
||||
my_env=os.environ.copy()
|
||||
if self.csdr_dynamic_bufsize: my_env["CSDR_DYNAMIC_BUFSIZE_ON"]="1";
|
||||
if self.csdr_print_bufsizes: my_env["CSDR_PRINT_BUFSIZES"]="1";
|
||||
self.process = subprocess.Popen(command, stdout=subprocess.PIPE, shell=True, preexec_fn=os.setpgrp, env=my_env)
|
||||
logger.debug("Command = %s", command)
|
||||
my_env = os.environ.copy()
|
||||
if self.csdr_dynamic_bufsize:
|
||||
my_env["CSDR_DYNAMIC_BUFSIZE_ON"] = "1"
|
||||
if self.csdr_print_bufsizes:
|
||||
my_env["CSDR_PRINT_BUFSIZES"] = "1"
|
||||
|
||||
out = subprocess.PIPE if self.output.supports_type("audio") else subprocess.DEVNULL
|
||||
self.process = subprocess.Popen(command, stdout=out, shell=True, preexec_fn=os.setpgrp, env=my_env)
|
||||
|
||||
def watch_thread():
|
||||
rc = self.process.wait()
|
||||
logger.debug("dsp thread ended with rc=%d", rc)
|
||||
if (rc == 0 and self.running and not self.modification_lock.locked()):
|
||||
if rc == 0 and self.running and not self.modification_lock.locked():
|
||||
logger.debug("restarting since rc = 0, self.running = true, and no modification")
|
||||
self.restart()
|
||||
|
||||
threading.Thread(target = watch_thread).start()
|
||||
threading.Thread(target=watch_thread).start()
|
||||
|
||||
self.output.add_output("audio", partial(self.process.stdout.read, int(self.get_fft_bytes_to_read()) if self.demodulator == "fft" else 256))
|
||||
if self.output.supports_type("audio"):
|
||||
self.output.send_output(
|
||||
"audio",
|
||||
partial(
|
||||
self.process.stdout.read, int(self.get_fft_bytes_to_read()) if self.demodulator == "fft" else 256
|
||||
),
|
||||
)
|
||||
|
||||
# open control pipes for csdr
|
||||
if self.bpf_pipe:
|
||||
@ -465,24 +622,28 @@ class dsp(object):
|
||||
if self.bpf_pipe:
|
||||
self.set_bpf(self.low_cut, self.high_cut)
|
||||
if self.smeter_pipe:
|
||||
self.smeter_pipe_file=open(self.smeter_pipe,"r")
|
||||
self.smeter_pipe_file = open(self.smeter_pipe, "r")
|
||||
|
||||
def read_smeter():
|
||||
raw = self.smeter_pipe_file.readline()
|
||||
if len(raw) == 0:
|
||||
return None
|
||||
else:
|
||||
return float(raw.rstrip("\n"))
|
||||
self.output.add_output("smeter", read_smeter)
|
||||
|
||||
self.output.send_output("smeter", read_smeter)
|
||||
if self.meta_pipe != None:
|
||||
# TODO make digiham output unicode and then change this here
|
||||
self.meta_pipe_file=open(self.meta_pipe, "r", encoding="cp437")
|
||||
self.meta_pipe_file = open(self.meta_pipe, "r", encoding="cp437")
|
||||
|
||||
def read_meta():
|
||||
raw = self.meta_pipe_file.readline()
|
||||
if len(raw) == 0:
|
||||
return None
|
||||
else:
|
||||
return raw.rstrip("\n")
|
||||
self.output.add_output("meta", read_meta)
|
||||
|
||||
self.output.send_output("meta", read_meta)
|
||||
|
||||
if self.dmr_control_pipe:
|
||||
self.dmr_control_pipe_file = open(self.dmr_control_pipe, "w")
|
||||
@ -503,10 +664,11 @@ class dsp(object):
|
||||
self.modification_lock.release()
|
||||
|
||||
def restart(self):
|
||||
if not self.running: return
|
||||
if not self.running:
|
||||
return
|
||||
self.stop()
|
||||
self.start()
|
||||
|
||||
def __del__(self):
|
||||
self.stop()
|
||||
del(self.process)
|
||||
del self.process
|
||||
|
||||
Reference in New Issue
Block a user