Josh,
I think the "trellis" part of GRC is ready for prime time.
All my tests work so far and produce the expected results.
I am attaching the following files:
trellis.py (all grc trellis defs)
__init__.py
Conversion.py (I spoted a bug in the chunks_to_symbols block and I
corrected it)
Sinks.py (I added the "numerical display block that Martin Dvh
developed--is currently part of the grnuradio trunk")
also attached find an example graph with corresponding help files
that tests a simple convolutionally encoded QPSK system.
----
Let me also take a minute and thank you for your contribution. I believe
is very important especially for those of us who need to educate
students: with GRC a student can start using gnuradio in less than a
week! I am definately planning to test this hypothesis over the summer
with a group of undergrads.
My wish list for GRC at this point includes:
1) A way to build hierarhical block diagrams, ie, group a bunch
of blocks together and define a new block so I can reuse it within GRC.
2) global variables of type string and vector
(so that I can define globally some filenames and some constellations
that are used in multiple blocks).
Thanks
Achilleas
"""
GNU Radio Companion is a graphical interface into the GNU Radio project.
Copyright (C) 2007 Josh Blum
GNU Radio Companion 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
of the License, or (at your option) any later version.
GNU Radio Companion 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 program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
"""
"""
Trellis/Misc.py
Josh Blum
"""
from DataType import *
from gnuradio import gr
from SignalBlockDefs import *
def Encoder(sb):
import Constants
from gnuradio import trellis
type = Enum([('Byte-->Byte', (trellis.encoder_bb, Byte(),Byte())),
('Byte-->Short', (trellis.encoder_bs, Byte(),Short())),
('Byte-->Int', (trellis.encoder_bi, Byte(),Int())),
('Short-->Short', (trellis.encoder_ss, Short(),Short())),
('Short-->Int', (trellis.encoder_si, Short(),Int())),
('Int-->Int', (trellis.encoder_ii, Int(),Int())),
],3)
sb.add_input_socket('in', Variable(type, index=1))
sb.add_output_socket('out', Variable(type, index=2))
sb.add_param('Type', type, False, type=True)
sb.add_param('File Path', FileOpen(Constants.DEFAULT_FILE_PATH))
sb.add_param('Initial State', Int())
sb.set_docs('''Trellis Encoder. For the time being the underlying FSM
can only be read from a file...''')
def make(fg, type, filename, s0):
fsm=trellis.fsm(filename.parse())
block = type.parse()[0](fsm, s0.parse())
return block
return sb, make
def Metrics(sb):
from gnuradio import trellis
type = Enum([('Complex', (trellis.metrics_c,
Complex(),ComplexVector())),
('Float', (trellis.metrics_f, Float(),FloatVector())),
('Int', (trellis.metrics_i, Int(),IntVector())),
('Short', (trellis.metrics_s, Short(),ShortVector())),],1)
sb.add_input_socket('in', Variable(type, index=1))
sb.add_output_socket('out', Float())
sb.add_param('Type', type, False, type=True)
sb.add_param('Output cardinality', Int())
sb.add_param('Dimensionality', Int())
sb.add_param('Constellation', Variable(type, index=2))
sb.add_param('Metric Type',
Enum([('Euclidean', trellis.TRELLIS_EUCLIDEAN),
('Hard Symbol', trellis.TRELLIS_HARD_SYMBOL),
('Hard Bit', trellis.TRELLIS_HARD_BIT)],0))
sb.set_docs('''Generate metrics required for Viterbi or SISO
algorithms.''')
def make(fg, type, O, D, Con, MetrType):
block = type.parse()[0](O.parse(), D.parse(), Con.parse(),
MetrType.parse())
return block
return sb, make
def Viterbi(sb):
import Constants
from gnuradio import trellis
type = Enum([('Byte', (trellis.viterbi_b, Byte())),
('Short', (trellis.viterbi_s, Short())),
('Int', (trellis.viterbi_i, Int())),
],1)
sb.add_input_socket('in', Float())
sb.add_output_socket('out', Variable(type, index=1))
sb.add_param('Type', type, False, type=True)
sb.add_param('File Path', FileOpen(Constants.DEFAULT_FILE_PATH))
sb.add_param('Length', Int())
sb.add_param('Initial State', Int())
sb.add_param('Final State', Int())
sb.set_docs('''Viterbi Decoder. FSM is specified through its filename
only...''')
def make(fg, type, filename, K, s0, sK):
fsm=trellis.fsm(filename.parse())
block = type.parse()[0](fsm, K.parse(), s0.parse(), sK.parse())
return block
return sb, make
def Viterbi_Combined(sb):
import Constants
from gnuradio import trellis
type = Enum([
('Complex->Byte', (trellis.viterbi_combined_cb, Complex(),
ComplexVector(), Byte())),
('Complex->Short', (trellis.viterbi_combined_cs, Complex(),
ComplexVector(), Short())),
('Complex->Int', (trellis.viterbi_combined_ci, Complex(),
ComplexVector(), Int())),
('Float->Byte', (trellis.viterbi_combined_fb, Float(),
FloatVector(), Byte())),
('Float->Short', (trellis.viterbi_combined_fs, Float(),
FloatVector(), Short())),
('Float->Int', (trellis.viterbi_combined_fi, Float(),
FloatVector(), Int())),
('Int->Byte', (trellis.viterbi_combined_ib, Float(),
FloatVector(), Byte())),
('Int->Short', (trellis.viterbi_combined_is, Float(),
FloatVector(), Short())),
('Int->Int', (trellis.viterbi_combined_ii, Float(), FloatVector(),
Int())),
('Short->Byte', (trellis.viterbi_combined_sb, Short(),
ShortVector(), Byte())),
('Short->Short', (trellis.viterbi_combined_ss, Short(),
ShortVector(), Short())),
('Short->Int', (trellis.viterbi_combined_si, Short(),
ShortVector(), Int())),
],4)
sb.add_input_socket('in', Variable(type, index=1))
sb.add_output_socket('out', Variable(type, index=3))
sb.add_param('Type', type, False, type=True)
sb.add_param('File Path', FileOpen(Constants.DEFAULT_FILE_PATH))
sb.add_param('Length', Int())
sb.add_param('Initial State', Int())
sb.add_param('Final State', Int())
sb.add_param('Dimensionality', Int())
sb.add_param('Constellation', Variable(type, index=2))
sb.add_param('Metric Type',
Enum([('Euclidean', trellis.TRELLIS_EUCLIDEAN),
('Hard Symbol', trellis.TRELLIS_HARD_SYMBOL),
('Hard Bit', trellis.TRELLIS_HARD_BIT)],0))
sb.set_docs('''Viterbi Decoder combined with metric calculation. FSM is
specified through its filename only...''')
def make(fg, type, filename, K, s0, sK, D, Con, MetrType):
fsm=trellis.fsm(filename.parse())
block = type.parse()[0](fsm, K.parse(), s0.parse(), sK.parse(),
D.parse(), Con.parse(), MetrType.parse())
return block
return sb, make
def Soft_In_Soft_Out(sb):
import Constants
from gnuradio import trellis
sb.add_input_socket('in', Float())
sb.add_output_socket('out', Float())
sb.add_param('File Path', FileOpen(Constants.DEFAULT_FILE_PATH))
sb.add_param('Length', Int())
sb.add_param('Initial State', Int())
sb.add_param('Final State', Int())
sb.add_param('Generate a-posteriori info on Input',
Enum([('Yes', True), ('No', False)]))
sb.add_param('Generate a-posteriori info on Output',
Enum([('Yes', True), ('No', False)]))
sb.add_param('min-sum or sum-product',
Enum([('Min-Sum', trellis.TRELLIS_MIN_SUM),
('Sum-Product', trellis.TRELLIS_SUM_PRODUCT)],0))
sb.set_docs('''BCJR Algorithm. FSM is specified through its filename
only...''')
def make(fg, filename, K, s0, sK,posti, posto, siso_type):
fsm=trellis.fsm(filename.parse())
block = trellis.siso_f(fsm, K.parse(), s0.parse(), sK.parse(),
posti.parse(), posto.parse(), siso_type.parse())
return block
return sb, make
def Soft_In_Soft_Out_Combined(sb):
import Constants
from gnuradio import trellis
type = Enum([
#('Complex', (trellis.siso_combined_c, Complex(),
ComplexVector())),
('Float', (trellis.siso_combined_f, Float(), FloatVector())),
#('Int', (trellis.siso_combined_i, Int(), IntVector())),
#('Short', (trellis.siso_combined_s, Short(), ShortVector())),
],0)
sb.add_input_socket('in', Variable(type, index=1))
sb.add_output_socket('out', Float())
sb.add_param('Type', type, False, type=True)
sb.add_param('File Path', FileOpen(Constants.DEFAULT_FILE_PATH))
sb.add_param('Length', Int())
sb.add_param('Initial State', Int())
sb.add_param('Final State', Int())
sb.add_param('Generate a-posteriori info on Input',
Enum([('Yes', True), ('No', False)]))
sb.add_param('Generate a-posteriori info on Output',
Enum([('Yes', True), ('No', False)]))
sb.add_param('min-sum or sum-product',
Enum([('Min-Sum', trellis.TRELLIS_MIN_SUM),
('Sum-Product', trellis.TRELLIS_SUM_PRODUCT)],0))
sb.add_param('Dimensionality', Int())
sb.add_param('Constellation', Variable(type, index=2))
sb.add_param('Metric Type',
Enum([('Euclidean', trellis.TRELLIS_EUCLIDEAN),
('Hard Symbol', trellis.TRELLIS_HARD_SYMBOL),
('Hard Bit', trellis.TRELLIS_HARD_BIT)],0))
sb.set_docs('''BCJR Algorithm combined with metric calculation. FSM is
specified through its filename only...''')
def make(fg, type, filename, K, s0, sK, posti, posto, siso_type, dim,
constel, metr_type):
fsm=trellis.fsm(filename.parse())
block = type.parse()[0](fsm, K.parse(), s0.parse(), sK.parse(),
posti.parse(), posto.parse(), siso_type.parse(), dim.parse(), constel.parse(),
metr_type.parse())
return block
return sb, make
def Interleaver(sb):
import Constants
from gnuradio import trellis
type = Enum([
('Complex', (Complex(),gr.sizeof_gr_complex)),
('Float', (Float(),gr.sizeof_float)),
('Int', (Int(),gr.sizeof_int)),
('Short',(Short(),gr.sizeof_short)),
('Byte',(Byte(),1)),
],3)
sb.add_input_socket('in', Variable(type, index=0))
sb.add_output_socket('out', Variable(type, index=0))
sb.add_param('Type', type, False, type=True)
sb.add_param('File Path', FileOpen(Constants.DEFAULT_FILE_PATH))
sb.add_param('Symbols per Block', Int(1))
sb.set_docs('''Interleave chunks of symbols. Interleaver specification
is given by specifying a file...''')
def make(fg, type, filename, SpB):
inter=trellis.interleaver(filename.parse())
block = trellis.permutation(inter.K(), inter.INTER(),
SpB.parse(), type.parse()[1])
return block
return sb, make
def Deinterleaver(sb):
import Constants
from gnuradio import trellis
type = Enum([
('Complex', (Complex(),gr.sizeof_gr_complex)),
('Float', (Float(),gr.sizeof_float)),
('Int', (Int(),gr.sizeof_int)),
('Short',(Short(),gr.sizeof_short)),
('Byte',(Byte(),1)),
],3)
sb.add_input_socket('in', Variable(type, index=0))
sb.add_output_socket('out', Variable(type, index=0))
sb.add_param('Type', type, False, type=True)
sb.add_param('File Path', FileOpen(Constants.DEFAULT_FILE_PATH))
sb.add_param('Symbols per Block', Int(1))
sb.set_docs('''Deinterleave chunks of symbols. Interleaver
specification is given by specifying a file...''')
def make(fg, type, filename, SpB):
inter=trellis.interleaver(filename.parse())
block = trellis.permutation(inter.K(), inter.DEINTER(),
SpB.parse(), type.parse()[1])
return block
return sb, make
"""
GNU Radio Companion is a graphical interface into the GNU Radio project.
Copyright (C) 2007 Josh Blum
GNU Radio Companion 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
of the License, or (at your option) any later version.
GNU Radio Companion 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 program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
"""
"""
SignalBlockDefs/__init__.py
Josh Blum
Build a signal block represented by a specific tag.
"""
from Elements import SignalBlock
from DataType import *
all_choices = [
('Complex', Complex()),
('Float', Float()),
('Int', Int()),
('Short', Short()),
('Byte', Byte()),]
all_vector_choices = [('Complex Vector', (ComplexVector(), Complex())),
('Float Vector',
(FloatVector(), Float())),
('Int Vector',
(IntVector(), Int())),
('Short Vector',
(ShortVector(), Short())),
('Byte Vector',
(ByteVector(), Byte()))]
default_samp_rate = '$samp_rate'
audio_rates = [('16KHz', int(16e3)),
('22.05KHz', int(22.05e3)),
('24KHz', int(24e3)),
('32KHz', int(32e3)),
('44.1KHz', int(44.1e3)),
('48KHz', int(48e3)),]
default_audio_rate_index = 3
import Sources
import Sinks
import Conversions
import Operators
import Filters
import Modulators
import Misc
import USRP
import Coders
import Trellis
TAGS = (
('Sources', [
('Signal Source', Sources.SignalSource),
('Noise Source', Sources.NoiseSource),
('Vector Source', Sources.VectorSource),
('Random Source', Sources.RandomVector),
('Null Source', Sources.NullSource),
('File Source', Sources.FileSource),
('Audio Source', Sources.AudioSource),
('USRP Source', USRP.USRPSource),
]),
('Sinks', [
('Scope Sink', Sinks.ScopeSink),
('FFT Sink', Sinks.FFTSink),
('Constellation Sink', Sinks.ConstellationSink),
('Null Sink', Sinks.NullSink),
('File Sink', Sinks.FileSink),
('Audio Sink', Sinks.AudioSink),
('USRP Sink', USRP.USRPSink),
('Numerical Display Sink',
Sinks.NumericalDisplaySink),
]),
('Operations', [
('Add Constant', Operators.AddConstant),
('Add Constant Vector',
Operators.AddConstantVector),
('Add', Operators.Add),
('Add Vector', Operators.AddVector),
('Subtract', Operators.Subtract),
('Multiply Constant',
Operators.MultiplyConstant),
('Multiply Constant Vector',
Operators.MultiplyConstantVector),
('Multiply', Operators.Multiply),
('Multiply Vector', Operators.MultiplyVector),
('Divide', Operators.Divide),
('nLog10', Operators.nLog10),
('nLog10 Vector', Operators.nLog10Vector),
]),
('Conversions', [
('Complex Components',
Conversions.ComplexComponents),
('Complex Conjugate',
Conversions.ComplexConjugate),
('Float to Complex',
Conversions.FloatToComplex),
('Complex to Float',
Conversions.ComplexToFloat),
('Float to Short', Conversions.FloatToShort),
('Short to Float', Conversions.ShortToFloat),
('Float to Char', Conversions.FloatToChar),
('Char to Float', Conversions.CharToFloat),
('Float to UChar', Conversions.FloatToUChar),
('UChar to Float', Conversions.UCharToFloat),
('Complex to IShort',
Conversions.ComplexToIShort),
('IShort to Complex',
Conversions.IShortToComplex),
('Unpacked to Packed',
Conversions.UnpackedToPacked),
('Packed to Unpacked',
Conversions.PackedToUnpacked),
('Unpack k Bits', Conversions.UnpackKBits),
('Binary Slicer', Conversions.BinarySlicer),
('Chunks to Symbols',
Conversions.ChunksToSymbols),
('Interleave', Conversions.Interleave),
('Deinterleave', Conversions.Deinterleave),
('Streams to Stream',
Conversions.StreamsToStream),
('Stream to Streams',
Conversions.StreamToStreams),
('Streams to Vector',
Conversions.StreamsToVector),
('Vector to Streams',
Conversions.VectorToStreams),
('Stream to Vector',
Conversions.StreamToVector),
('Vector to Stream',
Conversions.VectorToStream),
]),
('Coders', [('Constellation Decoder',
Coders.ConstellationDecoder),
('Differential Encoder',
Coders.DifferentialEncoder),
('Differential Decoder',
Coders.DifferentialDecoder),
('Differential Phasor',
Coders.DifferentialPhasor),
('Coorelate Access Code',
Coders.CoorelateAccessCode),
]),
('Filters', [
('Low Pass Filter', Filters.LowPassFilter),
('High Pass Filter', Filters.HighPassFilter),
('Band Pass Filter', Filters.BandPassFilter),
('Band Reject Filter',
Filters.BandRejectFilter),
('Window', Filters.Window),
('Root Raised Cosine',
Filters.RootRaisedCosine),
('Single Pole IIR Filter',
Filters.SinglePoleIIRFilter),
('Hilbert', Filters.Hilbert),
('Power Squelch', Filters.PowerSquelch),
('Decimate', Filters.Decimate),
('Interpolate', Filters.Interpolate),
('Automatic Gain Control',
Filters.AutomaticGainControl),
('CMA Filter', Filters.CMAFilter),
('Costas Loop', Filters.CostasLoop),
('Clock Recovery', Filters.ClockRecovery),
('FFT', Filters.FFT),
('IFFT', Filters.IFFT),
]),
('Modulators', [
('Frequency Modulation',
Modulators.FrequencyModulation),
('Phase Modulation',
Modulators.PhaseModulation),
('Quadrature Demodulation',
Modulators.QuadratureDemodulation),
('WFM Receive', Modulators.WFMReceive),
('WFM Transmit', Modulators.WFMTransmit),
('NBFM Receive', Modulators.NBFMReceive),
('NBFM Transmit', Modulators.NBFMTransmit),
('AM Demodulation', Modulators.AMDemod),
('FM Demodulation', Modulators.FMDemod),
('PSK Modulation', Modulators.PSKMod),
('PSK Demodulation', Modulators.PSKDemod),
('GMSK Modulation', Modulators.GMSKMod),
('GMSK Demodulation', Modulators.GMSKDemod),
]),
('Misc', [
('Throttle', Misc.Throttle),
('Head', Misc.Head),
('About', Misc.About),
('Note', Misc.Note),
]),
('Trellis', [
('Trellis Encoder', Trellis.Encoder),
('Metrics', Trellis.Metrics),
('Viterbi Decoder', Trellis.Viterbi),
('Viterbi Decoder Combined with Metric',
Trellis.Viterbi_Combined),
('BCJR Algorithm', Trellis.Soft_In_Soft_Out),
('BCJR Algorithm Combined with Metric',
Trellis.Soft_In_Soft_Out_Combined),
('Interleaver', Trellis.Interleaver),
('Deinterleaver', Trellis.Deinterleaver),
])
,)
def get_signal_block(parent, coor, rot, lookup_tag, id):
""" Create a new signal based on a few input parameters. """
sb = SignalBlock(parent, coor, rot, lookup_tag, id)
for category, signal_blocks in TAGS:
for tag, builder in signal_blocks:
if lookup_tag == tag: return builder(sb)
raise TagNotFoundException(lookup_tag)
# remove the bad tags #
for category,tags in TAGS:
tags_to_remove = list()
for tag in tags:
try: get_signal_block(None, (0,0), 0, tag[0], '')
except (ImportError, AttributeError), e:
print e, " in %s! -> continuing..."%tag[0]
tags_to_remove.append(tag)
for tag in tags_to_remove: tags.remove(tag)
class TagNotFoundException(Exception):
def __init__(self, value): self.value = value
def __str__(self): return 'Exception! The tag: %s could not be
found'%repr(self.value)
"""
GNU Radio Companion is a graphical interface into the GNU Radio project.
Copyright (C) 2007 Josh Blum
GNU Radio Companion 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
of the License, or (at your option) any later version.
GNU Radio Companion 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 program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
"""
"""
SignalBlockDefs/Conversions.py
Josh Blum
Convert between data types.
"""
from DataType import *
from gnuradio import gr
from SignalBlockDefs import *
from Constants import *
def ComplexComponents(sb):
sb.add_input_socket('in', Complex())
sb.add_output_socket('out', Float())
sb.add_param('Output Type', Enum([('Real', gr.complex_to_real),
('Imaginary', gr.complex_to_imag),
('Magnitude', gr.complex_to_mag),
('Phase Angle', gr.complex_to_arg)]), type=True)
return sb, lambda fg, type: type.parse()(1)
def ComplexConjugate(sb):
sb.add_input_socket('in', Complex())
sb.add_output_socket('out', Complex())
return sb, lambda fg: gr.conjugate_cc()
def FloatToComplex(sb):
sb.add_input_socket('in', Float())
sb.add_input_socket('jin', Float(), optional=True)
sb.add_output_socket('out', Complex())
sb.set_docs('''The imaginary input socket (jin) can be unconnected.''')
return sb, lambda fg: gr.float_to_complex()
def ComplexToFloat(sb):
sb.add_output_socket('out', Float())
sb.add_output_socket('jout', Float(), optional=True)
sb.add_input_socket('in', Complex())
sb.set_docs('''The imaginary output socket (jout) can be
unconnected.''')
return sb, lambda fg: gr.complex_to_float()
def ShortToFloat(sb):
sb.add_input_socket('in', Short())
sb.add_output_socket('out', Float())
return sb, lambda fg: gr.short_to_float()
def FloatToShort(sb):
sb.add_input_socket('in', Float())
sb.add_output_socket('out', Short())
return sb, lambda fg: gr.float_to_short()
def CharToFloat(sb):
sb.add_input_socket('in', Byte())
sb.add_output_socket('out', Float())
return sb, lambda fg: gr.char_to_float()
def FloatToChar(sb):
sb.add_input_socket('in', Float())
sb.add_output_socket('out', Byte())
return sb, lambda fg: gr.float_to_char()
def UCharToFloat(sb):
sb.add_input_socket('in', Byte())
sb.add_output_socket('out', Float())
return sb, lambda fg: gr.uchar_to_float()
def FloatToUChar(sb):
sb.add_input_socket('in', Float())
sb.add_output_socket('out', Byte())
return sb, lambda fg: gr.float_to_uchar()
def IShortToComplex(sb):
fcn = gr.interleaved_short_to_complex
sb.add_input_socket('in', Short())
sb.add_output_socket('out', Complex())
return sb, lambda fg: fcn()
def ComplexToIShort(sb):
fcn = gr.complex_to_interleaved_short
sb.add_input_socket('in', Complex())
sb.add_output_socket('out', Short())
return sb, lambda fg: fcn()
def UnpackedToPacked(sb):
type = Enum([('Byte', (gr.unpacked_to_packed_bb, Byte())),
('Short', (gr.unpacked_to_packed_ss,
Short())),
('Int', (gr.unpacked_to_packed_ii,
Int())),])
sb.add_input_socket('in', Variable(type, index=1))
sb.add_output_socket('out', Variable(type, index=1))
sb.add_param('Type', type, False, type=True)
sb.add_param('Bits per Chunk', Int(1, min=0))
sb.add_param('Endianness', Enum([('MSB', gr.GR_MSB_FIRST), ('LSB',
gr.GR_LSB_FIRST)]))
sb.set_docs('''Unpacked to Packed''')
return sb, lambda fg, type, bits_per_chunk, endianness:
type.parse()[0](bits_per_chunk.parse(), endianness.parse())
def PackedToUnpacked(sb):
type = Enum([('Byte', (gr.packed_to_unpacked_bb, Byte())),
('Short', (gr.packed_to_unpacked_ss,
Short())),
('Int', (gr.packed_to_unpacked_ii,
Int())),])
sb.add_input_socket('in', Variable(type, index=1))
sb.add_output_socket('out', Variable(type, index=1))
sb.add_param('Type', type, False, type=True)
sb.add_param('Bits per Chunk', Int(1, min=0))
sb.add_param('Endianness', Enum([('MSB', gr.GR_MSB_FIRST), ('LSB',
gr.GR_LSB_FIRST)]))
sb.set_docs('''Packed to Unpacked''')
return sb, lambda fg, type, bits_per_chunk, endianness:
type.parse()[0](bits_per_chunk.parse(), endianness.parse())
def UnpackKBits(sb):
fcn = gr.unpack_k_bits_bb
sb.add_input_socket('in', Byte())
sb.add_output_socket('out', Byte())
sb.add_param('K', Int(10, min=0))
return sb, lambda fg, k: fcn(k.parse())
def BinarySlicer(sb):
fcn = gr.binary_slicer_fb
sb.add_input_socket('in', Float())
sb.add_output_socket('out', Byte())
return sb, lambda fg: fcn()
def ChunksToSymbols(sb):
type = Enum([('Byte->Complex', (gr.chunks_to_symbols_bc, Byte(),
Complex(), ComplexVector())),
('Byte->Float', (gr.chunks_to_symbols_bf, Byte(), Float(),
FloatVector())),
('Int->Complex', (gr.chunks_to_symbols_ic, Int(), Complex(),
ComplexVector())),
('Int->Float', (gr.chunks_to_symbols_if, Int(), Float(),
FloatVector())),
('Short->Complex', (gr.chunks_to_symbols_sc, Short(),
Complex(), ComplexVector())),
('Short->Float', (gr.chunks_to_symbols_sf, Short(), Float(),
FloatVector())),])
sb.add_input_socket('in', Variable(type, index=1))
sb.add_output_socket('out', Variable(type, index=2))
sb.add_param('Type', type, type=True)
sb.add_param('Symbol Table', Variable(type, index=3))
sb.add_param('Dimensions', Int(1, min=1))
def make(fg, type, sym_table, dimensions):
block = type.parse()[0](sym_table.parse(), dimensions.parse())
return block
return sb, make
def Interleave(sb):
type = Enum(all_choices, 1)
sb.add_input_socket('in', Variable(type))
sb.add_output_socket('out', Variable(type))
sb.add_param('Type', type, False, type=True)
sb.add_param('Num Inputs', Int(3, min=1, max=MAX_NUM_SOCKETS),
show_label=False,
input_sockets_controller=True)
sb.set_docs('''2 <= Num Inputs <= %d'''%MAX_NUM_SOCKETS)
return sb, lambda fg, type, num_inputs:
gr.interleave(type.parse().get_num_bytes())
def Deinterleave(sb):
type = Enum(all_choices, 1)
sb.add_input_socket('in', Variable(type))
sb.add_output_socket('out', Variable(type))
sb.add_param('Type', type, False, type=True)
sb.add_param('Num Outputs', Int(3, min=1, max=MAX_NUM_SOCKETS),
show_label=False,
output_sockets_controller=True)
sb.set_docs('''2 <= Num Outputs <= %d'''%MAX_NUM_SOCKETS)
return sb, lambda fg, type, num_outputs:
gr.deinterleave(type.parse().get_num_bytes())
def StreamsToStream(sb):
fcn = gr.streams_to_stream
type = Enum(all_choices, 1)
sb.add_input_socket('in', Variable(type))
sb.add_output_socket('out', Variable(type))
sb.add_param('Type', type, False, type=True)
sb.add_param('Num Streams', Int(3, min=1, max=MAX_NUM_SOCKETS),
show_label=False,
input_sockets_controller=True)
sb.set_docs('''\
Interleave N streams into a single stream.
2 <= Num Streams <= %d'''%MAX_NUM_SOCKETS)
return sb, lambda fg, type, num_streams:
fcn(type.parse().get_num_bytes(), num_streams.parse())
def StreamToStreams(sb):
fcn = gr.stream_to_streams
type = Enum(all_choices, 1)
sb.add_input_socket('in', Variable(type))
sb.add_output_socket('out', Variable(type))
sb.add_param('Type', type, False, type=True)
sb.add_param('Num Streams', Int(3, min=1, max=MAX_NUM_SOCKETS),
show_label=False,
output_sockets_controller=True)
sb.set_docs('''\
Deinterleave a stream into N streams.
2 <= Num Streams <= %d'''%MAX_NUM_SOCKETS)
return sb, lambda fg, type, num_streams:
fcn(type.parse().get_num_bytes(), num_streams.parse())
def StreamsToVector(sb):
fcn = gr.streams_to_vector
type = Enum(all_vector_choices, 1)
sb.add_input_socket('in', Variable(type, index=1))
sb.add_output_socket('vout', Variable(type, index=0))
sb.add_param('Type', type, False, type=True)
sb.add_param('Num Streams', Int(3, min=1, max=MAX_NUM_SOCKETS),
show_label=False,
input_sockets_controller=True)
sb.set_docs('''\
Convert N streams into a vector-stream of length N.
1 <= Num Streams <= %d'''%MAX_NUM_SOCKETS)
return sb, lambda fg, type, num_streams:
fcn(type.parse()[1].get_num_bytes(), num_streams.parse())
def VectorToStreams(sb):
fcn = gr.vector_to_streams
type = Enum(all_vector_choices, 1)
sb.add_input_socket('vin', Variable(type, index=0))
sb.add_output_socket('out', Variable(type, index=1))
sb.add_param('Type', type, False, type=True)
sb.add_param('Num Streams', Int(3, min=1, max=MAX_NUM_SOCKETS),
show_label=False,
output_sockets_controller=True)
sb.set_docs('''\
Convert a vector-stream of length N into N streams.
1 <= Num Streams <= %d'''%MAX_NUM_SOCKETS)
return sb, lambda fg, type, num_streams:
fcn(type.parse()[1].get_num_bytes(), num_streams.parse())
def StreamToVector(sb):
fcn = gr.stream_to_vector
type = Enum(all_vector_choices, 1)
sb.add_input_socket('in', Variable(type, index=1))
sb.add_output_socket('vout', Variable(type, index=0))
sb.add_param('Type', type, False, type=True)
sb.add_param('Items / Block', Int(1, min=1))
sb.set_docs('''Separate a single stream into a vector-stream of size
N.''')
return sb, lambda fg, type, items_per_block:
fcn(type.parse()[1].get_num_bytes(), items_per_block.parse())
def VectorToStream(sb):
fcn = gr.vector_to_stream
type = Enum(all_vector_choices, 1)
sb.add_input_socket('vin', Variable(type, index=0))
sb.add_output_socket('out', Variable(type, index=1))
sb.add_param('Type', type, False, type=True)
sb.add_param('Items / Block', Int(1, min=1))
sb.set_docs('''Merge a vector-stream of size N into a single
stream.''')
return sb, lambda fg, type, items_per_block:
fcn(type.parse()[1].get_num_bytes(), items_per_block.parse())
"""
GNU Radio Companion is a graphical interface into the GNU Radio project.
Copyright (C) 2007 Josh Blum
GNU Radio Companion 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
of the License, or (at your option) any later version.
GNU Radio Companion 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 program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
"""
"""
SignalBlockDefs/Sinks.py
Josh Blum
Various data sinks.
"""
from DataType import *
from gnuradio import gr
from SignalBlockDefs import *
def NullSink(sb):
type = Enum(all_choices, 1)
sb.add_input_socket('in', Variable(type))
sb.add_param('Type', type, False, type=True)
sb.set_docs('''\
If the data stream is not throttled, the null sink will use 100% CPU. \
Manually connect a throttle to the null sink to save the CPU.''')
return sb, lambda fg, type: gr.null_sink(type.parse().get_num_bytes())
def FileSink(sb):
import Constants
type = Enum(all_choices, 1)
sb.add_input_socket('in', Variable(type))
sb.add_param('Type', type, False, type=True)
sb.add_param('File Path', FileSave(Constants.DEFAULT_FILE_PATH))
sb.set_docs('''\
If the data stream is not throttled, the file sink will use 100% CPU. \
Manually connect a throttle to the file sink to save the CPU.''')
return sb, lambda fg, type, file_path:
gr.file_sink(type.parse().get_num_bytes(), file_path.parse())
def AudioSink(sb):
from gnuradio import audio
sb.add_input_socket('in', Float())
sb.add_param('Sampling Rate', Enum(audio_rates,
default_audio_rate_index), type=True)
return sb, lambda fg, samp_rate: audio.sink(samp_rate.parse())
def FFTSink(sb):
from gnuradio.wxgui import fftsink
type = Enum([('Complex', (fftsink.fft_sink_c, Complex())),
('Float', (fftsink.fft_sink_f,
Float())),], 1)
sb.add_input_socket('in', Variable(type, index=1))
sb.add_param('Type', type, False, type=True)
sb.add_param('Title', String('FFT'))
sb.add_param('Y per div', Int(20))
sb.add_param('Reference Level', Int(20))
sb.add_param('Size', Int(512))
sb.add_param('Sampling Rate', Int(default_samp_rate))
sb.add_param('Options', Enum([('Off', None), ('Average Values', 1),
('Peak Hold', 2)]))
sb.set_docs('''The fft sink has a throttle automatically attatched to
it at runtime to save the CPU.''')
def make(fg, type, title, y_per_div, ref_lvl, size, samp_rate, options):
options = options.parse()
peak_hold = (options == 2)
average = (options == 1)
block = type.parse()[0](fg, fg.get_panel(),
title=title.parse(), y_per_div=y_per_div.parse(),
ref_level=ref_lvl.parse(), fft_size=size.parse(),
sample_rate=samp_rate.parse())
block.set_peak_hold(peak_hold) #set the peak hold option
outside the contructor
block.set_average(average) #set the
average option outside the contructor
fg.add_window(block.win, 1, title.parse())
fg.add_callback(block.set_ref_level, ref_lvl)
fg.add_callback(block.set_y_per_div, y_per_div)
th = gr.throttle(type.parse()[1].get_num_bytes(),
samp_rate.parse())
fg.connect(th, block)
return th
return sb, make
def ScopeSink(sb):
from gnuradio.wxgui import scopesink
type = Enum([('Complex', (scopesink.scope_sink_c, Complex())),
('Float', (scopesink.scope_sink_f,
Float())),], 1)
sb.add_input_socket('in', Variable(type, index=1))
sb.add_param('Type', type, False, type=True)
sb.add_param('Title', String('Scope'))
sb.add_param('Size', Int(512))
sb.add_param('Sampling Rate', Int(default_samp_rate))
sb.add_param('Frame Decimation', Int(1))
sb.add_param('Vertical Scale', Float(0))
sb.add_param('Time Scale', Float(.001))
sb.set_docs('''\
The scope sink has a throttle automatically attatched to it at runtime to save
the CPU. \
Complex inputs to the scope show up as 2 channels (real and imaginary). \
A Vertical Scale of 0, sets the scope to auto-scale.''')
def make(fg, type, title, size, samp_rate, frame_decim, v_scale,
t_scale):
v_scale = v_scale.parse()
if v_scale == 0: v_scale = None # v_scale = None means
auto-scale
block = type.parse()[0](fg, fg.get_panel(),
title=title.parse(),
size=size.parse(), sample_rate=samp_rate.parse(),
frame_decim=frame_decim.parse(),
v_scale=v_scale, t_scale=t_scale.parse())
fg.add_window(block.win, 0, title.parse())
fg.add_callback(block.set_sample_rate, samp_rate)
th = gr.throttle(type.parse()[1].get_num_bytes(),
samp_rate.parse())
fg.connect(th, block)
return th
return sb, make
def ConstellationSink(sb):
from gnuradio.wxgui import scopesink
fcn = scopesink.scope_sink_c
sb.add_input_socket('in', Complex())
sb.add_param('Title', String('Constellation'))
sb.add_param('Size', Int(512))
sb.add_param('Sampling Rate', Int(default_samp_rate))
sb.add_param('Frame Decimation', Int(1))
sb.add_param('Marker', Enum([('Plus', 0), ('Dot', 1)]))
sb.add_param('Vertical Scale (Q)', Float(.01))
sb.add_param('Horizontal Scale (I)', Float(.01))
sb.set_docs('''\
The constellation sink has a throttle automatically attatched to it at runtime
to save the CPU. ''')
def make(fg, title, size, samp_rate, frame_decim, marker, v_scale,
h_scale):
block = fcn(fg, fg.get_panel(), title=title.parse(),
size=size.parse(), sample_rate=samp_rate.parse(),
frame_decim=frame_decim.parse(),
v_scale=v_scale, t_scale=h_scale.parse())
block.win.info.xy = True #true for X:Y
marker = marker.parse()
if marker == 0: block.win.set_format_plus()
elif marker == 1: block.win.set_format_dot()
fg.add_window(block.win, 2, title.parse())
fg.add_callback(block.set_sample_rate, samp_rate)
th = gr.throttle(Complex().get_num_bytes(), samp_rate.parse())
fg.connect(th, block)
return th
return sb, make
def NumericalDisplaySink(sb):
from gnuradio.wxgui import numbersink
type = Enum([('Complex', (numbersink.number_sink_c, Complex())),
('Float', (numbersink.number_sink_f, Float())),], 1)
sb.add_input_socket('in', Variable(type, index=1))
sb.add_param('Type', type, False, type=True)
sb.add_param('Units', String(''))
sb.add_param('Title', String(''))
#sb.add_param('SizeX', Int(400))
#sb.add_param('SizeY', Int(50))
sb.add_param('Decimal places', Int(6))
sb.add_param('Sampling Rate', Int(default_samp_rate))
sb.add_param('Number Rate', Int(15))
sb.add_param('Base', Float(0))
sb.add_param('Factor', Float(1.0))
sb.add_param('Min', Float(-100.0))
sb.add_param('Max', Float(100.0))
sb.add_param('Averaging', Enum([('Yes', True), ('No', False)],0))
sb.add_param('alpha', Float(1.0e-6))
sb.add_param('Peak Hold', Enum([('Yes', True), ('No', False)],1))
sb.set_docs('''Display input value......''')
#def make(fg, type, unit, label, sizex, sizey, decimal_places,
sample_rate, number_rate, base_value, factor, minval, maxval, average,
avg_alpha, peak_hold):
def make(fg, type, unit, label, decimal_places, sample_rate,
number_rate, base_value, factor, minval, maxval, average, avg_alpha, peak_hold):
block = type.parse()[0](fg, fg.get_panel(),
unit=unit.parse(),
label=label.parse(),
#size=(sizex.parse(),sizey.parse()),
size=(400,50),
decimal_places=decimal_places.parse(),
sample_rate=sample_rate.parse(),
number_rate=number_rate.parse(),
base_value=base_value.parse(),
factor=factor.parse(),
minval=minval.parse(),
maxval=maxval.parse(),
average=average.parse(),
avg_alpha=avg_alpha.parse(),
peak_hold=peak_hold.parse()
)
#block.set_peak_hold(peak_hold) #set the peak hold option
outside the contructor
#block.set_average(average) #set the
average option outside the contructor
fg.add_window(block.win, 1, label.parse())
#th = gr.throttle(type.parse()[1].get_num_bytes(),
sample_rate.parse())
#fg.connect(th, block)
return block
return sb, make
<?xml version='1.0' encoding='UTF-8'?>
<flow_graph>
<timestamp>1171910581.45</timestamp>
<hostname>ernesto.eecs.umich.edu</hostname>
<version>0.60</version>
<valid>True</valid>
<window_width>1200</window_width>
<window_height>800</window_height>
<vars>
<var>
<key>packet_length</key>
<value>1000</value>
<min/>
<max/>
</var>
<var>
<key>EsN0_db</key>
<value>8.0</value>
<min/>
<max/>
</var>
</vars>
<signal_blocks>
<signal_block>
<tag>Multiply</tag>
<id>Multiply0</id>
<x_coordinate>865</x_coordinate>
<y_coordinate>302</y_coordinate>
<rotation>0</rotation>
<params>
<param>3</param>
<param>2</param>
</params>
</signal_block>
<signal_block>
<tag>Add</tag>
<id>Add0</id>
<x_coordinate>396</x_coordinate>
<y_coordinate>204</y_coordinate>
<rotation>0</rotation>
<params>
<param>0</param>
<param>2</param>
</params>
</signal_block>
<signal_block>
<tag>Short to Float</tag>
<id>Short to Float0</id>
<x_coordinate>1025</x_coordinate>
<y_coordinate>279</y_coordinate>
<rotation>0</rotation>
<params/>
</signal_block>
<signal_block>
<tag>Noise Source</tag>
<id>Noise Source0</id>
<x_coordinate>136</x_coordinate>
<y_coordinate>284</y_coordinate>
<rotation>0</rotation>
<params>
<param>0</param>
<param>1</param>
<param>(0.5*10^(-$EsN0_db/10))^0.5</param>
<param>67</param>
</params>
</signal_block>
<signal_block>
<tag>Chunks to Symbols</tag>
<id>Chunks to Symbols0</id>
<x_coordinate>86</x_coordinate>
<y_coordinate>175</y_coordinate>
<rotation>0</rotation>
<params>
<param>4</param>
<param>1,1j,-1j,-1</param>
<param>1</param>
</params>
</signal_block>
<signal_block>
<tag>Trellis Encoder</tag>
<id>Trellis Encoder0</id>
<x_coordinate>709</x_coordinate>
<y_coordinate>37</y_coordinate>
<rotation>0</rotation>
<params>
<param>3</param>
<param>/n/ernesto/home/anastas/gnuradio_tests/awgn1o2_4.fsm</param>
<param>0</param>
</params>
</signal_block>
<signal_block>
<tag>Random Source</tag>
<id>Random Source0</id>
<x_coordinate>28</x_coordinate>
<y_coordinate>11</y_coordinate>
<rotation>0</rotation>
<params>
<param>2</param>
<param>$packet_length</param>
<param>0</param>
<param>1</param>
<param>71</param>
<param>0</param>
</params>
</signal_block>
<signal_block>
<tag>Interleaver</tag>
<id>Interleaver0</id>
<x_coordinate>289</x_coordinate>
<y_coordinate>42</y_coordinate>
<rotation>0</rotation>
<params>
<param>3</param>
<param>/n/ernesto/home/anastas/gnuradio_tests/inter1.int</param>
<param>1</param>
</params>
</signal_block>
<signal_block>
<tag>Numerical Display Sink</tag>
<id>Numerical Display Sink0</id>
<x_coordinate>701</x_coordinate>
<y_coordinate>437</y_coordinate>
<rotation>0</rotation>
<params>
<param>1</param>
<param/>
<param>BER</param>
<param>3</param>
<param>1e4</param>
<param>1</param>
<param>0</param>
<param>1.0</param>
<param>0.0</param>
<param>1.0</param>
<param>0</param>
<param>1e-06</param>
<param>1</param>
</params>
</signal_block>
<signal_block>
<tag>Deinterleaver</tag>
<id>Deinterleaver0</id>
<x_coordinate>192</x_coordinate>
<y_coordinate>704</y_coordinate>
<rotation>0</rotation>
<params>
<param>3</param>
<param>/n/ernesto/home/anastas/gnuradio_tests/inter1.int</param>
<param>1</param>
</params>
</signal_block>
<signal_block>
<tag>Subtract</tag>
<id>Subtract0</id>
<x_coordinate>690</x_coordinate>
<y_coordinate>237</y_coordinate>
<rotation>0</rotation>
<params>
<param>3</param>
<param>2</param>
</params>
</signal_block>
<signal_block>
<tag>Viterbi Decoder Combined with Metric</tag>
<id>Viterbi Decoder Combined with Metric0</id>
<x_coordinate>66</x_coordinate>
<y_coordinate>425</y_coordinate>
<rotation>0</rotation>
<params>
<param>1</param>
<param>/n/ernesto/home/anastas/gnuradio_tests/awgn1o2_4.fsm</param>
<param>$packet_length</param>
<param>-1</param>
<param>-1</param>
<param>1</param>
<param>1,1j,-1j,-1</param>
<param>0</param>
</params>
</signal_block>
</signal_blocks>
<connections>
<connection>
<input_signal_block_id>Add0</input_signal_block_id>
<input_socket_index>1</input_socket_index>
<output_signal_block_id>Noise Source0</output_signal_block_id>
<output_socket_index>0</output_socket_index>
</connection>
<connection>
<input_signal_block_id>Subtract0</input_signal_block_id>
<input_socket_index>0</input_socket_index>
<output_signal_block_id>Random Source0</output_signal_block_id>
<output_socket_index>0</output_socket_index>
</connection>
<connection>
<input_signal_block_id>Viterbi Decoder Combined with Metric0</input_signal_block_id>
<input_socket_index>0</input_socket_index>
<output_signal_block_id>Add0</output_signal_block_id>
<output_socket_index>0</output_socket_index>
</connection>
<connection>
<input_signal_block_id>Multiply0</input_signal_block_id>
<input_socket_index>0</input_socket_index>
<output_signal_block_id>Subtract0</output_signal_block_id>
<output_socket_index>0</output_socket_index>
</connection>
<connection>
<input_signal_block_id>Multiply0</input_signal_block_id>
<input_socket_index>1</input_socket_index>
<output_signal_block_id>Subtract0</output_signal_block_id>
<output_socket_index>0</output_socket_index>
</connection>
<connection>
<input_signal_block_id>Short to Float0</input_signal_block_id>
<input_socket_index>0</input_socket_index>
<output_signal_block_id>Multiply0</output_signal_block_id>
<output_socket_index>0</output_socket_index>
</connection>
<connection>
<input_signal_block_id>Chunks to Symbols0</input_signal_block_id>
<input_socket_index>0</input_socket_index>
<output_signal_block_id>Trellis Encoder0</output_signal_block_id>
<output_socket_index>0</output_socket_index>
</connection>
<connection>
<input_signal_block_id>Add0</input_signal_block_id>
<input_socket_index>0</input_socket_index>
<output_signal_block_id>Chunks to Symbols0</output_signal_block_id>
<output_socket_index>0</output_socket_index>
</connection>
<connection>
<input_signal_block_id>Numerical Display Sink0</input_signal_block_id>
<input_socket_index>0</input_socket_index>
<output_signal_block_id>Short to Float0</output_signal_block_id>
<output_socket_index>0</output_socket_index>
</connection>
<connection>
<input_signal_block_id>Trellis Encoder0</input_signal_block_id>
<input_socket_index>0</input_socket_index>
<output_signal_block_id>Interleaver0</output_signal_block_id>
<output_socket_index>0</output_socket_index>
</connection>
<connection>
<input_signal_block_id>Interleaver0</input_signal_block_id>
<input_socket_index>0</input_socket_index>
<output_signal_block_id>Random Source0</output_signal_block_id>
<output_socket_index>0</output_socket_index>
</connection>
<connection>
<input_signal_block_id>Deinterleaver0</input_signal_block_id>
<input_socket_index>0</input_socket_index>
<output_signal_block_id>Viterbi Decoder Combined with Metric0</output_signal_block_id>
<output_socket_index>0</output_socket_index>
</connection>
<connection>
<input_signal_block_id>Subtract0</input_signal_block_id>
<input_socket_index>1</input_socket_index>
<output_signal_block_id>Deinterleaver0</output_signal_block_id>
<output_socket_index>0</output_socket_index>
</connection>
</connections>
</flow_graph>
10
5 8 6 1 9 7 2 3 0 4
2 4 4
0 2
0 2
1 3
1 3
0 3
3 0
1 2
2 1
AWGN CC from Proakis-Salehi pg 779
GM1o2_4=[1+D^2, 1+D+D^2] = [5, 7] (in decimal);
_______________________________________________
Discuss-gnuradio mailing list
Discuss-gnuradio@gnu.org
http://lists.gnu.org/mailman/listinfo/discuss-gnuradio
