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