sdrangel/scriptsapi/config.py

#!/usr/bin/env python3
"""
Sends a sequence of commands recorded in a JSON file which is in the
form of a list of commands.

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"
app_gui = False
nb_devicesets = 0
nb_featuresets = 0

requests_methods = {
    "GET": requests.get,
    "PATCH": requests.patch,
    "POST": requests.post,
    "PUT": requests.put,
    "DELETE": requests.delete
}

# ======================================================================
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 file containing commands. Mandatory", metavar="FILE", type="string")
    parser.add_option("-i", "--init", dest="initialize", help="Initialize instance before running script", action="store_true")
    parser.add_option("-1", "--ignore-first-posts", dest="ignore_first_posts", help="Ignore first deviceset or featureset post in sequence", action="store_true")

    (options, args) = parser.parse_args()

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

    return options

# ======================================================================
def get_instance_details():
    global app_gui
    global nb_devicesets
    global nb_featuresets
    r = requests_methods["GET"](url=base_url)
    if r.status_code // 100 == 2:
        app_gui = r.json()["appname"] == "SDRangel"
        nb_devicesets = r.json()["devicesetlist"]["devicesetcount"]
        nb_featuresets = r.json()["featuresetlist"]["featuresetcount"]

# ======================================================================
def initialize():
    global nb_devicesets
    global nb_featuresets
    for i_ds in reversed(range(nb_devicesets)):
        if app_gui and i_ds == 0:
            r_ds = requests_methods["GET"](url=f"{base_url}/deviceset/0")
            if r_ds.status_code // 100 == 2:
                nb_channels = r_ds.json()["channelcount"]
                for i_chan in reversed(range(nb_channels)):
                    requests_methods["DELETE"](url=f"{base_url}/deviceset/0/channel/{i_chan}")
        else:
            r_del = requests_methods["DELETE"](url=f"{base_url}/deviceset")
            if r_del.status_code // 100 == 2:
                nb_devicesets -= 1
    for i_fs in reversed(range(nb_featuresets)):
        if app_gui and i_fs == 0:
            r_fs = requests_methods["GET"](url=f"{base_url}/featureset/0")
            if r_fs.status_code // 100 == 2:
                nb_features = r_fs.json()["featurecount"]
                for i_feat in reversed(range(nb_features)):
                    requests_methods["DELETE"](url=f"{base_url}/featureset/0/feature/{i_feat}")
        else:
            r_del = requests_methods["DELETE"](url=f"{base_url}/featureset")
            if r_del.status_code // 100 == 2:
                nb_featuresets -= 1


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

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

        get_instance_details()

        if options.initialize:
            initialize()

        nb_featureset_posts = 0
        nb_deviceset_posts = 0

        with open(options.json_file) as json_file:
            commands = json.load(json_file)
            for command in commands:
                endpoint = command.get('endpoint', None)
                if endpoint is None:
                    continue
                url = base_url + endpoint
                http_method = command.get('method', None)
                if endpoint == "/deviceset" and http_method == "POST":
                    nb_deviceset_posts += 1
                    if nb_deviceset_posts == 1 and options.ignore_first_posts:
                        print("First deviceset creation ignored")
                        continue
                if endpoint == "/featureset" and http_method == "POST":
                    nb_featureset_posts += 1
                    if nb_featureset_posts == 1 and options.ignore_first_posts:
                        print("First featureset creation ignored")
                        continue
                method = requests_methods.get(http_method, None) if http_method is not None else None
                if method is not None:
                    request_params = command.get('params', None)
                    request_json = command.get('payload', None)
                    r = method(url=url, params=request_params, json=request_json)
                    if r.status_code // 100 == 2:
                        print("Done: %s" % command.get('msg', ''))
                    else:
                        print("Error %d:%s" % (r.status_code, command.get('msg', '')))
                        print(r.text)
                        exit(1)
                else:
                    print("requests method error")
                    exit(1)
                delay = command.get('delay', None)
                if delay is not None and isinstance(delay, int):
                    time.sleep(delay/1000)

        print("All done!")

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


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