tx code for 3.7

max-skip-test
Max 9 years ago
parent 40d3f82359
commit b25a33e4fa
  1. 16
      op25/gr-op25_repeater/tx/generate-tsbks.py
  2. 3
      op25/gr-op25_repeater/tx/hackrf-tx.sh
  3. 171
      op25/gr-op25_repeater/tx/op25_c4fm_mod.py
  4. 303
      op25/gr-op25_repeater/tx/op25_tx.py
  5. 2019
      op25/gr-op25_repeater/tx/p25craft.py

@ -0,0 +1,16 @@
#! /usr/bin/python
from p25craft import make_fakecc_tsdu
# should generate file p25.out
if __name__ == '__main__':
params = {
'wacn' : 0xbee00,
'system_id': 0x290,
'cc_freq': 925000000,
'vc_freq': 924900000,
'nac': 0x293,
'subsystem_id': 1,
'site_id': 1}
make_fakecc_tsdu(params)

@ -0,0 +1,3 @@
#! /bin/sh
./op25_tx.py --gains 'RF:0,IF:0' -n 2 -r -e -i --args 'hackrf' -f 925005000

@ -0,0 +1,171 @@
#
# Copyright 2005,2006,2007 Free Software Foundation, Inc.
#
# OP25 4-Level Modulator Block
# Copyright 2009, 2014 Max H. Parke KA1RBI
#
# coeffs for shaping and cosine filters from Eric Ramsey thesis
#
# 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 pre-modulation block.
"""
from gnuradio import gr, gru, eng_notation
from gnuradio.digital import modulation_utils
from gnuradio import filter, digital, blocks
from gnuradio.eng_option import eng_option
from optparse import OptionParser
import math
# default values (used in __init__ and add_options)
_def_output_sample_rate = 48000
_def_excess_bw = 0.2
_def_reverse = False
_def_verbose = False
_def_log = False
# /////////////////////////////////////////////////////////////////////////////
# modulator
# /////////////////////////////////////////////////////////////////////////////
class p25_mod_bf(gr.hier_block2):
def __init__(self,
output_sample_rate=_def_output_sample_rate,
excess_bw=_def_excess_bw,
reverse=_def_reverse,
verbose=_def_verbose,
log=_def_log):
"""
Hierarchical block for RRC-filtered P25 FM modulation.
The input is a dibit (P25 symbol) stream (char, not packed) and the
output is the float "C4FM" signal at baseband, suitable for application
to an FM modulator stage
Input is at the base symbol rate (4800), output sample rate is
typically either 32000 (USRP TX chain) or 48000 (sound card)
@param output_sample_rate: output sample rate
@type output_sample_rate: integer
@param excess_bw: Root-raised cosine filter excess bandwidth
@type excess_bw: float
@param reverse: reverse polarity flag
@type reverse: bool
@param verbose: Print information about modulator?
@type verbose: bool
@param debug: Print modulation data to files?
@type debug: bool
"""
gr.hier_block2.__init__(self, "p25_c4fm_mod_bf",
gr.io_signature(1, 1, gr.sizeof_char), # Input signature
gr.io_signature(1, 1, gr.sizeof_float)) # Output signature
input_sample_rate = 4800 # P25 baseband symbol rate
lcm = gru.lcm(input_sample_rate, output_sample_rate)
self._interp_factor = int(lcm // input_sample_rate)
self._decimation = int(lcm // output_sample_rate)
self._excess_bw = excess_bw
mod_map = [1.0/3.0, 1.0, -(1.0/3.0), -1.0]
self.C2S = digital.chunks_to_symbols_bf(mod_map)
if reverse:
self.polarity = blocks.multiply_const_ff(-1)
else:
self.polarity = blocks.multiply_const_ff( 1)
ntaps = 11 * self._interp_factor
rrc_taps = filter.firdes.root_raised_cosine(
self._interp_factor, # gain (since we're interpolating by sps)
lcm, # sampling rate
input_sample_rate, # symbol rate
self._excess_bw, # excess bandwidth (roll-off factor)
ntaps)
# rrc_coeffs work slightly differently: each input sample
# (from mod_map above) at 4800 rate, then 9 zeros are inserted
# to bring to 48000 rate, then this filter is applied:
# rrc_filter = gr.fir_filter_fff(1, rrc_coeffs)
# FIXME: how to insert the 9 zero samples using gr ?
# rrc_coeffs = [0, -0.003, -0.006, -0.009, -0.012, -0.014, -0.014, -0.013, -0.01, -0.006, 0, 0.007, 0.014, 0.02, 0.026, 0.029, 0.029, 0.027, 0.021, 0.012, 0, -0.013, -0.027, -0.039, -0.049, -0.054, -0.055, -0.049, -0.038, -0.021, 0, 0.024, 0.048, 0.071, 0.088, 0.098, 0.099, 0.09, 0.07, 0.039, 0, -0.045, -0.091, -0.134, -0.17, -0.193, -0.199, -0.184, -0.147, -0.085, 0, 0.105, 0.227, 0.36, 0.496, 0.629, 0.751, 0.854, 0.933, 0.983, 1, 0.983, 0.933, 0.854, 0.751, 0.629, 0.496, 0.36, 0.227, 0.105, 0, -0.085, -0.147, -0.184, -0.199, -0.193, -0.17, -0.134, -0.091, -0.045, 0, 0.039, 0.07, 0.09, 0.099, 0.098, 0.088, 0.071, 0.048, 0.024, 0, -0.021, -0.038, -0.049, -0.055, -0.054, -0.049, -0.039, -0.027, -0.013, 0, 0.012, 0.021, 0.027, 0.029, 0.029, 0.026, 0.02, 0.014, 0.007, 0, -0.006, -0.01, -0.013, -0.014, -0.014, -0.012, -0.009, -0.006, -0.003, 0]
self.rrc_filter = filter.interp_fir_filter_fff(self._interp_factor, rrc_taps)
# FM pre-emphasis filter
shaping_coeffs = [-0.018, 0.0347, 0.0164, -0.0064, -0.0344, -0.0522, -0.0398, 0.0099, 0.0798, 0.1311, 0.121, 0.0322, -0.113, -0.2499, -0.3007, -0.2137, -0.0043, 0.2825, 0.514, 0.604, 0.514, 0.2825, -0.0043, -0.2137, -0.3007, -0.2499, -0.113, 0.0322, 0.121, 0.1311, 0.0798, 0.0099, -0.0398, -0.0522, -0.0344, -0.0064, 0.0164, 0.0347, -0.018]
self.shaping_filter = filter.fir_filter_fff(1, shaping_coeffs)
if verbose:
self._print_verbage()
if log:
self._setup_logging()
self.connect(self, self.C2S, self.polarity, self.rrc_filter, self.shaping_filter)
if (self._decimation > 1):
self.decimator = filter.rational_resampler_fff(1, self._decimation)
self.connect(self.shaping_filter, self.decimator, self)
else:
self.connect(self.shaping_filter, self)
def _print_verbage(self):
print "\nModulator:"
print "RRS roll-off factor: %f" % self._excess_bw
print "interpolation: %d decimation: %d" %(self._interp_factor, self._decimation)
def _setup_logging(self):
print "Modulation logging turned on."
self.connect(self.C2S,
gr.file_sink(gr.sizeof_float, "tx_chunks2symbols.dat"))
self.connect(self.polarity,
gr.file_sink(gr.sizeof_float, "tx_polarity.dat"))
self.connect(self.rrc_filter,
gr.file_sink(gr.sizeof_float, "tx_rrc_filter.dat"))
self.connect(self.shaping_filter,
gr.file_sink(gr.sizeof_float, "tx_shaping_filter.dat"))
if (self._decimation > 1):
self.connect(self.decimator,
gr.file_sink(gr.sizeof_float, "tx_decimator.dat"))
def add_options(parser):
"""
Adds QPSK modulation-specific options to the standard parser
"""
parser.add_option("", "--excess-bw", type="float", default=_def_excess_bw,
help="set RRC excess bandwith factor [default=%default] (PSK)")
add_options=staticmethod(add_options)
def extract_kwargs_from_options(options):
"""
Given command line options, create dictionary suitable for passing to __init__
"""
return modulation_utils.extract_kwargs_from_options(dqpsk_mod.__init__,
('self',), options)
extract_kwargs_from_options=staticmethod(extract_kwargs_from_options)
#
# Add these to the mod/demod registry
#
modulation_utils.add_type_1_mod('op25_c4fm', p25_mod_bf)

@ -0,0 +1,303 @@
#!/usr/bin/env python
#
# Copyright 2005,2006,2007 Free Software Foundation, Inc.
#
# GNU Radio Multichannel APCO P25 Tx
# (c) Copyright 2009, 2014 Max H. Parke KA1RBI
#
# This file is part of GNU Radio and part of OP25
#
# This program 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.
#
"""
Transmit N simultaneous narrow band P25 C4FM signals.
They will be centered at the frequency specified on the command line,
and are spaced at 25kHz steps from there.
There are three main ways to run this program:
A. Combined audio capture, speech coding, and USRP transmission [default]
B. USRP transmission only, receives P25 symbol input via UDP channel [-p]
C. USRP transmission only, P25 symbol input taken from file(s) [-i]
By using method B, the coding and transmission functions can be
split over separate machines. Run the op25_remote_tx script prior to
running this script when using method B.
"""
from gnuradio import gr, eng_notation, blocks, digital
from gnuradio import audio, filter, analog
from gnuradio.eng_option import eng_option
from optparse import OptionParser
from usrpm import usrp_dbid
import math
import sys
import osmosdr
import op25_repeater
from gnuradio.wxgui import stdgui2, fftsink2
import wx
import op25_c4fm_mod
########################################################
# instantiate one transmit chain for each call
class file_pipeline(gr.hier_block2):
def __init__(self, lo_freq, if_rate, input_file):
gr.hier_block2.__init__(self, "file_pipeline",
gr.io_signature(0, 0, 0), # Input signature
gr.io_signature(1, 1, gr.sizeof_gr_complex)) # Output signature
fs = blocks.file_source(gr.sizeof_gr_complex, input_file, True)
agc = analog.feedforward_agc_cc(160, 1.0)
# Local oscillator
lo = analog.sig_source_c (if_rate, # sample rate
analog.GR_SIN_WAVE, # waveform type
lo_freq, #frequency
1.0, # amplitude
0) # DC Offset
mixer = blocks.multiply_cc ()
self.connect (fs, agc, (mixer, 0))
self.connect (lo, (mixer, 1))
self.connect (mixer, self)
class pipeline(gr.hier_block2):
def __init__(self, vocoder, lo_freq, audio_rate, if_rate):
gr.hier_block2.__init__(self, "pipeline",
gr.io_signature(0, 0, 0), # Input signature
gr.io_signature(1, 1, gr.sizeof_gr_complex)) # Output signature
c4fm = op25_c4fm_mod.p25_mod_bf(output_sample_rate=audio_rate,
log=False,
verbose=True)
interp_factor = if_rate / audio_rate
low_pass = 2.88e3
interp_taps = filter.firdes.low_pass(1.0, if_rate, low_pass, low_pass * 0.1, filter.firdes.WIN_HANN)
interpolator = filter.interp_fir_filter_fff (int(interp_factor), interp_taps)
max_dev = 12.5e3
k = 2 * math.pi * max_dev / if_rate
adjustment = 1.5 # adjust for proper c4fm deviation level
modulator = analog.frequency_modulator_fc (k * adjustment)
# Local oscillator
lo = analog.sig_source_c (if_rate, # sample rate
analog.GR_SIN_WAVE, # waveform type
lo_freq, #frequency
1.0, # amplitude
0) # DC Offset
mixer = blocks.multiply_cc ()
self.connect (vocoder, c4fm, interpolator, modulator, (mixer, 0))
self.connect (lo, (mixer, 1))
self.connect (mixer, self)
class fm_tx_block(stdgui2.std_top_block):
def __init__(self, frame, panel, vbox, argv):
MAX_CHANNELS = 7
stdgui2.std_top_block.__init__ (self, frame, panel, vbox, argv)
parser = OptionParser (option_class=eng_option)
parser.add_option("-T", "--tx-subdev-spec", type="subdev", default=None,
help="select USRP Tx side A or B")
parser.add_option("-e","--enable-fft", action="store_true", default=False,
help="enable spectrum plot (and use more CPU)")
parser.add_option("-f", "--freq", type="eng_float", default=None,
help="set Tx frequency to FREQ [required]", metavar="FREQ")
parser.add_option("-i","--file-input", action="store_true", default=False,
help="input from baseband-0.dat, baseband-1.dat ...")
parser.add_option("-g", "--audio-gain", type="eng_float", default=1.0,
help="input audio gain multiplier")
parser.add_option("-n", "--nchannels", type="int", default=1,
help="number of Tx channels [1,4]")
parser.add_option("-a", "--udp-addr", type="string", default="127.0.0.1",
help="UDP host IP address")
parser.add_option("--args", type="string", default="",
help="device args")
parser.add_option("--gains", type="string", default="",
help="gains")
parser.add_option("-p", "--udp-port", type="int", default=0,
help="UDP port number")
parser.add_option("-r","--repeat", action="store_true", default=False,
help="continuously replay input file")
parser.add_option("-S", "--stretch", type="int", default=0,
help="elastic buffer trigger value")
parser.add_option("-v","--verbose", action="store_true", default=False,
help="print out stats")
parser.add_option("-I", "--audio-input", type="string", default="", help="pcm input device name. E.g., hw:0,0 or /dev/dsp")
(options, args) = parser.parse_args ()
if len(args) != 0:
parser.print_help()
sys.exit(1)
if options.nchannels < 1 or options.nchannels > MAX_CHANNELS:
sys.stderr.write ("op25_tx: nchannels out of range. Must be in [1,%d]\n" % MAX_CHANNELS)
sys.exit(1)
if options.freq is None:
sys.stderr.write("op25_tx: must specify frequency with -f FREQ\n")
parser.print_help()
sys.exit(1)
# ----------------------------------------------------------------
# Set up constants and parameters
self.u = osmosdr.sink (options.args) # the USRP sink (consumes samples)
gain_names = self.u.get_gain_names()
for name in gain_names:
gain_range = self.u.get_gain_range(name)
print "gain: name: %s range: start %d stop %d step %d" % (name, gain_range[0].start(), gain_range[0].stop(), gain_range[0].step())
if options.gains:
for tuple in options.gains.split(","):
name, gain = tuple.split(":")
gain = int(gain)
print "setting gain %s to %d" % (name, gain)
self.u.set_gain(gain, name)
self.usrp_rate = 320000
print 'setting sample rate'
self.u.set_sample_rate(self.usrp_rate)
self.u.set_center_freq(int(options.freq))
#self.u.set_bandwidth(self.usrp_rate)
#self.u = blocks.file_sink(gr.sizeof_gr_complex, 'usrp-samp.dat')
#self.dac_rate = self.u.dac_rate() # 128 MS/s
#self.usrp_interp = 400
#self.u.set_interp_rate(self.usrp_interp)
#self.usrp_rate = self.dac_rate / self.usrp_interp # 320 kS/s
#self.sw_interp = 10
#self.audio_rate = self.usrp_rate / self.sw_interp # 32 kS/s
self.audio_rate = 32000
# if not self.set_freq(options.freq):
# freq_range = self.subdev.freq_range()
# print "Failed to set frequency to %s. Daughterboard supports %s to %s" % (
# eng_notation.num_to_str(options.freq),
# eng_notation.num_to_str(freq_range[0]),
# eng_notation.num_to_str(freq_range[1]))
# raise SystemExit
# self.subdev.set_enable(True) # enable transmitter
# instantiate vocoders
self.vocoders = []
if options.file_input:
i = 0
t = blocks.file_source(gr.sizeof_char, "baseband-%d.dat" % i, options.repeat)
self.vocoders.append(t)
elif options.udp_port > 0:
self.udp_sources = []
for i in range (options.nchannels):
t = gr.udp_source(1, options.udp_addr, options.udp_port + i, 216)
self.udp_sources.append(t)
arity = 2
t = gr.packed_to_unpacked_bb(arity, gr.GR_MSB_FIRST)
self.vocoders.append(t)
self.connect(self.udp_sources[i], self.vocoders[i])
if 1: # else:
input_audio_rate = 8000
#self.audio_input = audio.source(input_audio_rate, options.audio_input)
af = 1333
audio_input = analog.sig_source_s( input_audio_rate, analog.GR_SIN_WAVE, af, 15000)
t = op25_repeater.vocoder(True, # 0=Decode,True=Encode
options.verbose, # Verbose flag
options.stretch, # flex amount
"", # udp ip address
0, # udp port
False) # dump raw u vectors
self.connect(audio_input, t)
self.vocoders.append(t)
sum = blocks.add_cc ()
# Instantiate N NBFM channels
step = 100e3
offset = (0 * step, -1 * step, +1 * step, 2 * step, -2 * step, 3 * step, -3 * step)
for i in range (options.nchannels):
t = pipeline(self.vocoders[i], offset[i],
self.audio_rate, self.usrp_rate)
self.connect(t, (sum, i))
t = file_pipeline(offset[2], self.usrp_rate, '2013-320k-filt.dat')
self.connect(t, (sum, options.nchannels))
gain = blocks.multiply_const_cc (0.75 / (options.nchannels+1))
# connect it all
self.connect (sum, gain)
self.connect (gain, self.u)
# plot an FFT to verify we are sending what we want
if options.enable_fft:
post_mod = fftsink2.fft_sink_c(panel, title="Post Modulation",
fft_size=512, sample_rate=self.usrp_rate,
y_per_div=20, ref_level=40)
self.connect (sum, post_mod)
vbox.Add (post_mod.win, 1, wx.EXPAND)
#if options.debug:
# self.debugger = tx_debug_gui.tx_debug_gui(self.subdev)
# self.debugger.Show(True)
def set_freq(self, target_freq):
"""
Set the center frequency we're interested in.
@param target_freq: frequency in Hz
@rypte: bool
Tuning is a two step process. First we ask the front-end to
tune as close to the desired frequency as it can. Then we use
the result of that operation and our target_frequency to
determine the value for the digital up converter. Finally, we feed
any residual_freq to the s/w freq translater.
"""
r = self.u.tune(self.subdev.which(), self.subdev, target_freq)
if r:
print "r.baseband_freq =", eng_notation.num_to_str(r.baseband_freq)
print "r.dxc_freq =", eng_notation.num_to_str(r.dxc_freq)
print "r.residual_freq =", eng_notation.num_to_str(r.residual_freq)
print "r.inverted =", r.inverted
# Could use residual_freq in s/w freq translator
return True
return False
def main ():
sys.stderr.write("GNU Radio Multichannel APCO P25 Tx (c) Copyright 2009, KA1RBI\n")
app = stdgui2.stdapp(fm_tx_block, "Multichannel APCO P25 Tx", nstatus=1)
app.MainLoop ()
if __name__ == '__main__':
main ()

File diff suppressed because it is too large Load Diff
Loading…
Cancel
Save