在pkt.py中，mod_pkts类中实现了对数据负载进行打包并基带调制、发送的过程，打包时可以由用户设定包的头部，默认是AKA同步矢量；调制过程为基带调制，输出复基带信号，在调制中又可以选择差分或者不差分。demod_pkt类实现拆包、解调的过程。
在mod_pkt中，过程如下：
数据负载——>通过send_pkt函数进行打包并传递给message_source——>message_source——>调制器——>输出
send_pkt的作用其实就是将数据进行打包，并将其放入message_source中，message_source实际上是个队列，可以通过message_source.mesgq.insert_tail()将数据插入队列当中。
mod_pkt类：
class mod_pkts(gr.hier_block2):
"""
Wrap an arbitrary digital modulator in our packet handling framework.
Send packets by calling send_pkt
"""
def __init__(self, modulator, preamble=None, access_code=None, msgq_limit=2,
pad_for_usrp=True, use_whitener_offset=False, modulate=True):
"""
Hierarchical block for sending packets
Packets to be sent are enqueued by calling send_pkt.
The output is the complex modulated signal at baseband.
Args:
modulator: instance of modulator class (gr_block or hier_block2) (complex baseband out)
access_code: AKA sync vector (string of 1's and 0's between 1 and 64 long)
msgq_limit: maximum number of messages in message queue (int)
pad_for_usrp: If true, packets are padded such that they end up a multiple of 128 samples
use_whitener_offset: If true, start of whitener XOR string is incremented each packet
See gmsk_mod for remaining parameters
"""
gr.hier_block2.__init__(self, "mod_pkts",
gr.io_signature(0, 0, 0), # Input signature
gr.io_signature(1, 1, gr.sizeof_gr_complex)) # Output signature
self._modulator = modulator
self._pad_for_usrp = pad_for_usrp
self._use_whitener_offset = use_whitener_offset
self._whitener_offset = 0
if access_code is None:
access_code = packet_utils.default_access_code
if not packet_utils.is_1_0_string(access_code):
raise ValueError, "Invalid access_code %r. Must be string of 1's and 0's" % (access_code,)
self._access_code = access_code
if preamble is None:
preamble = packet_utils.default_preamble
if not packet_utils.is_1_0_string(preamble):
raise ValueError, "Invalid preamble %r. Must be string of 1's and 0's" % (preamble,)
self._preamble = preamble
# accepts messages from the outside world
self._pkt_input = blocks.message_source(gr.sizeof_char, msgq_limit)
self.connect(self._pkt_input, self._modulator, self)
def send_pkt(self, payload='', eof=False):
"""
Send the payload.
Args:
payload: data to send (string)
"""
if eof:
msg = gr.message(1) # tell self._pkt_input we're not sending any more packets
else:
# print "original_payload =", string_to_hex_list(payload)
pkt = packet_utils.make_packet(payload,
self._modulator.samples_per_symbol(),
self._modulator.bits_per_symbol(),
self._preamble,
self._access_code,
self._pad_for_usrp,
self._whitener_offset)
#print "pkt =", string_to_hex_list(pkt)
msg = gr.message_from_string(pkt)
if self._use_whitener_offset is True:
self._whitener_offset = (self._whitener_offset + 1) % 16
self._pkt_input.msgq().insert_tail(msg)
demod_pkt类：
class demod_pkts(gr.hier_block2):
"""
Wrap an arbitrary digital demodulator in our packet handling framework.
The input is complex baseband. When packets are demodulated, they are passed to the
app via the callback.
"""
def __init__(self, demodulator, access_code=None, callback=None, threshold=-1):
"""
Hierarchical block for demodulating and deframing packets.
The input is the complex modulated signal at baseband.
Demodulated packets are sent to the handler.
Args:
demodulator: instance of demodulator class (gr_block or hier_block2) (complex baseband in)
access_code: AKA sync vector (string of 1's and 0's)
callback: function of two args: ok, payload (ok: bool; payload: string)
threshold: detect access_code with up to threshold bits wrong (-1 -> use default) (int)
"""
gr.hier_block2.__init__(self, "demod_pkts",
gr.io_signature(1, 1, gr.sizeof_gr_complex), # Input signature
gr.io_signature(0, 0, 0)) # Output signature
self._demodulator = demodulator
if access_code is None:
access_code = packet_utils.default_access_code
if not packet_utils.is_1_0_string(access_code):
raise ValueError, "Invalid access_code %r. Must be string of 1's and 0's" % (access_code,)
self._access_code = access_code
if threshold == -1:
threshold = 12 # FIXME raise exception
self._rcvd_pktq = gr.msg_queue() # holds packets from the PHY
self.correlator = digital.correlate_access_code_bb(access_code, threshold)
self.framer_sink = digital.framer_sink_1(self._rcvd_pktq)
self.connect(self, self._demodulator, self.correlator, self.framer_sink)
self._watcher = _queue_watcher_thread(self._rcvd_pktq, callback)
参数中的modulator和demodulator也都是gnuradio定义的类，通过benchmark_tx/rx.py在进行发送或接收Packet时传递给上述两个类。在transmit_path中，通过实例化digital.mod_pkts类实现发送器：
self.packet_transmitter = \
digital.mod_pkts(self.modulator,
access_code=None,
msgq_limit=4,
pad_for_usrp=True)
transmit_path内部实际上比较简单，结构如下：
——>包发送器——>乘法器（乘常数）——>输出
调制解调在modulation_utils中定义：
def type_1_mods():
return _type_1_modulators
def add_type_1_mod(name, mod_class):
_type_1_modulators[name] = mod_class
# Type 1 demodulators accept complex baseband input and produce a stream of bits, packed
# 1 bit / byte as their output. Their output is completely unambiguous. There is no need
# to resolve phase or polarity ambiguities.
_type_1_demodulators = {}
def type_1_demods():
return _type_1_demodulators
def add_type_1_demod(name, demod_class):
_type_1_demodulators[name] = demod_class
调制解调为通用的调制解调，可通过向其传递不同的参数选择不同的调制方式，可选的调制方式通过add_type_1_mod进行注册。比如在qpsk.py中就通过如下语句进行注册：
#
# Add these to the mod/demod registry
#
modulation_utils.add_type_1_mod('qpsk', qpsk_mod)
modulation_utils.add_type_1_demod('qpsk', qpsk_demod)
modulation_utils.add_type_1_constellation('qpsk', qpsk_constellation)
modulation_utils.add_type_1_mod('dqpsk', dqpsk_mod)
modulation_utils.add_type_1_demod('dqpsk', dqpsk_demod)
modulation_utils.add_type_1_constellation('dqpsk', dqpsk_constellation)
同时可以通过extract_kwargs_from_options（）为其传递参数
mod_kwargs = self._modulator_class.extract_kwargs_from_options(options) # transmitter
<span style="white-space:pre"> </span>self.modulator = self._modulator_class(**mod_kwargs)
self.packet_transmitter = \
digital.mod_pkts(self.modulator,
access_code=None,
msgq_limit=4,
pad_for_usrp=True)
def extract_kwargs_from_options(function, excluded_args, options):
"""
Given a function, a list of excluded arguments and the result of
parsing command line options, create a dictionary of key word
arguments suitable for passing to the function. The dictionary
will be populated with key/value pairs where the keys are those
that are common to the function's argument list (minus the
excluded_args) and the attributes in options. The values are the
corresponding values from options unless that value is None.
In that case, the corresponding dictionary entry is not populated.
(This allows different modulations that have the same parameter
names, but different default values to coexist. The downside is
that --help in the option parser will list the default as None,
but in that case the default provided in the __init__ argument
list will be used since there is no kwargs entry.)
Args:
function: the function whose parameter list will be examined
excluded_args: function arguments that are NOT to be added to the dictionary (sequence of strings)
options: result of command argument parsing (optparse.Values)
"""
# Try this in C++ ;)
args, varargs, varkw, defaults = inspect.getargspec(function)
d = {}
for kw in [a for a in args if a not in excluded_args]:
if hasattr(options, kw):
if getattr(options, kw) is not None:
d[kw] = getattr(options, kw)
return d
最后来看benchmark_tx，过程如下：
——>transmit_path——>uhd_transmitter
通过txpath = transmit_path(modulator, options)，实例化transmit_path。
对uhd_transmitter实例化：
self.sink = uhd_transmitter(options.args, symbol_rate,
options.samples_per_symbol, options.tx_freq,
options.lo_offset, options.tx_gain,
options.spec, options.antenna,
options.clock_source, options.verbose)
时间： 2024-10-31 16:09:59