jack2/tools/jack_control

#!/usr/bin/python3

from __future__ import print_function
name_base = 'org.jackaudio'
control_interface_name = name_base + '.JackControl'
configure_interface_name = name_base + '.Configure'
service_name = name_base + '.service'

import sys
import os
from traceback import print_exc

import dbus

def bool_convert(str_value):
    if str_value.lower() == "false":
        return False

    if str_value.lower() == "off":
        return False

    if str_value.lower() == "no":
        return False

    if str_value == "0":
        return False

    if str_value.lower() == "(null)":
        return False

    return bool(str_value)

def dbus_type_to_python_type(dbus_value):
    if type(dbus_value) == dbus.Boolean:
        return bool(dbus_value)
    if type(dbus_value) == dbus.Int32 or type(dbus_value) == 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_type_to_type_string(dbus_value):
    if type(dbus_value) == dbus.Boolean:
        return "bool"
    if type(dbus_value) == dbus.Int32:
        return "sint"
    if type(dbus_value) == dbus.UInt32:
        return "uint"
    if type(dbus_value) == dbus.Byte:
        return "char"
    if type(dbus_value) == dbus.String:
        return "str"

    return None                         # throw exception here?

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('shit')
    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():
    print("Usage: %s [command] [command] ..." % os.path.basename(sys.argv[0]))
    print("Commands:")
    print("    exit                       - exit jack dbus service (stops jack server if currently running)")
    print("    help                       - print this help text")
    print("    status                     - check whether jack server is started, return value is 0 if running and 1 otherwise")
    print("    start                      - start jack server if not currently started")
    print("    stop                       - stop jack server if currently started")
    print("    sm                         - switch master to currently selected driver")
    print("    dl                         - get list of available drivers")
    print("    dg                         - get currently selected driver")
    print("    ds <driver>                - select driver")
    print("    dp                         - get parameters of currently selected driver")
    print("    dpd <param>                - get long description for driver parameter")
    print("    dps <param> <value>        - set driver parameter")
    print("    dpr <param>                - reset driver parameter to its default value")
    print("    asd <driver>               - add slave driver")
    print("    rsd <driver>               - remove slave driver")
    print("    il                         - get list of available internals")
    print("    ip <name>                  - get parameters of given internal")
    print("    ipd <name> <param>         - get long description for internal parameter")
    print("    ips <name> <param> <value> - set internal parameter")
    print("    ipr <name> <param>         - reset internal parameter to its default value")
    print("    iload <name>               - load internal")
    print("    iunload <name>             - unload internal")
    print("    ep                         - get engine parameters")
    print("    epd <param>                - get long description for engine parameter")
    print("    eps <param> <value>        - set engine parameter")
    print("    epr <param>                - reset engine parameter to its default value")

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]))))


class Context:
    def __init__(self, control_iface: dbus.Interface,
                 configure_iface: dbus.Interface) -> None:
        self.index = 1
        self.control_iface = control_iface
        self.configure_iface =  configure_iface


handlers = {}


def register(command: str):
    def register_handler(handler):
        handlers[command] = handler
        return handler
    return register_handler


@register('exit')
def handle_exit(ctx: Context) -> None:
    print("--- exit")
    ctx.control_iface.Exit()


@register('status')
def handle_status(ctx: Context) -> None:
    print("--- status")
    if ctx.control_iface.IsStarted():
        print("started")
        sys.exit()
    else:
        print("stopped")
        sys.exit(1)


@register('start')
def handle_start(ctx: Context) -> None:
    print("--- start")
    ctx.control_iface.StartServer()


@register('stop')
def handle_stop(ctx: Context) -> None:
    print("--- stop")
    ctx.control_iface.StopServer()


@register('sm')
def handle_sm(ctx: Context) -> None:
    print("--- switch master driver")
    ctx.control_iface.SwitchMaster()


@register('ism')
def handle_ism(ctx: Context) -> None:
    if ctx.control_iface.IsManuallyActivated():
        print("Manually activated")
    else:
        print("Automatically activated")


@register('dl')
def handle_dl(ctx: Context) -> None:
    print("--- drivers list")
    is_range, is_strict, is_fake_values, values = ctx.configure_iface.GetParameterConstraint(['engine', 'driver'])
    for value in values:
        print(value[1])


@register('dg')
def handle_dg(ctx: Context) -> None:
    print("--- get selected driver")
    isset, default, value = ctx.configure_iface.GetParameterValue(['engine', 'driver'])
    print(value)


@register('ds')
def handle_ds(ctx: Context) -> None:
    if ctx.index >= len(sys.argv):
        print("driver select command requires driver name argument")
        sys.exit(1)

    arg = sys.argv[ctx.index]
    ctx.index += 1

    print("--- driver select \"%s\"" % arg)
    ctx.configure_iface.SetParameterValue(['engine', 'driver'], dbus.String(arg))


@register('dp')
def handle_dp(ctx: Context) -> None:
    print("--- get driver parameters (type:isset:default:value)")
    get_parameters(ctx.configure_iface, ['driver'])


@register('dpd')
def handle_dpd(ctx: Context) -> None:
    if ctx.index >= len(sys.argv):
        print("get driver parameter long description command requires parameter name argument")
        sys.exit(1)

    param = sys.argv[ctx.index]
    ctx.index += 1

    print("--- get driver parameter description (%s)" % param)
    type_char, name, short_descr, long_descr = ctx.configure_iface.GetParameterInfo(['driver', param])
    print(long_descr)
    maybe_print_param_constraint(ctx.configure_iface, ['driver', param])


@register('dps')
def handle_dps(ctx: Context) -> None:
    if ctx.index + 1 >= len(sys.argv):
        print("driver parameter set command requires parameter name and value arguments")
        sys.exit(1)

    param = sys.argv[ctx.index]
    ctx.index += 1
    value = sys.argv[ctx.index]
    ctx.index += 1

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

    type_char, name, short_descr, long_descr = ctx.configure_iface.GetParameterInfo(['driver', param])
    ctx.configure_iface.SetParameterValue(['driver', param], python_type_to_jackdbus_type(value, type_char))


@register('dpr')
def handle_dpr(ctx: Context) -> None:
    if ctx.index >= len(sys.argv):
        print("driver parameter reset command requires parameter name argument")
        sys.exit(1)

    param = sys.argv[ctx.index]
    ctx.index += 1

    print("--- driver param reset \"%s\"" % param)
    ctx.configure_iface.ResetParameterValue(['driver', param])


@register('ep')
def handle_ep(ctx: Context) -> None:
    print("--- get engine parameters (type:isset:default:value)")
    get_parameters(ctx.configure_iface, ['engine'])


@register('epd')
def handle_epd(ctx: Context) -> None:
    if ctx.index >= len(sys.argv):
        print("get engine parameter long description command requires parameter name argument")
        sys.exit(1)

    param_name = sys.argv[ctx.index]
    ctx.index += 1

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

    type_char, name, short_descr, long_descr = ctx.configure_iface.GetParameterInfo(['engine', param_name])
    print(long_descr)
    maybe_print_param_constraint(ctx.configure_iface, ['engine', param_name])


@register('eps')
def handle_eps(ctx: Context) -> None:
    if ctx.index + 1 >= len(sys.argv):
        print("engine parameter set command requires parameter name and value arguments")
        sys.exit(1)

    param = sys.argv[ctx.index]
    ctx.index += 1
    value = sys.argv[ctx.index]
    ctx.index += 1

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

    type_char, name, short_descr, long_descr = ctx.configure_iface.GetParameterInfo(['engine', param])
    ctx.configure_iface.SetParameterValue(['engine', param], python_type_to_jackdbus_type(value, type_char))


@register('epr')
def handle_epr(ctx: Context) -> None:
    if ctx.index >= len(sys.argv):
        print("engine parameter reset command requires parameter name")
        sys.exit(1)

    param = sys.argv[ctx.index]
    ctx.index += 1

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

    type_char, name, short_descr, long_descr = ctx.configure_iface.GetParameterInfo(['engine', param])
    ctx.configure_iface.ResetParameterValue(['engine', param])


@register('il')
def handle_il(ctx: Context) -> None:
    print("--- internals list")
    is_leaf, internals = ctx.configure_iface.ReadContainer(['internals'])
    for internal in internals:
        print(internal)


@register('ip')
def handle_ip(ctx: Context) -> None:
    print("--- get internal parameters (type:isset:default:value)")

    if ctx.index >= len(sys.argv):
        print("internal parameters command requires internal name argument")
        sys.exit()

    internal_name = sys.argv[ctx.index]
    ctx.index += 1

    get_parameters(ctx.configure_iface, ['internals', internal_name])


@register('ipd')
def handle_ipd(ctx: Context) -> None:
    if ctx.index + 1 >= len(sys.argv):
        print("get internal parameter long description command requires internal and parameter name arguments")
        sys.exit(1)

    name = sys.argv[ctx.index]
    ctx.index += 1
    param = sys.argv[ctx.index]
    ctx.index += 1

    print("--- get internal parameter description (%s)" % param)
    type_char, name, short_descr, long_descr = ctx.configure_iface.GetParameterInfo(['internals', name, param])
    print(long_descr)


@register('ips')
def handle_ips(ctx: Context) -> None:
    if ctx.index + 2 >= len(sys.argv):
        print("internal parameter set command requires internal, parameter name and value arguments")
        sys.exit(1)

    internal_name = sys.argv[ctx.index]
    ctx.index += 1
    param = sys.argv[ctx.index]
    ctx.index += 1
    value = sys.argv[ctx.index]
    ctx.index += 1

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

    type_char, name, short_descr, long_descr = ctx.configure_iface.GetParameterInfo(['internals', internal_name, param])
    ctx.configure_iface.SetParameterValue(['internals', internal_name, param], python_type_to_jackdbus_type(value, type_char))


@register('ipr')
def handle_ipr(ctx: Context) -> None:
    if ctx.index + 1 >= len(sys.argv):
        print("reset internal parameter command requires internal and parameter name arguments")
        sys.exit(1)

    internal_name = sys.argv[ctx.index]
    ctx.index += 1
    param = sys.argv[ctx.index]
    ctx.index += 1

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

    ctx.configure_iface.ResetParameterValue(['internals', internal_name, param])


@register('iload')
def handle_iload(ctx: Context) -> None:
    print("--- load internal")

    if ctx.index >= len(sys.argv):
        print("load internal command requires internal name argument")
        sys.exit(1)

    name = sys.argv[ctx.index]
    ctx.index += 1
    result = ctx.control_iface.LoadInternal(name)


@register('iunload')
def handle_iunload(ctx: Context) -> None:
    print("--- unload internal")

    if ctx.index >= len(sys.argv):
        print("unload internal command requires internal name argument")
        sys.exit(1)

    name = sys.argv[ctx.index]
    ctx.index += 1
    result = ctx.control_iface.UnloadInternal(name)


@register('asd')
def handle_asd(ctx: Context) -> None:
    print("--- add slave driver")

    if ctx.index >= len(sys.argv):
        print("add slave driver command requires driver name argument")
        sys.exit(1)

    name = sys.argv[ctx.index]
    ctx.index += 1
    result = ctx.control_iface.AddSlaveDriver(name)


@register('rsd')
def handle_rsd(ctx: Context) -> None:
    print("--- remove slave driver")

    if ctx.index >= len(sys.argv):
        print("remove slave driver command requires driver name argument")
        sys.exit(1)

    name = sys.argv[ctx.index]
    ctx.index += 1
    result = ctx.control_iface.RemoveSlaveDriver(name)


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

    bus = dbus.SessionBus()
    controller = bus.get_object(service_name, "/org/jackaudio/Controller")
    ctx = Context(dbus.Interface(controller, control_interface_name),
                  dbus.Interface(controller, configure_interface_name))

    while ctx.index < len(sys.argv):
        arg = sys.argv[ctx.index]
        ctx.index += 1
        handler = handlers.get(arg)
        if handler is None:
            print("Unknown command '%s'" % arg)
        else:
            try:
                handler(ctx)
            except dbus.DBusException as e:
                print("DBus exception: %s" % str(e))
                return 1


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