A tool for trimming IQ recordings produced by GQRX and other SDR tools

cfile_trim.py

#!/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()