Open Source implementation of APCO P25
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 
op25/op25/gr-op25_repeater/apps/p25_demodulator.py

449 lines
17 KiB

#
# Copyright 2005,2006,2007 Free Software Foundation, Inc.
#
# OP25 Demodulator Block
# Copyright 2009, 2010, 2011, 2012, 2013, 2014, 2015, 2016, 2017, 2018, 2019, 2020 Max H. Parke KA1RBI
#
# This file is part of GNU Radio and part of OP25
#
# This is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3, or (at your option)
# any later version.
#
# It is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this; see the file COPYING. If not, write to
# the Free Software Foundation, Inc., 51 Franklin Street,
# Boston, MA 02110-1301, USA.
#
"""
P25 C4FM/CQPSK demodulation block.
"""
import sys
from gnuradio import gr, eng_notation
from gnuradio import filter, analog, digital, blocks
from gnuradio.eng_option import eng_option
import op25
import op25_repeater
from math import pi
sys.path.append('tx')
import op25_c4fm_mod
# default values (used in __init__ and add_options)
_def_output_sample_rate = 48000
_def_if_rate = 24000
_def_gain_mu = 0.025
_def_costas_alpha = 0.04
_def_symbol_rate = 4800
_def_symbol_deviation = 600.0
_def_bb_gain = 1.0
_def_excess_bw = 0.2
_def_gmsk_mu = None
_def_mu = 0.5
_def_freq_error = 0.0
_def_omega_relative_limit = 0.005
# /////////////////////////////////////////////////////////////////////////////
# demodulator
# /////////////////////////////////////////////////////////////////////////////
def get_decim(speed):
s = int(speed)
if_freqs = [24000, 25000, 32000]
for i_f in if_freqs:
if s % i_f != 0:
continue
q = s // i_f
if q & 1:
continue
if q >= 40 and q & 3 == 0:
decim = q//4
decim2 = 4
else:
decim = q//2
decim2 = 2
return decim, decim2
return None
class p25_demod_base(gr.hier_block2):
def __init__(self,
if_rate = None,
filter_type = None,
excess_bw = _def_excess_bw,
symbol_rate = _def_symbol_rate):
"""
Hierarchical block for P25 demodulation base class
@param if_rate: sample rate of complex input channel
@type if_rate: int
"""
self.if_rate = if_rate
self.symbol_rate = symbol_rate
self.bb_sinks = []
self.baseband_amp = blocks.multiply_const_ff(_def_bb_gain)
coeffs = op25_c4fm_mod.c4fm_taps(sample_rate=self.if_rate, span=9, generator=op25_c4fm_mod.transfer_function_rx).generate()
sps = self.if_rate // self.symbol_rate
if filter_type == 'rrc':
ntaps = 7 * sps
if ntaps & 1 == 0:
ntaps += 1
coeffs = filter.firdes.root_raised_cosine(1.0, if_rate, symbol_rate, excess_bw, ntaps)
coeffs = [x*1.5 for x in coeffs]
if filter_type == 'nxdn':
coeffs = op25_c4fm_mod.c4fm_taps(sample_rate=self.if_rate, span=9, generator=op25_c4fm_mod.transfer_function_nxdn, symbol_rate=self.symbol_rate).generate()
gain_adj = 1.8 # for nxdn48 6.25 KHz
if self.symbol_rate == 4800:
gain_adj = 0.77 # nxdn96 12.5 KHz
coeffs = coeffs * gain_adj
if filter_type == 'gmsk':
# lifted from gmsk.py
_omega = sps
_gain_mu = _def_gmsk_mu
_mu = _def_mu
if not _gain_mu:
_gain_mu = 0.175
_gain_omega = .25 * _gain_mu * _gain_mu # critically damped
self.symbol_filter = blocks.multiply_const_ff(1.0)
self.fsk4_demod = digital.clock_recovery_mm_ff(_omega, _gain_omega,
_mu, _gain_mu,
_def_omega_relative_limit)
self.slicer = digital.binary_slicer_fb()
elif filter_type == 'fsk4mm':
self.symbol_filter = filter.fir_filter_fff(1, coeffs)
_omega = sps
_gain_mu = _def_gmsk_mu
_mu = _def_mu
if not _gain_mu:
_gain_mu = 0.0175
_gain_omega = .25 * _gain_mu * _gain_mu # critically damped
self.fsk4_demod = digital.clock_recovery_mm_ff(_omega, _gain_omega,
_mu, _gain_mu,
_def_omega_relative_limit)
levels = [ -2.0, 0.0, 2.0, 4.0 ]
self.slicer = op25_repeater.fsk4_slicer_fb(levels)
else:
self.symbol_filter = filter.fir_filter_fff(1, coeffs)
autotuneq = gr.msg_queue(2)
self.fsk4_demod = op25.fsk4_demod_ff(autotuneq, self.if_rate, self.symbol_rate)
levels = [ -2.0, 0.0, 2.0, 4.0 ]
self.slicer = op25_repeater.fsk4_slicer_fb(levels)
def set_symbol_rate(self, rate):
self.symbol_rate = rate
def set_baseband_gain(self, k):
self.baseband_amp.set_k(k)
def disconnect_bb(self):
# assumes lock held or init
if not len(self.bb_sinks):
return
for t in self.bb_sinks:
self.disconnect(t[0], t[1])
self.bb_sinks = []
def connect_bb(self, src, sink):
# assumes lock held or init
if src == 'symbol_filter':
b = self.symbol_filter
elif src == 'baseband_amp':
b = self.baseband_amp
self.connect(b, sink)
self.bb_sinks.append([b, sink])
class p25_demod_fb(p25_demod_base):
def __init__(self,
input_rate = None,
filter_type = None,
excess_bw = _def_excess_bw,
if_rate = _def_if_rate,
symbol_rate = _def_symbol_rate):
"""
Hierarchical block for P25 demodulation.
The float input is fsk4-demodulated
@param input_rate: sample rate of complex input channel
@type input_rate: int
"""
gr.hier_block2.__init__(self, "p25_demod_fb",
gr.io_signature(1, 1, gr.sizeof_float), # Input signature
gr.io_signature(1, 1, gr.sizeof_char)) # Output signature
p25_demod_base.__init__(self, if_rate=if_rate, symbol_rate=symbol_rate, filter_type=filter_type)
self.input_rate = input_rate
self.if_rate = if_rate
self.float_sink = None
if input_rate != if_rate:
assert if_rate < input_rate
assert input_rate % if_rate == 0 ### input rate must be multiple of if rate
decim = input_rate // if_rate
maxf = min(if_rate // 2, 6000) ### lpf cutoff at most 6 KHz
lpf_coeffs = filter.firdes.low_pass(1.0, input_rate, maxf, maxf//8, filter.firdes.WIN_HAMMING)
self.bb_decim = filter.fir_filter_fff(decim, lpf_coeffs)
self.connect(self, self.bb_decim, self.baseband_amp, self.symbol_filter, self.fsk4_demod, self.slicer, self)
else:
self.connect(self, self.baseband_amp, self.symbol_filter, self.fsk4_demod, self.slicer, self)
def disconnect_float(self):
# assumes lock held or init
if not self.float_sink:
return
self.disconnect(self.float_sink[0], self.float_sink[1])
self.float_sink = None
def connect_float(self, sink):
# assumes lock held or init
self.disconnect_float()
self.connect(self.fsk4_demod, sink)
self.float_sink = [self.fsk4_demod, sink]
class p25_demod_cb(p25_demod_base):
def __init__(self,
input_rate = None,
demod_type = 'cqpsk',
filter_type = None,
excess_bw = _def_excess_bw,
relative_freq = 0,
offset = 0,
if_rate = _def_if_rate,
gain_mu = _def_gain_mu,
costas_alpha = _def_costas_alpha,
symbol_rate = _def_symbol_rate,
use_old_decim = False):
"""
Hierarchical block for P25 demodulation.
The complex input is tuned, decimated and demodulated
@param input_rate: sample rate of complex input channel
@type input_rate: int
"""
gr.hier_block2.__init__(self, "p25_demod_cb",
gr.io_signature(1, 1, gr.sizeof_gr_complex), # Input signature
gr.io_signature(1, 1, gr.sizeof_char)) # Output signature
# gr.io_signature(0, 0, 0)) # Output signature
p25_demod_base.__init__(self, if_rate=if_rate, symbol_rate=symbol_rate, filter_type=filter_type)
self.input_rate = input_rate
self.if_rate = if_rate
self.symbol_rate = symbol_rate
self.connect_state = None
self.offset = 0
self.sps = 0.0
self.lo_freq = 0
self.float_sink = None
self.complex_sink = None
self.if1 = 0
self.if2 = 0
self.t_cache = {}
if filter_type == 'rrc':
self.set_baseband_gain(0.61)
self.mixer = blocks.multiply_cc()
if use_old_decim:
decimator_values = None # disable two stage decimator
else:
decimator_values = get_decim(input_rate)
if decimator_values:
self.decim, self.decim2 = decimator_values
self.if1 = input_rate / self.decim
self.if2 = self.if1 / self.decim2
sys.stderr.write( 'Using two-stage decimator for speed=%d, decim=%d/%d if1=%d if2=%d\n' % (input_rate, self.decim, self.decim2, self.if1, self.if2))
bpf_coeffs = filter.firdes.complex_band_pass(1.0, input_rate, -self.if1/2, self.if1/2, self.if1/2, filter.firdes.WIN_HAMMING)
self.t_cache[0] = bpf_coeffs
fa = 6250
fb = self.if2 / 2
if filter_type == 'nxdn' and self.symbol_rate == 2400: # nxdn48 6.25 KHz
fa = 3125
lpf_coeffs = filter.firdes.low_pass(1.0, self.if1, (fb+fa)/2, fb-fa, filter.firdes.WIN_HAMMING)
self.bpf = filter.fir_filter_ccc(self.decim, bpf_coeffs)
self.lpf = filter.fir_filter_ccf(self.decim2, lpf_coeffs)
resampled_rate = self.if2
self.relative_limit = (input_rate // 2) - (self.if1 // 2)
self.bfo = analog.sig_source_c (self.if1, analog.GR_SIN_WAVE, 0, 1.0, 0)
self.connect(self, self.bpf, (self.mixer, 0))
self.connect(self.bfo, (self.mixer, 1))
else:
sys.stderr.write( 'Unable to use two-stage decimator for speed=%d\n' % (input_rate))
# local osc
self.lo = analog.sig_source_c (input_rate, analog.GR_SIN_WAVE, 0, 1.0, 0)
f1 = 7250
f2 = 1450
if filter_type == 'nxdn' and self.symbol_rate == 2400: # nxdn48 6.25 KHz
f1 = 3125
f2 = 625
lpf_coeffs = filter.firdes.low_pass(1.0, input_rate, f1, f2, filter.firdes.WIN_HANN)
decimation = int(input_rate / if_rate)
self.lpf = filter.fir_filter_ccf(decimation, lpf_coeffs)
resampled_rate = float(input_rate) / float(decimation) # rate at output of self.lpf
self.relative_limit = (input_rate // 2) - f1
self.connect(self, (self.mixer, 0))
self.connect(self.lo, (self.mixer, 1))
self.connect(self.mixer, self.lpf)
if self.if_rate != resampled_rate:
self.if_out = filter.pfb.arb_resampler_ccf(float(self.if_rate) / resampled_rate)
self.connect(self.lpf, self.if_out)
else:
self.if_out = self.lpf
fa = 6250
fb = fa + 625
cutoff_coeffs = filter.firdes.low_pass(1.0, self.if_rate, (fb+fa)/2, fb-fa, filter.firdes.WIN_HANN)
self.cutoff = filter.fir_filter_ccf(1, cutoff_coeffs)
omega = float(self.if_rate) / float(self.symbol_rate)
gain_omega = 0.1 * gain_mu * gain_mu
alpha = costas_alpha
beta = 0.125 * alpha * alpha
fmax = 2400 # Hz
fmax = 2*pi * fmax / float(self.if_rate)
self.clock = op25_repeater.gardner_costas_cc(omega, gain_mu, gain_omega, alpha, beta, fmax, -fmax)
self.agc = analog.feedforward_agc_cc(16, 1.0)
# Perform Differential decoding on the constellation
self.diffdec = digital.diff_phasor_cc()
# take angle of the difference (in radians)
self.to_float = blocks.complex_to_arg()
# convert from radians such that signal is in -3/-1/+1/+3
self.rescale = blocks.multiply_const_ff( (1 / (pi / 4)) )
# fm demodulator (needed in fsk4 case)
fm_demod_gain = if_rate / (2.0 * pi * _def_symbol_deviation)
self.fm_demod = analog.quadrature_demod_cf(fm_demod_gain)
self.connect_chain(demod_type)
self.connect(self.slicer, self)
self.set_relative_frequency(relative_freq)
def get_error_band(self):
return int(self.clock.get_error_band())
def get_freq_error(self): # get error in Hz (approx).
return int(self.clock.get_freq_error() * self.symbol_rate)
def is_muted(self):
return self.clock.is_muted()
def set_muted(self, v):
self.clock.set_muted(v)
def set_omega(self, omega):
sps = self.if_rate / float(omega)
if sps == self.sps:
return
self.sps = sps
print ('set_omega %d %f' % (omega, sps))
self.clock.set_omega(self.sps)
def set_relative_frequency(self, freq):
N = 500 # Hz, we tune the bpf in discrete steps of size N
if abs(freq) > self.relative_limit:
#print 'set_relative_frequency: error, relative frequency %d exceeds limit %d' % (freq, self.input_rate/2)
return False
if freq == self.lo_freq:
return True
self.lo_freq = freq
if self.if1:
r = freq % N
bpf_freq = freq - r
if r > N//2:
bpf_freq += N # round to nearest N Hz
if bpf_freq not in self.t_cache.keys():
if abs(bpf_freq) > self.relative_limit:
sys.stderr.write( 'set_relative_frequency: error, relative frequency %d(%d) exceeds limit %d\n' % (bpf_freq, freq, self.relative_limit))
return False
self.t_cache[bpf_freq] = filter.firdes.complex_band_pass(1.0, self.input_rate, -bpf_freq - self.if1/2, -bpf_freq + self.if1/2, self.if1/2, filter.firdes.WIN_HAMMING)
self.bpf.set_taps(self.t_cache[bpf_freq])
bfo_f = self.decim * -freq / float(self.input_rate)
bfo_f -= int(bfo_f)
while bfo_f < -0.5:
bfo_f += 1.0
while bfo_f > 0.5:
bfo_f -= 1.0
self.bfo.set_frequency(-bfo_f * self.if1)
else:
self.lo.set_frequency(self.lo_freq)
return True
# assumes lock held or init
def disconnect_chain(self):
if self.connect_state == 'cqpsk':
self.disconnect(self.if_out, self.cutoff, self.agc, self.clock, self.diffdec, self.to_float, self.rescale, self.slicer)
elif self.connect_state == 'fsk4':
self.disconnect(self.if_out, self.cutoff, self.fm_demod, self.baseband_amp, self.symbol_filter, self.fsk4_demod, self.slicer)
self.connect_state = None
# assumes lock held or init
def connect_chain(self, demod_type):
if self.connect_state == demod_type:
return # already in desired state
self.disconnect_chain()
self.connect_state = demod_type
if demod_type == 'fsk4':
self.connect(self.if_out, self.cutoff, self.fm_demod, self.baseband_amp, self.symbol_filter, self.fsk4_demod, self.slicer)
elif demod_type == 'cqpsk':
self.connect(self.if_out, self.cutoff, self.agc, self.clock, self.diffdec, self.to_float, self.rescale, self.slicer)
else:
print ('connect_chain failed, type: %s' % demod_type)
assert 0 == 1
if self.float_sink is not None:
self.connect_float(self.float_sink[1])
def disconnect_float(self):
# assumes lock held or init
if not self.float_sink:
return
self.disconnect(self.float_sink[0], self.float_sink[1])
self.float_sink = None
def connect_float(self, sink):
# assumes lock held or init
self.disconnect_float()
if self.connect_state == 'cqpsk':
self.connect(self.rescale, sink)
self.float_sink = [self.rescale, sink]
elif self.connect_state == 'fsk4':
self.connect(self.fsk4_demod, sink)
self.float_sink = [self.fsk4_demod, sink]
else:
print ('connect_float: state error', self.connect_state)
assert 0 == 1
def connect_complex(self, src, sink):
# assumes lock held or init
if src == 'clock':
self.connect(self.clock, sink)
elif src == 'diffdec':
self.connect(self.diffdec, sink)
elif src == 'mixer':
self.connect(self.agc, sink)
elif src == 'src':
self.connect(self, sink)
elif src == 'bpf':
self.connect(self.bpf, sink)
elif src == 'if_out':
self.connect(self.if_out, sink)
elif src == 'agc':
self.connect(self.agc, sink)