sdrangel/scriptsapi/dump.py

#!/usr/bin/env python3
"""
Dumps a SDRangel instance configuration to a sequence of commands in a JSON file
compatible with config.py

Each command is a JSON document with the following keys:
  - endpoint: URL suffix (API function) - mandatory
  - method: HTTP method (GET, PATCH, POST, PUT, DELETE) - mandatory
  - params: pairs of argument and values - optional
  - payload: request body in JSON format - optional
  - msg: descriptive message for console display - optional
  - delay: delay in milliseconds after command - optional
"""

import requests, traceback, sys, json, time
from optparse import OptionParser

base_url = "http://127.0.0.1:8091/sdrangel"


# ======================================================================
class Instance:
    def __init__(self):
        self.devicesets = []
        self.features = []

    def add_deviceset(self, json_data):
        ds = DeviceSet(len(self.devicesets), json_data["direction"], json_data["deviceHwType"], json_data)
        self.devicesets.append(ds)
        return ds

    def add_deviceset_item(self, direction):
        return {
            "endpoint": "/deviceset",
            "method": "POST",
            "params": {"direction": direction},
            "msg": f"Add device set"
        }

    def add_feature(self, json_data):
        feat = Feature(len(self.features), json_data["featureType"], json_data)
        self.features.append(feat)
        return feat

    def delete_all_features_items(self):
        nb_features = len(self.features)
        items = []
        for i_feat in reversed(range(nb_features)):
            items.append({
                "endpoint": f"/featureset/{self.index}/feature/{i_feat}",
                "method": "DELETE",
                "msg": f"Delete feature at index {i_feat} in feature set {self.index}"
            })
        return items

    def get_config_items(self):
        items = []
        for i_ds, deviceset in enumerate(self.devicesets):
            items.append(self.add_deviceset_item(deviceset.direction))
            items.append(deviceset.change_device_item())
            items.append(deviceset.set_device_item())
            for channel in deviceset.channels:
                items.append(channel.add_channel_item())
                items.append(channel.set_channel_item())
            items.append(deviceset.set_spectrum_item())
        for feature in self.features:
            items.append(feature.add_feature_item())
            items.append(feature.set_feature_item())
        return items

# ======================================================================
class DeviceSet:
    def __init__(self, index, direction, hwType, settings):
        self.index = index
        self.direction = direction
        self.hwType = hwType
        self.device_settings = settings
        self.spectrum_settings = {}
        self.channels = []

    def set_spectrum_settings(self, settings):
        self.spectrum_settings = settings

    def add_channel(self, json_data):
        ch = Channel(self.index, len(self.channels), json_data["direction"], json_data["channelType"], json_data)
        self.channels.append(ch)
        return ch

    def change_device_item(self):
        return {
            "endpoint": f"/deviceset/{self.index}/device",
            "method": "PUT",
            "payload": {
                "hwType": self.hwType,
                "direction": self.direction
            },
            "msg": f"Change device to {self.hwType} on device set {self.index}"
        }

    def set_device_item(self):
        return {
            "endpoint": f"/deviceset/{self.index}/device/settings",
            "method": "PUT",
            "payload": self.device_settings,
            "msg": f"Setup device {self.hwType} on device set {self.index}"
        }

    def set_spectrum_item(self):
        return {
            "endpoint": f"/deviceset/{self.index}/spectrum/settings",
            "method": "PUT",
            "payload": self.spectrum_settings,
            "msg": f"Setup spectrum on device set {self.index}"
        }

    def delete_all_channels_items(self):
        nb_channels = len(self.channels)
        items = []
        for i_chan in reversed(range(nb_channels)):
            items.append({
                "endpoint": f"/deviceset/{self.index}/channel/{i_chan}",
                "method": "DELETE",
                "msg": f"Delete channel at index {i_chan} in device set {self.index}"
            })
        return items

# ======================================================================
class Channel:
    def __init__(self, ds_index, index, direction, type, settings):
        self.ds_index = ds_index
        self.index = index
        self.direction = direction
        self.type = type
        self.channel_settings = settings

    def add_channel_item(self):
        return {
            "endpoint": f"/deviceset/{self.ds_index}/channel",
            "method": "POST",
            "payload": {
                "channelType": self.type,
                "direction": self.direction
            },
            "msg": f"Add channel {self.type} in device set index {self.ds_index}"
        }

    def set_channel_item(self):
        return {
            "endpoint": f"/deviceset/{self.ds_index}/channel/{self.index}/settings",
            "method": "PUT",
            "payload": self.channel_settings,
            "msg": f"Setup channel {self.type} at {self.index} in device set {self.ds_index}"
        }


# ======================================================================
class Feature:
    def __init__(self, index, type, settings):
        self.index = index
        self.type = type
        self.feature_settings = settings

    def add_feature_item(self):
        return {
            "endpoint": f"/featureset/feature",
            "method": "POST",
            "payload": {
                "featureType": self.type
            },
            "msg": f"Add feature {self.type} in feature set"
        }

    def set_feature_item(self):
        return {
            "endpoint": f"/featureset/feature/{self.index}/settings",
            "method": "PUT",
            "payload": self.feature_settings,
            "msg": f"Setup feature {self.type} at {self.index} in feature set"
        }


# ======================================================================
def getInputOptions():

    parser = OptionParser(usage="usage: %%prog [-t]\n")
    parser.add_option("-a", "--address", dest="address", help="Address and port. Default: 127.0.0.1:8091", metavar="ADDRESS", type="string")
    parser.add_option("-j", "--json-file", dest="json_file", help="JSON output file where commands are stored. Mandatory", metavar="FILE", type="string")

    (options, args) = parser.parse_args()

    if (options.address == None):
        options.address = "127.0.0.1:8091"

    return options


# ======================================================================
def dump():
    instance = None
    r = requests.get(url=f"{base_url}")
    if r.status_code // 100 == 2:
        instance = Instance()
        nb_devicesets = r.json()["devicesetlist"]["devicesetcount"]
        for i_deviceset in range(nb_devicesets):
            r_ds = requests.get(url=f"{base_url}/deviceset/{i_deviceset}")
            if r_ds.status_code // 100 == 2:
                nb_channels = r_ds.json()["channelcount"]
                r_dev = requests.get(url=f"{base_url}/deviceset/{i_deviceset}/device/settings")
                if r_dev.status_code // 100 == 2:
                    ds = instance.add_deviceset(r_dev.json())
                    for i_channel in range(nb_channels):
                        r_chan = requests.get(url=f"{base_url}/deviceset/{i_deviceset}/channel/{i_channel}/settings")
                        if r_chan.status_code // 100 == 2:
                            ds.add_channel(r_chan.json())
                    r_spec = requests.get(url=f"{base_url}/deviceset/{i_deviceset}/spectrum/settings")
                    if r_spec.status_code // 100 == 2:
                        ds.set_spectrum_settings(r_spec.json())
        nb_features = r.json()["featureset"]["featurecount"]
        for i_feature in range(nb_features):
            r_feat = requests.get(url=f"{base_url}/featureset/feature/{i_feature}/settings")
            if r_feat.status_code // 100 == 2:
                instance.add_feature(r_feat.json())
    return instance


# ======================================================================
def main():
    try:
        options = getInputOptions()

        global base_url
        base_url = "http://%s/sdrangel" % options.address

        instance = dump()
        json_items = instance.get_config_items()

        with open(options.json_file,'w') as json_file:
            json.dump(json_items, json_file, indent=2)

        print("All done!")

    except Exception as ex:
        tb = traceback.format_exc()
        print(tb, file=sys.stderr)


# ======================================================================
if __name__ == "__main__":
    main()