jack_control.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0-or-later
# SPDX-FileCopyrightText: Copyright © 2007-2023 Nedko Arnaudov
# SPDX-FileCopyrightText: Copyright © 2008-2009 Grame
# SPDX-FileCopyrightText: Copyright © 2013 Florian Paul Schmidt
# SPDX-FileCopyrightText: Copyright © 2018 Charalampos Stratakis (RedHat)
# SPDX-FileCopyrightText: Copyright © 2020 Joshua Rubin
# SPDX-FileCopyrightText: Copyright © 2020 jneen.net
# SPDX-FileCopyrightText: Copyright © 2021 Peter Sutton
# SPDX-FileCopyrightText: Copyright © 2022 Nikita Zlobin
# SPDX-FileCopyrightText: Copyright © 2022 David Runge

from __future__ import print_function
import dbus
import sys

name_base = 'org.jackaudio'
control_interface_name = name_base + '.JackControl'
configure_interface_name = name_base + '.Configure'
service_name = name_base + '.service'
control_iface = None
configure_iface = None


def bool_convert(str_value):
    if str_value == "0" or str_value.lower() in ["false", "off", "no", "(null)"]:
        return False

    return bool(str_value)


def dbus_type_to_python_type(dbus_value):
    t = type(dbus_value)
    if t == dbus.Boolean:
        return bool(dbus_value)
    if t == dbus.Int32 or t == dbus.UInt32:
        return int(dbus_value)
    return dbus_value


def python_type_to_jackdbus_type(value, type_char):
    type_char = str(type_char)
    if type_char == "b":
        return bool_convert(value)
    elif type_char == "y":
        return dbus.Byte(ord(value))
    elif type_char == "i":
        return dbus.Int32(value)
    elif type_char == "u":
        return dbus.UInt32(value)

    return value


def dbus_typesig_to_type_string(type_char):
    type_char = str(type_char)
    if type_char == 'i':
        return "sint"
    if type_char == 'u':
        return "uint"
    if type_char == 'y':
        return "char"
    if type_char == 's':
        return "str"
    if type_char == 'b':
        return "bool"

    print('err: unknown dbus typesig "%s"' % type_char)
    return None                         # throw exception here?


def get_parameters(iface, path):
    params = iface.GetParametersInfo(path)

    # print params
    for param in params:
        typestr = dbus_typesig_to_type_string(param[0])
        name = param[1]
        # print name
        descr = param[2]
        # print descr
        isset, default, value = iface.GetParameterValue(path + [name])
        # print typestr
        if bool(isset):
            isset = "set"
        else:
            isset = "notset"
        value = dbus_type_to_python_type(value)
        default = dbus_type_to_python_type(default)

        print("%20s: %s (%s:%s:%s:%s)" % (name, descr, typestr, isset, default, value))


def print_help():
    help_output = (
        "Usage: jack_control [command] [command] ...\n"
        "Commands:\n"
        "    shell                      - execute commands from stdin until End Of File (Ctrl+D in terminal)\n"
        "    exit                       - exit jack dbus service (stops jack server if currently running)\n"
        "    help                       - print this help text\n"
        "    status                     - check whether jack server is started, "
        "return value is 0 if running and 1 otherwise\n"
        "    status2                    - show whether jack server is started, continue\n"
        "    start                      - start jack server if not currently started\n"
        "    stop                       - stop jack server if currently started\n"
        "    sm                         - switch master to currently selected driver\n"
        "    dl                         - get list of available drivers\n"
        "    dg                         - get currently selected driver\n"
        "    ds <driver>                - select driver\n"
        "    dp                         - get parameters of currently selected driver\n"
        "    dpd <param>                - get long description for driver parameter\n"
        "    dps <param> <value>        - set driver parameter\n"
        "    dpr <param>                - reset driver parameter to its default value\n"
        "    asd <driver>               - add slave driver\n"
        "    rsd <driver>               - remove slave driver\n"
        "    il                         - get list of available internals\n"
        "    ip <name>                  - get parameters of given internal\n"
        "    ipd <name> <param>         - get long description for internal parameter\n"
        "    ips <name> <param> <value> - set internal parameter\n"
        "    ipr <name> <param>         - reset internal parameter to its default value\n"
        "    iload <name>               - load internal\n"
        "    iunload <name>             - unload internal\n"
        "    ep                         - get engine parameters\n"
        "    epd <param>                - get long description for engine parameter\n"
        "    eps <param> <value>        - set engine parameter\n"
        "    epr <param>                - reset engine parameter to its default value\n"
    )
    print(help_output)

def maybe_print_param_constraint(iface, param):
    is_range, is_strict, is_fake, values = iface.GetParameterConstraint(param)
    if is_range:
        print()
        print(("allowed range: %s to %s (inclusive)" % (values[0][0], values[1][0])))
    elif len(values):
        print()
        if is_strict:
            print("allowed values:")
        else:
            print("suggested values:")

        max_len = 0
        for value in values:
            if len(str(value[0])) > max_len:
                max_len = len(str(value[0]))
        for value in values:
            print(("%*s'%s' - %s" % (1 + max_len - len(str(value[0])), "", str(value[0]), str(value[1]))))


def parse_argv(argv):
    global control_iface, configure_iface

    # check arguments
    index = 0
    while index < len(argv):
        arg = argv[index]
        index += 1
        try:
            if arg == 'exit':
                print("--- exit")
                control_iface.Exit()
            elif arg == 'status':
                print("--- status")
                if control_iface.IsStarted():
                    return (0, "started")
                else:
                    return (1, "stopped")
            elif arg == "status2":
                print("--- status")
                if control_iface.IsStarted():
                    print("started")
                else:
                    print("stopped")
            elif arg == "show":
                parse_argv(["status2", "dg", "dp", "ep"])
            elif arg == 'start':
                print("--- start")
                control_iface.StartServer()
            elif arg == 'stop':
                print("--- stop")
                control_iface.StopServer()
            elif arg == 'sm':
                print("--- switch master driver")
                control_iface.SwitchMaster()
            elif arg == 'ism':
                if control_iface.IsManuallyActivated():
                    print("Manually activated")
                else:
                    print("Automatically activated")
            elif arg == 'dl':
                print("--- drivers list")
                is_range, is_strict, is_fake_values, values = configure_iface.GetParameterConstraint(
                    ['engine', 'driver']
                )
                for value in values:
                    print(value[1])
            elif arg == 'dg':
                print("--- get selected driver")
                isset, default, value = configure_iface.GetParameterValue(['engine', 'driver'])
                print(value)
            elif arg == 'ds':
                if index >= len(argv):
                    return (1, "driver select command requires driver name argument")

                arg = argv[index]
                index += 1

                print("--- driver select \"%s\"" % arg)
                configure_iface.SetParameterValue(['engine', 'driver'], dbus.String(arg))
            elif arg == 'dp':
                print("--- get driver parameters (type:isset:default:value)")
                get_parameters(configure_iface, ['driver'])
            elif arg == 'dpd':
                if index >= len(argv):
                    return (1, "get driver parameter long description command requires parameter name argument")

                param = argv[index]
                index += 1

                print("--- get driver parameter description (%s)" % param)
                type_char, name, short_descr, long_descr = configure_iface.GetParameterInfo(['driver', param])
                print(long_descr)
                maybe_print_param_constraint(configure_iface, ['driver', param])
            elif arg == 'dps':
                if index + 1 >= len(argv):
                    return (1, "driver parameter set command requires parameter name and value arguments")

                param = argv[index]
                index += 1
                value = argv[index]
                index += 1

                print("--- driver param set \"%s\" -> \"%s\"" % (param, value))

                type_char, name, short_descr, long_descr = configure_iface.GetParameterInfo(['driver', param])
                configure_iface.SetParameterValue(['driver', param], python_type_to_jackdbus_type(value, type_char))
            elif arg == 'dpr':
                if index >= len(argv):
                    return (1, "driver parameter reset command requires parameter name argument")

                param = argv[index]
                index += 1

                print("--- driver param reset \"%s\"" % param)
                configure_iface.ResetParameterValue(['driver', param])
            elif arg == 'ep':
                print("--- get engine parameters (type:isset:default:value)")
                get_parameters(configure_iface, ['engine'])
            elif arg == 'epd':
                if index >= len(argv):
                    return (1, "get engine parameter long description command requires parameter name argument")

                param_name = argv[index]
                index += 1

                print("--- get engine parameter description (%s)" % param_name)

                type_char, name, short_descr, long_descr = configure_iface.GetParameterInfo(['engine', param_name])
                print(long_descr)
                maybe_print_param_constraint(configure_iface, ['engine', param_name])
            elif arg == 'eps':
                if index + 1 >= len(argv):
                    return (1, "engine parameter set command requires parameter name and value arguments")

                param = argv[index]
                index += 1
                value = argv[index]
                index += 1

                print("--- engine param set \"%s\" -> \"%s\"" % (param, value))

                type_char, name, short_descr, long_descr = configure_iface.GetParameterInfo(['engine', param])
                configure_iface.SetParameterValue(['engine', param], python_type_to_jackdbus_type(value, type_char))
            elif arg == 'epr':
                if index >= len(argv):
                    return (1, "engine parameter reset command requires parameter name")

                param = argv[index]
                index += 1

                print("--- engine param reset \"%s\"" % param)

                type_char, name, short_descr, long_descr = configure_iface.GetParameterInfo(['engine', param])
                configure_iface.ResetParameterValue(['engine', param])
            elif arg == 'il':
                print("--- internals list")
                is_leaf, internals = configure_iface.ReadContainer(['internals'])
                for internal in internals:
                    print(internal)
            elif arg == 'ip':
                print("--- get internal parameters (type:isset:default:value)")

                if index >= len(argv):
                    return (1, "internal parameters command requires internal name argument")

                internal_name = argv[index]
                index += 1

                get_parameters(configure_iface, ['internals', internal_name])
            elif arg == 'ipd':
                if index + 1 >= len(argv):
                    return (
                        1,
                        "get internal parameter long description command requires internal and parameter name arguments"
                    )

                name = argv[index]
                index += 1
                param = argv[index]
                index += 1

                print("--- get internal parameter description (%s)" % param)
                type_char, name, short_descr, long_descr = configure_iface.GetParameterInfo(['internals', name, param])
                print(long_descr)
            elif arg == 'ips':
                if index + 2 >= len(argv):
                    return (1, "internal parameter set command requires internal, parameter name and value arguments")

                internal_name = argv[index]
                index += 1
                param = argv[index]
                index += 1
                value = argv[index]
                index += 1

                print("--- internal param set \"%s\" -> \"%s\"" % (param, value))

                type_char, name, short_descr, long_descr = configure_iface.GetParameterInfo(
                    ['internals', internal_name, param]
                )
                configure_iface.SetParameterValue(
                    ['internals', internal_name, param],
                    python_type_to_jackdbus_type(value, type_char),
                )
            elif arg == 'ipr':
                if index + 1 >= len(argv):
                    return (1, "reset internal parameter command requires internal and parameter name arguments")

                internal_name = argv[index]
                index += 1
                param = argv[index]
                index += 1

                print("--- internal param reset \"%s\"" % param)

                configure_iface.ResetParameterValue(['internals', internal_name, param])
            elif arg == 'iload':
                print("--- load internal")

                if index >= len(argv):
                    return (1, "load internal command requires internal name argument")

                name = argv[index]
                index += 1
                control_iface.LoadInternal(name)
            elif arg == 'iunload':
                print("--- unload internal")

                if index >= len(argv):
                    return (1, "unload internal command requires internal name argument")

                name = argv[index]
                index += 1
                control_iface.UnloadInternal(name)
            elif arg == 'asd':
                print("--- add slave driver")

                if index >= len(argv):
                    return (1, "add slave driver command requires driver name argument")

                name = argv[index]
                index += 1
                control_iface.AddSlaveDriver(name)
            elif arg == 'rsd':
                print("--- remove slave driver")

                if index >= len(argv):
                    return (1, "remove slave driver command requires driver name argument")

                name = argv[index]
                index += 1
                control_iface.RemoveSlaveDriver(name)
            elif arg == 'help':
                print_help()
            else:
                return (0, "Unknown command '%s'" % arg)
        except dbus.DBusException as e:
            return (1, "DBus exception: %s" % str(e))

    return (0, "")


def shell():
    from click.parser import split_arg_string
    from sys import stdin
    while True:
        try:
            cmd = stdin.readline()
            if (cmd == ''):
                break
            s, t = parse_argv(split_arg_string(cmd))
            print("%i: %s" % (s, t), flush=True)
        except Exception:
            break


def main():
    global control_iface, configure_iface

    if len(sys.argv) == 1 or sys.argv[1] in ["-h", "--help"]:
        print_help()
        return 0

    bus = dbus.SessionBus()
    controller = bus.get_object(service_name, "/org/jackaudio/Controller")

    control_iface = dbus.Interface(controller, control_interface_name)
    configure_iface = dbus.Interface(controller, configure_interface_name)

    # check arguments
    if sys.argv[1] == "shell":
        shell()
    else:
        s, t = parse_argv(sys.argv[1:])
        if (t):
            print(t)
        return s


if __name__ == '__main__':
    sys.exit(main())