#!/usr/bin/env python3 # -*- coding: utf-8 -*- # # SPDX-License-Identifier: GPL-3.0 # # GNU Radio Python Flow Graph # Title: kinetica # Author: PE0SAT # Description: kinetica # GNU Radio version: 3.10.12.0 from PyQt5 import Qt from gnuradio import qtgui from PyQt5 import QtCore from gnuradio import analog from gnuradio import blocks import pmt from gnuradio import digital from gnuradio import filter from gnuradio.filter import firdes from gnuradio.filter import pfb import satellites import satellites.components.datasinks import satellites.components.deframers import satellites.components.demodulators import sip import threading from gnuradio import gr from gnuradio.filter import firdes from gnuradio.fft import window import sys import signal from argparse import ArgumentParser from gnuradio.eng_arg import eng_float, intx from gnuradio import eng_notation class kinetica(gr.top_block, Qt.QWidget): def __init__(self): gr.top_block.__init__(self, "kinetica", catch_exceptions=True) Qt.QWidget.__init__(self) self.setWindowTitle("kinetica") qtgui.util.check_set_qss() try: self.setWindowIcon(Qt.QIcon.fromTheme('gnuradio-grc')) except BaseException as exc: print(f"Qt GUI: Could not set Icon: {str(exc)}", file=sys.stderr) 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("gnuradio/flowgraphs", "kinetica") try: geometry = self.settings.value("geometry") if geometry: self.restoreGeometry(geometry) except BaseException as exc: print(f"Qt GUI: Could not restore geometry: {str(exc)}", file=sys.stderr) self.flowgraph_started = threading.Event() ################################################## # Variables ################################################## self.fsk_rate = fsk_rate = 9600 self.sps = sps = 4 self.samp_rate = samp_rate = 78125 self.samp_per_sym = samp_per_sym = 5 self.rf_center_freq = rf_center_freq = 400500000 self.cut_off = cut_off = (fsk_rate/2.0) * 1.12 self.ccsds_frame_size = ccsds_frame_size = 255 self.VFO = VFO = 0 self.IQ = IQ = True ################################################## # Blocks ################################################## self._VFO_range = qtgui.Range(-100000, 100000, 100, 0, 200) self._VFO_win = qtgui.RangeWidget(self._VFO_range, self.set_VFO, "'VFO'", "counter_slider", float, QtCore.Qt.Horizontal) self.top_layout.addWidget(self._VFO_win) self.satellites_print_timestamp_0 = satellites.print_timestamp('%Y-%m-%d %H:%M:%S', True) self.satellites_hexdump_sink_0 = satellites.components.datasinks.hexdump_sink(options="") self.satellites_fsk_demodulator_0 = satellites.components.demodulators.fsk_demodulator(baudrate = fsk_rate, samp_rate = sps*fsk_rate, iq = IQ, subaudio = False, options="") self.satellites_decode_rs_ccsds_0 = satellites.decode_rs(True, 1) self.satellites_ccsds_deframer_0 = satellites.components.deframers.ccsds_rs_deframer(frame_size = ccsds_frame_size, precoding = None, rs_en = False, scrambler = "CCSDS", syncword_threshold = 2, options="") self.qtgui_time_sink_x_0 = qtgui.time_sink_f( 4096, #size sps*fsk_rate, #samp_rate "", #name 1, #number of inputs None # parent ) 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 = ['red', '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.qwidget(), Qt.QWidget) self.top_layout.addWidget(self._qtgui_time_sink_x_0_win) self.qtgui_freq_sink_x_0 = qtgui.freq_sink_c( 4096, #size window.WIN_BLACKMAN_hARRIS, #wintype (rf_center_freq+VFO), #fc samp_rate, #bw "", #name 1, None # parent ) self.qtgui_freq_sink_x_0.set_update_time(0.10) self.qtgui_freq_sink_x_0.set_y_axis((-140), 10) self.qtgui_freq_sink_x_0.set_y_label('Relative Gain', 'dB') self.qtgui_freq_sink_x_0.set_trigger_mode(qtgui.TRIG_MODE_FREE, 0.0, 0, "") self.qtgui_freq_sink_x_0.enable_autoscale(False) self.qtgui_freq_sink_x_0.enable_grid(True) self.qtgui_freq_sink_x_0.set_fft_average(1.0) self.qtgui_freq_sink_x_0.enable_axis_labels(True) self.qtgui_freq_sink_x_0.enable_control_panel(True) self.qtgui_freq_sink_x_0.set_fft_window_normalized(False) labels = ['', '', '', '', '', '', '', '', '', ''] widths = [1, 1, 1, 1, 1, 1, 1, 1, 1, 1] colors = ["red", "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] for i in range(1): if len(labels[i]) == 0: self.qtgui_freq_sink_x_0.set_line_label(i, "Data {0}".format(i)) else: self.qtgui_freq_sink_x_0.set_line_label(i, labels[i]) self.qtgui_freq_sink_x_0.set_line_width(i, widths[i]) self.qtgui_freq_sink_x_0.set_line_color(i, colors[i]) self.qtgui_freq_sink_x_0.set_line_alpha(i, alphas[i]) self._qtgui_freq_sink_x_0_win = sip.wrapinstance(self.qtgui_freq_sink_x_0.qwidget(), Qt.QWidget) self.top_layout.addWidget(self._qtgui_freq_sink_x_0_win) self.pfb_arb_resampler_xxx_0 = pfb.arb_resampler_ccf( (sps*fsk_rate/samp_rate), taps=None, flt_size=32, atten=100) self.pfb_arb_resampler_xxx_0.declare_sample_delay(0) self.low_pass_filter_0_0 = filter.fir_filter_ccf( 1, firdes.low_pass( 1, (sps*fsk_rate), cut_off, 1e3, window.WIN_HAMMING, 6.76)) self.digital_binary_slicer_fb_0 = digital.binary_slicer_fb() self.blocks_throttle2_0 = blocks.throttle( gr.sizeof_gr_complex*1, (samp_rate*4), True, 0 if "auto" == "auto" else max( int(float(0.1) * (samp_rate*4)) if "auto" == "time" else int(0.1), 1) ) self.blocks_pack_k_bits_bb_0 = blocks.pack_k_bits_bb(8) self.blocks_multiply_xx_0 = blocks.multiply_vcc(1) self.blocks_file_source_0 = blocks.file_source(gr.sizeof_gr_complex*1, 'gqrx_20260728_223629_400500000_78125_fc.raw', False, 0, 0) self.blocks_file_source_0.set_begin_tag(pmt.PMT_NIL) self.blocks_file_sink_0 = blocks.file_sink(gr.sizeof_char*1, '/tmp/kinetica.bin', False) self.blocks_file_sink_0.set_unbuffered(False) self.analog_sig_source_x_0 = analog.sig_source_c(samp_rate, analog.GR_COS_WAVE, VFO, 1, 0, 0) ################################################## # Connections ################################################## self.msg_connect((self.satellites_ccsds_deframer_0, 'out'), (self.satellites_decode_rs_ccsds_0, 'in')) self.msg_connect((self.satellites_decode_rs_ccsds_0, 'out'), (self.satellites_print_timestamp_0, 'in')) self.msg_connect((self.satellites_print_timestamp_0, 'out'), (self.satellites_hexdump_sink_0, 'in')) self.connect((self.analog_sig_source_x_0, 0), (self.blocks_multiply_xx_0, 1)) self.connect((self.blocks_file_source_0, 0), (self.blocks_throttle2_0, 0)) self.connect((self.blocks_multiply_xx_0, 0), (self.pfb_arb_resampler_xxx_0, 0)) self.connect((self.blocks_pack_k_bits_bb_0, 0), (self.blocks_file_sink_0, 0)) self.connect((self.blocks_throttle2_0, 0), (self.blocks_multiply_xx_0, 0)) self.connect((self.digital_binary_slicer_fb_0, 0), (self.blocks_pack_k_bits_bb_0, 0)) self.connect((self.low_pass_filter_0_0, 0), (self.qtgui_freq_sink_x_0, 0)) self.connect((self.low_pass_filter_0_0, 0), (self.satellites_fsk_demodulator_0, 0)) self.connect((self.pfb_arb_resampler_xxx_0, 0), (self.low_pass_filter_0_0, 0)) self.connect((self.satellites_fsk_demodulator_0, 0), (self.digital_binary_slicer_fb_0, 0)) self.connect((self.satellites_fsk_demodulator_0, 0), (self.qtgui_time_sink_x_0, 0)) self.connect((self.satellites_fsk_demodulator_0, 0), (self.satellites_ccsds_deframer_0, 0)) def closeEvent(self, event): self.settings = Qt.QSettings("gnuradio/flowgraphs", "kinetica") self.settings.setValue("geometry", self.saveGeometry()) self.stop() self.wait() event.accept() def get_fsk_rate(self): return self.fsk_rate def set_fsk_rate(self, fsk_rate): self.fsk_rate = fsk_rate self.set_cut_off((self.fsk_rate/2.0) * 1.12) self.low_pass_filter_0_0.set_taps(firdes.low_pass(1, (self.sps*self.fsk_rate), self.cut_off, 1e3, window.WIN_HAMMING, 6.76)) self.pfb_arb_resampler_xxx_0.set_rate((self.sps*self.fsk_rate/self.samp_rate)) self.qtgui_time_sink_x_0.set_samp_rate(self.sps*self.fsk_rate) def get_sps(self): return self.sps def set_sps(self, sps): self.sps = sps self.low_pass_filter_0_0.set_taps(firdes.low_pass(1, (self.sps*self.fsk_rate), self.cut_off, 1e3, window.WIN_HAMMING, 6.76)) self.pfb_arb_resampler_xxx_0.set_rate((self.sps*self.fsk_rate/self.samp_rate)) self.qtgui_time_sink_x_0.set_samp_rate(self.sps*self.fsk_rate) 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_0.set_sampling_freq(self.samp_rate) self.blocks_throttle2_0.set_sample_rate((self.samp_rate*4)) self.pfb_arb_resampler_xxx_0.set_rate((self.sps*self.fsk_rate/self.samp_rate)) self.qtgui_freq_sink_x_0.set_frequency_range((self.rf_center_freq+self.VFO), self.samp_rate) def get_samp_per_sym(self): return self.samp_per_sym def set_samp_per_sym(self, samp_per_sym): self.samp_per_sym = samp_per_sym def get_rf_center_freq(self): return self.rf_center_freq def set_rf_center_freq(self, rf_center_freq): self.rf_center_freq = rf_center_freq self.qtgui_freq_sink_x_0.set_frequency_range((self.rf_center_freq+self.VFO), self.samp_rate) def get_cut_off(self): return self.cut_off def set_cut_off(self, cut_off): self.cut_off = cut_off self.low_pass_filter_0_0.set_taps(firdes.low_pass(1, (self.sps*self.fsk_rate), self.cut_off, 1e3, window.WIN_HAMMING, 6.76)) def get_ccsds_frame_size(self): return self.ccsds_frame_size def set_ccsds_frame_size(self, ccsds_frame_size): self.ccsds_frame_size = ccsds_frame_size def get_VFO(self): return self.VFO def set_VFO(self, VFO): self.VFO = VFO self.analog_sig_source_x_0.set_frequency(self.VFO) self.qtgui_freq_sink_x_0.set_frequency_range((self.rf_center_freq+self.VFO), self.samp_rate) def get_IQ(self): return self.IQ def set_IQ(self, IQ): self.IQ = IQ def main(top_block_cls=kinetica, options=None): qapp = Qt.QApplication(sys.argv) tb = top_block_cls() tb.start() tb.flowgraph_started.set() tb.show() def sig_handler(sig=None, frame=None): tb.stop() tb.wait() 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) qapp.exec_() if __name__ == '__main__': main()