openwebrx-clone/csdr/chain/demodulator.py

66 lines
2.5 KiB
Python
Raw Normal View History

2021-07-19 17:04:14 +00:00
from csdr.chain import Chain
2021-07-24 22:05:48 +00:00
from pycsdr.modules import Shift, FirDecimate, Bandpass, Squelch, FractionalDecimator, Writer
from pycsdr.types import Format
2021-08-06 22:09:40 +00:00
from csdr.chain.digiham import Dmr
2021-07-19 17:04:14 +00:00
class DemodulatorChain(Chain):
def __init__(self, samp_rate: int, audioRate: int, shiftRate: float, demodulator: Chain):
2021-07-31 22:49:20 +00:00
self.demodulator = demodulator
2021-07-19 17:04:14 +00:00
self.shift = Shift(shiftRate)
decimation, fraction = self._getDecimation(samp_rate, audioRate)
if_samp_rate = samp_rate / decimation
transition = 0.15 * (if_samp_rate / float(samp_rate))
# set the cutoff on the fist decimation stage lower so that the resulting output
# is already prepared for the second (fractional) decimation stage.
# this spares us a second filter.
self.decimation = FirDecimate(decimation, transition, 0.5 * decimation / (samp_rate / audioRate))
2021-07-19 17:04:14 +00:00
2021-07-25 17:36:03 +00:00
bp_transition = 320.0 / audioRate
2021-07-19 17:48:18 +00:00
self.bandpass = Bandpass(transition=bp_transition, use_fft=True)
2021-07-19 17:04:14 +00:00
2021-07-24 22:17:27 +00:00
readings_per_second = 4
# s-meter readings are available every 1024 samples
# the reporting interval is measured in those 1024-sample blocks
2021-07-25 17:36:03 +00:00
self.squelch = Squelch(5, int(audioRate / (readings_per_second * 1024)))
2021-07-19 17:04:14 +00:00
workers = [self.shift, self.decimation]
2021-07-19 17:04:14 +00:00
if fraction != 1.0:
workers += [FractionalDecimator(Format.COMPLEX_FLOAT, fraction)]
workers += [self.bandpass, self.squelch, demodulator]
2021-07-19 17:04:14 +00:00
super().__init__(*workers)
def setShiftRate(self, rate: float):
self.shift.setRate(rate)
def setSquelchLevel(self, level: float):
self.squelch.setSquelchLevel(level)
def setBandpass(self, low_cut: float, high_cut: float):
self.bandpass.setBandpass(low_cut, high_cut)
2021-07-24 22:05:48 +00:00
def setPowerWriter(self, writer: Writer):
self.squelch.setPowerWriter(writer)
2021-07-31 22:49:20 +00:00
def setMetaWriter(self, writer: Writer):
self.demodulator.setMetaWriter(writer)
2021-08-06 22:09:40 +00:00
def setDmrFilter(self, filter: int) -> None:
if isinstance(self.demodulator, Dmr):
self.demodulator.setSlotFilter(filter)
2021-07-19 17:04:14 +00:00
def _getDecimation(self, input_rate, output_rate):
if output_rate <= 0:
raise ValueError("invalid output rate: {rate}".format(rate=output_rate))
decimation = 1
target_rate = output_rate
while input_rate / (decimation + 1) >= target_rate:
decimation += 1
fraction = float(input_rate / decimation) / output_rate
return decimation, fraction