Skip to content

Instantly share code, notes, and snippets.

@ckuethe
Created January 23, 2025 20:47
Show Gist options
  • Save ckuethe/8655bd0c6a9e5f754b0e88aa986d93b5 to your computer and use it in GitHub Desktop.
Save ckuethe/8655bd0c6a9e5f754b0e88aa986d93b5 to your computer and use it in GitHub Desktop.
A tool for trimming IQ recordings produced by GQRX and other SDR tools
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# SPDX-License-Identifier: GPL-3.0
# GNU Radio Python Flow Graph
# Author: Chris Kuethe <[email protected]>
# GNU Radio version: 3.10.1.1
import signal
import sys
from argparse import ArgumentParser, Namespace
from gnuradio import blocks, eng_notation, gr
from gnuradio.eng_arg import eng_float, intx
from pmt import PMT_NIL
def get_args() -> Namespace:
ap = ArgumentParser(description="Trim a cfile recording, such as that produced by gqrx or the file sink block")
def _time(s) -> int:
x = [int(i, 10) for i in s.split(":")]
if len(x) != 3:
raise ValueError
return int(x[0] * 3600 + x[1] * 60 + x[2])
ap.add_argument("-i", "--input", type=str, metavar="FILE", required=True)
ap.add_argument(
"-t",
"--duration",
type=_time,
metavar="H:MM:SS",
help="Duration to copy",
required=True,
)
ap.add_argument(
"-r",
"--samp-rate",
type=float,
required=True,
metavar="HZ",
help="Sample rate of the file",
)
ap.add_argument(
"-s",
"--skip",
type=_time,
default="0:00:00",
metavar="H:MM:SS",
help="Duration to seek before copying",
)
ap.add_argument(
"-o",
"--output",
type=str,
metavar="FILE",
help="Output filename. Default: <input>.edit.cfile",
)
rv = ap.parse_args()
if not rv.output:
rv.output = rv.input + ".edit.cfile"
return rv
class IqTrimmer(gr.top_block):
def __init__(self, args: Namespace):
gr.top_block.__init__(self, "IQ Trimmer", catch_exceptions=True)
##################################################
# Variables
##################################################
skip_items = int(args.skip * args.samp_rate)
proc_items = int(args.duration * args.samp_rate)
##################################################
# Blocks
##################################################
self.blocks_skiphead = blocks.skiphead(gr.sizeof_gr_complex * 1, skip_items)
self.blocks_head = blocks.head(gr.sizeof_gr_complex * 1, proc_items)
self.blocks_file_source = blocks.file_source(gr.sizeof_gr_complex * 1, args.input, False, 0, 0)
self.blocks_file_source.set_begin_tag(PMT_NIL)
self.blocks_file_sink = blocks.file_sink(gr.sizeof_gr_complex * 1, args.output, False)
self.blocks_file_sink.set_unbuffered(False)
##################################################
# Connections
##################################################
# file_source : read file
# head : read through until the desired end time
# skiphead: skip any junk at the beginning
# file_sink: and write the result to disk
self.connect((self.blocks_file_source, 0), (self.blocks_head, 0))
self.connect((self.blocks_head, 0), (self.blocks_skiphead, 0))
self.connect((self.blocks_skiphead, 0), (self.blocks_file_sink, 0))
def main():
args = get_args()
print(args)
iqt = IqTrimmer(args)
def sig_handler(sig=None, frame=None):
iqt.stop()
iqt.wait()
sys.exit(0)
signal.signal(signal.SIGINT, sig_handler)
signal.signal(signal.SIGTERM, sig_handler)
iqt.start()
iqt.wait()
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment