make_sigmf.py

#!/usr/bin/env python3

import os
import time
from astropy.time import Time
from astropy.coordinates import SkyCoord
import json
from urllib.request import urlopen
import astropy.units as u
from astropy.table import Table

from argparse import ArgumentParser

from telescope import Telescope


def make_sigmf(iqfile, no_rename=False):
    # time_first_sample, rate, frequency, rx_gain, bandwidth, reference, time_source, format, channel, antenna, sdr_type

    filename_metadata = iqfile + ".metadata"
    with open(filename_metadata, "r") as file_metadata:
        _ = file_metadata.readline()
        metadata = file_metadata.readline().strip().split(", ")
    time_first_sample, rate, frequency, rx_gain, bandwidth, reference, time_source, dataformat, channel, antenna, sdr_type, num_samples = metadata
    rate = float(rate)
    frequency = float(frequency)
    rx_gain = float(rx_gain)
    time_first_sample = Time(float(time_first_sample), format="unix")

    # sigmfname_data = sigmfname_meta.replace("-meta", "-data")
    if dataformat.strip('"') == "short":
        datatype = "ci16_le"
    elif dataformat.strip('"') == "float":
        datatype = "cf32_le"
    elif dataformat.strip('"') == "double":
        datatype = "cf64_le"
    else:
        datatype = "unknown"

    newfilename = iqfile.replace(".iq", "") + "_" + \
        f"{frequency/1e6:.3f}MHz" + \
        f"_{rate/1e6:.1f}Msps" + \
        "_" + datatype + \
        ".sigmf-data"
    sigmfname_meta = newfilename.replace("sigmf-data", "sigmf-meta")

    meta = {}
    meta["global"] = {}
    global_meta = meta["global"]
    global_meta["core:description"] = "Dwingeloo Radio Telescope, PI9CAM"
    global_meta["core:version"] = "1.0.0"
    global_meta[
        "core:license"] = "https://creativecommons.org/licenses/by/4.0/"
    global_meta["core:author"] = "Stichting CAMRAS"
    global_meta["core:recorder"] = "usrp_to_file"
    global_meta["core:dataset"] = os.path.basename(newfilename)
    global_meta["core:datatype"] = datatype
    global_meta["core:sample_rate"] = rate
    global_meta["core:hw"] = sdr_type.strip('"')

    meta["annotations"] = []

    meta["captures"] = []
    meta["captures"].append({
        "core:sample_start": 0,
        "core:frequency": frequency,
        "core:datetime": time_first_sample.datetime.isoformat(),
    })

    # CAMRAS
    try:
        dt = Telescope(consoleHost="console")
        time.sleep(1.1)  # Make sure everything is updated
        console_is_on = True
    except IOError:
        console_is_on = False

    if not console_is_on:
        apilink = urlopen("https://www.camras.nl/pointing.php")
        pointingdata = json.loads(apilink.read().decode())

        global_meta["camras:pointing:source"] = "https://www.camras.nl/pointing.php"
        global_meta["camras:pointing:tracking"] = pointingdata[
            'setpoint'].strip('"')
        global_meta[
            "camras:pointing:az_deg"] = f"{(pointingdata['az'] + 180) % 360:.4f}"
        global_meta["camras:pointing:el_deg"] = f"{pointingdata['el']:.4f}"
        if pointingdata['offset'].strip() == "":
            offset_az_deg = 0
            offset_el_deg = 0
        else:
            offset_az_str, offset_el_str = pointingdata['offset'].split(',')
            offset_az_str = offset_az_str.strip().rstrip('°')
            offset_el_str = offset_el_str.strip().rstrip('°')
            global_meta["camras:pointing:offset:az_deg"] = offset_az_str
            global_meta["camras:pointing:offset:el_deg"] = offset_el_str
        global_meta["camras:pointing:datetime"] = Time.now(
        ).iso[:19]  # Pointing is only accurate to the second
    else:
        global_meta["camras:pointing:source"] = "DT controller"
        tracker = dt.tracker.replace("console/", "")
        global_meta["camras:pointing:tracking"] = dt.tracker
        global_meta[
            "camras:pointing:az_deg"] = f"{(dt.az.value + 180) % 360:.4f}"
        global_meta["camras:pointing:el_deg"] = f"{dt.el.value:.4f}"
        global_meta[
            "camras:pointing:offset:az_deg"] = f"{dt.offset_az.value:+.4f}"
        global_meta[
            "camras:pointing:offset:el_deg"] = f"{dt.offset_el.value:+.4f}"
        global_meta["camras:focusbox_position_m"] = f"{dt.focusbox_pos:.3f}"

        global_meta["camras:pointing:datetime"] = Time.now(
        ).datetime.isoformat()

        if tracker == 'j2000tracker':
            sky = dt.getRaDec()
            global_meta["camras:pointing:ra_deg"] = f"{dt.radec.ra.deg:.4f}"
            global_meta["camras:pointing:dec_deg"] = f"{dt.radec.dec.deg:.4f}"
            global_meta["camras:pointing:radec_hmsdms"] = sky.to_string('hmsdms')
            objects = Table.read('camras_objects.ecsv')

            catalog = SkyCoord(objects["coord"])
            names = objects["name"]
            idx, d2d, d3d = sky.match_to_catalog_sky(catalog)
            separation = sky.separation(catalog[idx])
            if separation < 6 * u.deg:
                global_meta["camras:pointing:nearest_object"] = names[
                    idx].strip('"')
                global_meta[
                    "camras:pointing:nearest_object_separation_deg"] = f"{separation.deg:.4f}"
        elif tracker == 'sattracker':
            global_meta["camras:pointing:satname"] = dt.satname
            global_meta["camras:pointing:tle1"] = dt.tle1
            global_meta["camras:pointing:tle2"] = dt.tle2

        if tracker in ('j2000tracker', 'moontracker', 'suntracker',
                          'sattracker'):
            global_meta[
                "camras:pointing:model_enabled"] = f"{dt.model_enabled}"
            global_meta[
                "camras:pointing:refraction_enabled"] = f"{dt.refraction_enabled}"

    global_meta["camras:rx_gain"] = rx_gain
    global_meta["camras:bandwidth"] = bandwidth
    global_meta["camras:reference"] = reference.strip('"')
    global_meta["camras:time_source"] = time_source.strip('"')
    global_meta["camras:antenna"] = antenna.strip('"')
    global_meta["camras:channel"] = channel

    with open(sigmfname_meta, "w") as sigmfmeta:
        json.dump(meta, sigmfmeta, indent=4)

    if not no_rename:
        os.rename(iqfile, newfilename)
        os.remove(filename_metadata)

    print(sigmfname_meta)


def is_valid_file(parser, arg):
    if not os.path.exists(arg):
        parser.error(f"The file {arg} does not exist!")
    else:
        return arg


if __name__ == "__main__":
    parser = ArgumentParser(
        description="Make SIGMF metadata out of IQ dump, "
        "metadata CSV and a connection to the telescope console")
    parser.add_argument("iqfile",
                        metavar="IQ file",
                        type=lambda x: is_valid_file(parser, x))
    parser.add_argument(
        "-n",
        "--no-rename",
        help="Dry-run: do not rename files, just write the metadata",
        action="store_true")

    args = parser.parse_args()

    make_sigmf(args.iqfile, no_rename=args.no_rename)