Good morning,I want to generate a source controlled by voltage. Basically, I send messages to a source in order to change its amplitude. I have built a block to generate messages and pass them to the source. To do so, I created an "embedded python block " that generates and transmits a message every second, here is the python code of my block:
from gnuradio import gr, blocks import pmt import numpy import time#The block receives input data (the amplitude of a signal) then generates a message to be passed to a constant source so that the latter modifies its amplitude"
class message_generator(gr.sync_block):
def __init__(self):
gr.sync_block.__init__(
self,
name="message generator",
in_sig=[numpy.float32], #amplitude that the signal must have
out_sig=None
)
self.message_port_register_out(pmt.intern('out_port'))
def work(self, input_items, output_items):
self.msg_port_pub(pmt.intern('out_port'),
pmt.cons(pmt.intern("ampl"), pmt.from_double(input_items[0][0])))
time.sleep(1) return 1when I execute my flowgraph in gnuradio I have the error : self.msg_port_pub(pmt.intern('out_port'), pmt.cons(pmt.intern("ampl"), pmt.from_double(input_items[0][0]))) AttributeError: 'message_generator' object has no attribute 'msg_port_pub' . And I can't understand why if in the qa_python_message_passing.py <https://github.com/gnuradio/gnuradio/blob/master/gr-blocks/python/blocks/qa_python_message_passing.py> (below ) the function message_port_pub is called the same way as I do in mine.
class message_generator(gr.sync_block):
def __init__(self, msg_interval = 10):
gr.sync_block.__init__(
self,
name="message generator",
in_sig=[numpy.float32], #amplitude that the signal must have
out_sig=None
)
self.msg_list = []
#self.message_port_register_in(pmt.intern('in_port'))
self.message_port_register_out(pmt.intern('out_port'))
self.msg_interval = msg_interval
self.msg_ctr = 0
def work(self, input_items, output_items):
inLen = len(input_items[0])
# Create a new PMT for each input value and put it in the
message list
self.msg_list.append(pmt.from_double(input_items[0][0]))
while self.msg_ctr < len(self.msg_list) and \
(self.msg_ctr * self.msg_interval) < \
(self.nitems_read(0) + inLen):
self.message_port_pub(pmt.intern('out_port'),
pmt.cons(pmt.intern("ampl"),self.msg_list[self.msg_ctr]))
self.msg_ctr += 1
return inLen
PLease can you explain to me what is wrong on my code? I even tried to
put an attribute "msg_port_pub" when declaring the class but I still
have the same error.
from gnuradio import gr, blocks
import pmt
import numpy
import time
class message_generator(gr.sync_block):
def __init__(self):
gr.sync_block.__init__(
self,
name="message generator",
in_sig=[numpy.float32], #amplitude that the signal must have
out_sig=None
)
d_port= self.message_port_register_out(pmt.intern('out_port'))
self.msg_port_pub(d_port, self.buildMessage)
def buildMessage(self, data):
msg = pmt.cons(pmt.intern("ampl"), pmt.from_double(data))
return msg
def work(self, input_items, output_items):
messageToSend = builMessage(input_items=[0][0])
self.msg_port_pub(d_port, messageToSend)
time.sleep(1)
return 1
You can find attached my flowgraph and the resulting python code.
Thank you in advance and have a good day.
xx.grc
Description: application/gnuradio-grc
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# SPDX-License-Identifier: GPL-3.0
#
# GNU Radio Python Flow Graph
# Title: xx
# Author: me
# GNU Radio version: 3.8.2.0
from distutils.version import StrictVersion
if __name__ == '__main__':
import ctypes
import sys
if sys.platform.startswith('linux'):
try:
x11 = ctypes.cdll.LoadLibrary('libX11.so')
x11.XInitThreads()
except:
print("Warning: failed to XInitThreads()")
from PyQt5 import Qt
from gnuradio import qtgui
from gnuradio.filter import firdes
import sip
from gnuradio import analog
from gnuradio import blocks
from gnuradio import gr
import sys
import signal
from argparse import ArgumentParser
from gnuradio.eng_arg import eng_float, intx
from gnuradio import eng_notation
import epy_block_1
from gnuradio import qtgui
class xx(gr.top_block, Qt.QWidget):
def __init__(self):
gr.top_block.__init__(self, "xx")
Qt.QWidget.__init__(self)
self.setWindowTitle("xx")
qtgui.util.check_set_qss()
try:
self.setWindowIcon(Qt.QIcon.fromTheme('gnuradio-grc'))
except:
pass
self.top_scroll_layout = Qt.QVBoxLayout()
self.setLayout(self.top_scroll_layout)
self.top_scroll = Qt.QScrollArea()
self.top_scroll.setFrameStyle(Qt.QFrame.NoFrame)
self.top_scroll_layout.addWidget(self.top_scroll)
self.top_scroll.setWidgetResizable(True)
self.top_widget = Qt.QWidget()
self.top_scroll.setWidget(self.top_widget)
self.top_layout = Qt.QVBoxLayout(self.top_widget)
self.top_grid_layout = Qt.QGridLayout()
self.top_layout.addLayout(self.top_grid_layout)
self.settings = Qt.QSettings("GNU Radio", "xx")
try:
if StrictVersion(Qt.qVersion()) < StrictVersion("5.0.0"):
self.restoreGeometry(self.settings.value("geometry").toByteArray())
else:
self.restoreGeometry(self.settings.value("geometry"))
except:
pass
##################################################
# Variables
##################################################
self.samp_rate = samp_rate = 32000
##################################################
# Blocks
##################################################
self.qtgui_time_sink_x_0 = qtgui.time_sink_f(
1024, #size
samp_rate, #samp_rate
"", #name
1 #number of inputs
)
self.qtgui_time_sink_x_0.set_update_time(0.10)
self.qtgui_time_sink_x_0.set_y_axis(-1, 1)
self.qtgui_time_sink_x_0.set_y_label('Amplitude', "")
self.qtgui_time_sink_x_0.enable_tags(True)
self.qtgui_time_sink_x_0.set_trigger_mode(qtgui.TRIG_MODE_FREE, qtgui.TRIG_SLOPE_POS, 0.0, 0, 0, "")
self.qtgui_time_sink_x_0.enable_autoscale(False)
self.qtgui_time_sink_x_0.enable_grid(False)
self.qtgui_time_sink_x_0.enable_axis_labels(True)
self.qtgui_time_sink_x_0.enable_control_panel(False)
self.qtgui_time_sink_x_0.enable_stem_plot(False)
labels = ['Signal 1', 'Signal 2', 'Signal 3', 'Signal 4', 'Signal 5',
'Signal 6', 'Signal 7', 'Signal 8', 'Signal 9', 'Signal 10']
widths = [1, 1, 1, 1, 1,
1, 1, 1, 1, 1]
colors = ['blue', 'red', 'green', 'black', 'cyan',
'magenta', 'yellow', 'dark red', 'dark green', 'dark blue']
alphas = [1.0, 1.0, 1.0, 1.0, 1.0,
1.0, 1.0, 1.0, 1.0, 1.0]
styles = [1, 1, 1, 1, 1,
1, 1, 1, 1, 1]
markers = [-1, -1, -1, -1, -1,
-1, -1, -1, -1, -1]
for i in range(1):
if len(labels[i]) == 0:
self.qtgui_time_sink_x_0.set_line_label(i, "Data {0}".format(i))
else:
self.qtgui_time_sink_x_0.set_line_label(i, labels[i])
self.qtgui_time_sink_x_0.set_line_width(i, widths[i])
self.qtgui_time_sink_x_0.set_line_color(i, colors[i])
self.qtgui_time_sink_x_0.set_line_style(i, styles[i])
self.qtgui_time_sink_x_0.set_line_marker(i, markers[i])
self.qtgui_time_sink_x_0.set_line_alpha(i, alphas[i])
self._qtgui_time_sink_x_0_win = sip.wrapinstance(self.qtgui_time_sink_x_0.pyqwidget(), Qt.QWidget)
self.top_grid_layout.addWidget(self._qtgui_time_sink_x_0_win)
self.epy_block_1 = epy_block_1.message_generator()
self.blocks_vector_source_x_0 = blocks.vector_source_f((2, 0, 1), False, 1, [])
self.blocks_throttle_0 = blocks.throttle(gr.sizeof_float*1, samp_rate,True)
self.analog_sig_source_x_1 = analog.sig_source_f(samp_rate, analog.GR_CONST_WAVE, 1000, 1, 0, 0)
##################################################
# Connections
##################################################
self.msg_connect((self.epy_block_1, 'out_port'), (self.analog_sig_source_x_1, 'cmd'))
self.connect((self.analog_sig_source_x_1, 0), (self.blocks_throttle_0, 0))
self.connect((self.blocks_throttle_0, 0), (self.qtgui_time_sink_x_0, 0))
self.connect((self.blocks_vector_source_x_0, 0), (self.epy_block_1, 0))
def closeEvent(self, event):
self.settings = Qt.QSettings("GNU Radio", "xx")
self.settings.setValue("geometry", self.saveGeometry())
event.accept()
def get_samp_rate(self):
return self.samp_rate
def set_samp_rate(self, samp_rate):
self.samp_rate = samp_rate
self.analog_sig_source_x_1.set_sampling_freq(self.samp_rate)
self.blocks_throttle_0.set_sample_rate(self.samp_rate)
self.qtgui_time_sink_x_0.set_samp_rate(self.samp_rate)
def main(top_block_cls=xx, options=None):
if StrictVersion("4.5.0") <= StrictVersion(Qt.qVersion()) < StrictVersion("5.0.0"):
style = gr.prefs().get_string('qtgui', 'style', 'raster')
Qt.QApplication.setGraphicsSystem(style)
qapp = Qt.QApplication(sys.argv)
tb = top_block_cls()
tb.start()
tb.show()
def sig_handler(sig=None, frame=None):
Qt.QApplication.quit()
signal.signal(signal.SIGINT, sig_handler)
signal.signal(signal.SIGTERM, sig_handler)
timer = Qt.QTimer()
timer.start(500)
timer.timeout.connect(lambda: None)
def quitting():
tb.stop()
tb.wait()
qapp.aboutToQuit.connect(quitting)
qapp.exec_()
if __name__ == '__main__':
main()
