record.py

#!/usr/bin/env python3

from argparse import ArgumentParser
import subprocess
import os
from make_sigmf import make_sigmf, dataformat_to_str
from datetime import datetime

if __name__ == "__main__":
    parser = ArgumentParser(description="This application streams data from a single channel of a USRP device to a file; it wraps a C++ program")
    parser.add_argument("--args", help="multi uhd device address args", default="")
    parser.add_argument("--file", help="name of the file to write binary samples to (default deduced from date)")
    parser.add_argument("--type", help="sample type", choices=["double", "float", "short"], default="short")
    parser.add_argument("--nsamps", help="total number of samples to receive", default=0)
    parser.add_argument("--duration", help="total number of seconds to receive", default=0)
    parser.add_argument("--spb", help="sample per buffer", default=10000)
    parser.add_argument("--rate", help="rate of incoming samples", default=1e6)
    parser.add_argument("--freq", help="RF center frequency in Hz", required=True, default=0.0)
    parser.add_argument("--lo-offset", help="offset for frontent LO in Hz (optional)")
    parser.add_argument("--gain", help="gain for the RF chain", required=True, default=40)
    parser.add_argument("--ant", help="antena selection", default="TX/RX")
    parser.add_argument("--subdev", help="subdevice specification", default="")
    parser.add_argument("--channel", help="which channel to use", default=0)
    parser.add_argument("--bw", help="analog frontend filter bandwidth in Hz")
    parser.add_argument("--ref", help="reference source", choices=["internal", "external", "mimo", "gpsdo"], default="gpsdo")
    parser.add_argument("--wirefmt", help="wire format", choices=["sc8", "sc16", "s16"])
    parser.add_argument("--setup", help="seconds of setup time", default=1.0)
    parser.add_argument("--progress", help="periodically display short-term bandwidth", action="store_true")
    parser.add_argument("--stats", help="show average bandwidth on exit", action="store_true")
    parser.add_argument("--pps", help="use external pps signal", action="store_true")
    parser.add_argument("--int-second", help="align start of reception to integer second", action="store_true")
    parser.add_argument("--null", help="run without writing to file", action="store_true")
    parser.add_argument("--continue", help="don't abort on a bad packet", action="store_true")
    parser.add_argument("--ski-lo", help="skip checking LO lock status", action="store_true")
    parser.add_argument("--int-n", help="tune USRP with integer-N tuning", action="store_true")

    args = parser.parse_args()

    if args.file is None:
        datatype = dataformat_to_str(args.type)
        args.file = f"camras-{datetime.utcnow():%Y_%m_%d_%H_%M_%S}" + \
                    f"_{float(args.freq)/1e6:.3f}MHz" + \
                    f"_{float(args.rate)/1e6:.1f}Msps" + \
                    f"_{datatype}.sigmf-data"

    args_dict = vars(args)

    bool_args = [key for key, value in args_dict.items() if value == True and isinstance(value, bool)]

    print("Running set_70cm_device.py B210")
    subprocess.check_output(["set_70cm_device.py", "B210"])

    arg_list = []
    for key, value in args_dict.items():
        if isinstance(value, bool):
            if value == True:
                arg_list += ["--" + key.replace('_', '-')]
        elif value == "" or value is None:
            continue
        else:
            arg_list += ["--" + key.replace('_', '-'), str(value)]

    print("Running usrp_to_file " + " ".join(arg_list))
    try:
        exit_code = subprocess.call(["usrp_to_file"] + arg_list)
    except KeyboardInterrupt:
        if os.path.isfile(args.file):
            exit_code = 0
        else:
            raise

    if exit_code != 0:
        raise RuntimeError("Exit code of usrp_to_file was " + exit_code)

    sigmf_meta_filename = make_sigmf(args.file + ".metadata")

    os.remove(args.file + ".metadata")

    print("Created " + sigmf_meta_filename)