hinawa-dg00x-common-cli

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
# Copyright (C) 2018 Takashi Sakamoto

import signal

from gi.repository import GLib

from hinawa_utils.misc.cli_kit import CliKit
from hinawa_utils.dg00x.dg00x_unit import Dg00xUnit


def handle_clock_source(unit, args):
    ops = ('set', 'get')
    if len(args) > 0 and args[0] in ops:
        op = args[0]
        if op == ops[0] and len(args) == 2:
            if unit.get_property('is-locked'):
                print('Packet is-locked started.')
                return False
            source = args[1]
            unit.set_clock_source(source)
        else:
            print(unit.get_clock_source())
        return True
    print('Arguments for clock-source command:')
    print('  clock-source OPERATION [SOURCE]')
    print('    OPERATION: {0}'.format('|'.join(ops)))
    print('    SOURCE:    {0}'.format('|'.join(unit.SUPPORTED_CLOCK_SOURCES)))
    return False


def handle_sampling_rate(unit, args):
    ops = ('set', 'get')
    if len(args) > 0 and args[0] in ops:
        op = args[0]
        if op == ops[0] and len(args) == 2:
            if unit.get_property('is-locked'):
                print('Packet is-locked started.')
                return False
            rate = int(args[1])
            unit.set_local_sampling_rate(rate)
        else:
            print(unit.get_local_sampling_rate())
        return True
    print('Arguments for sampling-rate command:')
    print('  sampling-rate OPERATION [RATE]')
    print('    OPERATION: {0}'.format('|'.join(ops)))
    rates = [str(r) for r in unit.SUPPORTED_SAMPLING_RATES]
    print('    RATE:      {0}'.format('|'.join(rates)))
    return False


def handle_mixer_mode(unit, args):
    ops = ('set', 'get')
    if len(args) > 0 and args[0] in ops:
        op = args[0]
        if op == ops[0] and len(args) == 2:
            mode = int(args[1])
            unit.set_mixer_mode(mode)
        else:
            print(unit.get_mixer_mode())
        return True
    print('Arguments for mixer-mode command:')
    print('  sampling-rate OPERATION [MODE]')
    print('    OPERATION: {0}'.format('|'.join(ops)))
    print('    MODE:      0, 1')
    return False


def handle_opt_iface(unit, args):
    ops = ('set', 'get')
    if len(args) > 0 and args[0] in ops:
        op = args[0]
        if op == ops[0] and len(args) == 2:
            if unit.get_property('is-locked'):
                print('Packet is-locked started.')
                return False
            mode = args[1]
            unit.set_opt_iface(mode)
        else:
            print(unit.get_opt_iface())
        return True
    print('Arguments for optical-interface command:')
    print('  optical-interface OPERATION [MODE]')
    print('    OPERATION: {0}'.format('|'.join(ops)))
    print('    MODE:      {0}'.format(
        '|'.join(unit.SUPPORTED_OPTICAL_INTERFACES)))
    return False


def handle_listen_message(self, args):
    loop = GLib.MainLoop()

    def handle_unix_signal():
        loop.quit()

    def handle_message(unit, message):
        print('{0:08x}'.format(message))

    def handle_disconnect(unit, loop):
        loop.quit()
    self.connect('notified', handle_message)
    GLib.unix_signal_add(GLib.PRIORITY_HIGH, signal.SIGINT, handle_unix_signal)
    self.connect('notify::is-disconnected', handle_disconnect, loop)
    loop.run()
    return True


cmds = {
    'clock-source':         handle_clock_source,
    'sampling-rate':        handle_sampling_rate,
    'mixer-mode':           handle_mixer_mode,
    'optical-interface':    handle_opt_iface,
    'listen-message':       handle_listen_message,
}

fullpath = CliKit.seek_snd_unit_path()
if fullpath:
    with Dg00xUnit(fullpath) as unit:
        CliKit.dispatch_command(unit, cmds)