#
# Copyright 2005,2006 Free Software Foundation, Inc.
#
# This file is part of GNU Radio
#
# GNU Radio 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 2, or (at your option)
# any later version.
#
# GNU Radio 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 GNU Radio; see the file COPYING. If not, write to
# the Free Software Foundation, Inc., 51 Franklin Street,
# Boston, MA 02110-1301, USA.
#
# See gnuradio-examples/python/gmsk2 for examples
"""
differential QPSK modulation and demodulation.
"""
from gnuradio import gr, gru
from math import pi, sqrt
import cmath
import Numeric
from pprint import pprint
_use_gray_code = True
def make_constellation(m):
return [cmath.exp(i * 2 * pi / m * 1j) for i in range(m)]
# Common definition of constellations for Tx and Rx
constellation = {
2 : make_constellation(2), # BPSK
4 : make_constellation(4), # QPSK
8 : make_constellation(8) # 8PSK
}
if 0:
print "const(2) ="
pprint(constellation[2])
print "const(4) ="
pprint(constellation[4])
print "const(8) ="
pprint(constellation[8])
if _use_gray_code:
# -----------------------
# Do Gray code
# -----------------------
# binary to gray coding
binary_to_gray = {
2 : (0, 1),
4 : (0, 1, 3, 2),
8 : (0, 1, 3, 2, 7, 6, 4, 5)
}
# gray to binary
gray_to_binary = {
2 : (0, 1),
4 : (0, 1, 3, 2),
8 : (0, 1, 3, 2, 6, 7, 5, 4)
}
else:
# -----------------------
# Don't Gray code
# -----------------------
# identity mapping
binary_to_gray = {
2 : (0, 1),
4 : (0, 1, 2, 3),
8 : (0, 1, 2, 3, 4, 5, 6, 7)
}
# identity mapping
gray_to_binary = {
2 : (0, 1),
4 : (0, 1, 2, 3),
8 : (0, 1, 2, 3, 4, 5, 6, 7)
}
# /////////////////////////////////////////////////////////////////////////////
# QPSK mod/demod with steams of bytes as data i/o
# /////////////////////////////////////////////////////////////////////////////
class dqpsk_mod(gr.hier_block):
def __init__(self, fg, spb, excess_bw):
"""
Hierarchical block for RRC-filtered QPSK modulation.
The input is a byte stream (unsigned char) and the
output is the complex modulated signal at baseband.
@param fg: flow graph
@type fg: flow graph
@param spb: samples per baud >= 2
@type spb: integer
@param excess_bw: Root-raised cosine filter excess bandwidth
@type excess_bw: float
"""
if not isinstance(spb, int) or spb < 2:
raise TypeError, "sbp must be an integer >= 2"
self.spb = spb
ntaps = 11 * spb
bits_per_symbol = self.bits_per_baud()
arity = pow(2,bits_per_symbol)
print "bits_per_symbol =", bits_per_symbol
# turn bytes into k-bit vectors
self.bytes2chunks = \
gr.packed_to_unpacked_bb(bits_per_symbol, gr.GR_MSB_FIRST)
if True:
self.gray_coder = gr.map_bb(binary_to_gray[arity])
else:
self.gray_coder = None
self.diffenc = gr.diff_encoder_bb(arity)
self.chunks2symbols = gr.chunks_to_symbols_bc(constellation[arity])
# pulse shaping filter
self.rrc_taps = gr.firdes.root_raised_cosine(
spb, # gain (spb since we're interpolating by spb)
spb, # sampling rate
1.0, # symbol rate
excess_bw, # excess bandwidth (roll-off factor)
ntaps)
self.rrc_filter = gr.interp_fir_filter_ccf(spb, self.rrc_taps)
# Connect
if self.gray_coder:
fg.connect(self.bytes2chunks, self.gray_coder)
t = self.gray_coder
else:
t = self.bytes2chunks
fg.connect(t, self.diffenc, self.chunks2symbols, self.rrc_filter)
if 1:
fg.connect(self.gray_coder,
gr.file_sink(gr.sizeof_char, "graycoder.dat"))
fg.connect(self.diffenc,
gr.file_sink(gr.sizeof_char, "diffenc.dat"))
# Initialize base class
gr.hier_block.__init__(self, fg, self.bytes2chunks, self.rrc_filter)
def samples_per_baud(self):
return self.spb
def bits_per_baud(self=None): # staticmethod that's also callable on an instance
return 2
bits_per_baud = staticmethod(bits_per_baud) # make it a static method. RTFM
class dqpsk_demod__coherent_detection_of_differentially_encoded_psk(gr.hier_block):
def __init__(self, fg, spb, excess_bw, costas_alpha=0.005, gain_mu=0.05):
"""
Hierarchical block for RRC-filtered QPSK demodulation
The input is the complex modulated signal at baseband.
The output is a stream of bits packed 1 bit per byte (LSB)
@param fg: flow graph
@type fg: flow graph
@param spb: samples per baud >= 2
@type spb: float
@param excess_bw: Root-raised cosine filter excess bandwidth
@type excess_bw: float
@param costas_alpha: loop filter gain
@type costas_alphas: float
@param gain_mu:
@type gain_mu: float
"""
if spb < 2:
raise TypeError, "sbp must be >= 2"
self.spb = spb
bits_per_symbol = self.bits_per_baud()
arity = pow(2,bits_per_symbol)
print "bits_per_symbol =", bits_per_symbol
# Automatic gain control
self.preamp = gr.multiply_const_cc(10e-5)
self.agc = gr.agc_cc(1e-3, 1, 1)
# Costas loop (carrier tracking)
# FIXME: need to decide how to handle this more generally; do we pull it from higher layer?
costas_order = 4
beta = .25 * costas_alpha * costas_alpha
self.costas_loop = gr.costas_loop_cc(costas_alpha, beta, 0.05, -0.05, costas_order)
# RRC data filter
ntaps = 11 * spb
self.rrc_taps = gr.firdes.root_raised_cosine(
1.0, # gain
spb, # sampling rate
1.0, # symbol rate
excess_bw, # excess bandwidth (roll-off factor)
ntaps)
self.rrc_filter=gr.fir_filter_ccf(1, self.rrc_taps)
# symbol clock recovery
omega = spb
gain_omega = .25 * gain_mu * gain_mu
omega_rel_limit = 0.5
mu = 0.05
gain_mu = 0.1
self.clock_recovery=gr.clock_recovery_mm_cc(omega, gain_omega,
mu, gain_mu, omega_rel_limit)
# find closest constellation point
#rot = .707 + .707j
rot = 1
rotated_const = map(lambda pt: pt * rot, constellation[arity])
print "rotated_const =", rotated_const
self.diffdec = gr.diff_phasor_cc()
#self.diffdec = gr.diff_decoder_bb(arity)
self.slicer = gr.constellation_decoder_cb(rotated_const, range(arity))
self.gray_decoder = gr.map_bb(gray_to_binary[arity])
# unpack the k bit vector into a stream of bits
self.unpack = gr.unpack_k_bits_bb(bits_per_symbol)
fg.connect(self.preamp, self.agc, self.costas_loop, self.rrc_filter, self.clock_recovery,
self.diffdec, self.slicer, self.gray_decoder, self.unpack)
#fg.connect(self.preamp, self.agc, self.costas_loop, self.rrc_filter, self.clock_recovery,
# self.slicer, self.diffdec, self.gray_decoder, self.unpack)
# Debug sinks
if 1:
fg.connect(self.agc,
gr.file_sink(gr.sizeof_gr_complex, "agc.dat"))
fg.connect(self.costas_loop,
gr.file_sink(gr.sizeof_gr_complex, "costas_loop.dat"))
fg.connect(self.rrc_filter,
gr.file_sink(gr.sizeof_gr_complex, "rrc.dat"))
fg.connect(self.clock_recovery,
gr.file_sink(gr.sizeof_gr_complex, "clock_recovery.dat"))
fg.connect(self.slicer,
gr.file_sink(gr.sizeof_char, "slicer.dat"))
fg.connect(self.diffdec,
gr.file_sink(gr.sizeof_gr_complex, "diffdec.dat"))
#fg.connect(self.diffdec,
# gr.file_sink(gr.sizeof_char, "diffdec.dat"))
fg.connect(self.unpack,
gr.file_sink(gr.sizeof_char, "unpack.dat"))
# Initialize base class
gr.hier_block.__init__(self, fg, self.preamp, self.unpack)
def samples_per_baud(self):
return self.spb
def bits_per_baud(self=None): # staticmethod that's also callable on an instance
return 2
bits_per_baud = staticmethod(bits_per_baud) # make it a static method. RTFM
dqpsk_demod = dqpsk_demod__coherent_detection_of_differentially_encoded_psk