hinawa-bebob-connection-cli

#!/usr/bin/env python3

import gi
gi.require_version('GLib', '2.0')
gi.require_version('Hinawa', '4.0')
from gi.repository import GLib, Hinawa

from hinawa_utils.bebob.plug_parser import PlugParser
from hinawa_utils.bebob.extensions import BcoPlugInfo

from sys import argv, exit
from pathlib import Path
from threading import Thread

def handle_dump_connections(fcp):
    unit_plug_list = PlugParser.get_unit_plug_list(fcp)
    subunit_plug_list = PlugParser.get_subunit_plug_list(fcp)

    conns = PlugParser.get_avail_connections(fcp, unit_plug_list, subunit_plug_list)

    if len(conns) == 0:
        print('nothing avail.')
        return True

    for dst_seqid, avails in conns.items():
        if dst_seqid in unit_plug_list:
            info = unit_plug_list[dst_seqid]
        elif dst_seqid in subunit_plug_list:
            info = subunit_plug_list[dst_seqid]
        else:
            return False

        dst_spec = PlugParser.get_plug_spec(fcp, info)
        print('{0} ({1})'.format(dst_spec['name'], dst_seqid))

        for avail in avails:
            src_seqid, used = avail
            if src_seqid in unit_plug_list:
                info = unit_plug_list[src_seqid]
            elif src_seqid in subunit_plug_list:
                info = subunit_plug_list[src_seqid]
            else:
                return False

            src_spec = PlugParser.get_plug_spec(fcp, info)
            if used:
                print('   < {0} ({1})'.format(src_spec['name'], src_seqid))
            else:
                print('   * {0} ({1})'.format(src_spec['name'], src_seqid))

    return True

def handle_graph_connections(fcp):
    unit_plug_list = PlugParser.get_unit_plug_list(fcp)
    subunit_plug_list = PlugParser.get_subunit_plug_list(fcp)

    specs = {}
    for plug_list in (unit_plug_list, subunit_plug_list):
        for seqid, info in plug_list.items():
            spec = PlugParser.get_plug_spec(fcp, info)
            specs[seqid] = spec

    unit_plugs = {}
    subunit_plugs = {}

    for seqid, info in unit_plug_list.items():
        dir = info['dir']
        mode = info['mode']
        data = info['data']
        unit_type = data['unit-type']
        if unit_type not in unit_plugs:
            unit_plugs[unit_type] = {}
        if dir not in unit_plugs[unit_type]:
            unit_plugs[unit_type][dir] = {}
        unit_plugs[unit_type][dir][seqid] = info

    for seqid, info in subunit_plug_list.items():
        dir = info['dir']
        mode = info['mode']
        data = info['data']
        subunit_type = (data['subunit-type'], data['subunit-id'])
        if subunit_type not in subunit_plugs:
            subunit_plugs[subunit_type] = {}
        if dir not in subunit_plugs[subunit_type]:
            subunit_plugs[subunit_type][dir] = {}
        subunit_plugs[subunit_type][dir][seqid] = info

    print('digraph {')
    print('  rankdir = "LR"')

    for unit_type, entries in unit_plugs.items():
        if len(entries) == 0:
            continue

        for dir, plugs in entries.items():
            print('  subgraph cluster_unit_{0}_{1} {{'.format(unit_type, dir))
            print('    label = "{0} {1}"'.format(unit_type, dir))
            for seqid, info in plugs.items():
                spec = specs[seqid]
                label = '{0} ({1})'.format(spec['name'], spec['type'])
                shape = 'box' if spec['type'] == 'Sync' else 'ellipse'
                print('    {0} [label = "{1}", shape = "{2}"]'.format(
                    seqid, label, shape))
            print('  }')

    for subunit_type, entries in subunit_plugs.items():
        if len(entries) == 0:
            continue

        print('  subgraph cluster_subunit_{0[0]}_{0[1]} {{'.format(subunit_type))
        print('    label = "{0[0]} subunit {0[1]}"'.format(subunit_type))

        for dir, plugs in entries.items():
            print('    subgraph cluster_subunit_{0[0]}_{0[1]}_{1} {{'.format(
                subunit_type, dir))
            print('      label = "{0}"'.format(dir))
            for seqid, info in plugs.items():
                spec = specs[seqid]
                label = '{0} ({1})'.format(spec['name'], spec['type'])
                shape = 'box' if spec['type'] == 'Sync' else 'ellipse'
                print('    {0} [label = "{1}", shape = "{2}"]'.format(
                    seqid, label, shape))
            print('    }')
        print('  }')

    conns = PlugParser.get_avail_connections(fcp, unit_plug_list, subunit_plug_list)
    for dst_seqid, avails in conns.items():
        dst_spec = specs[dst_seqid]
        for avail in avails:
            src_seqid, used = avail
            src_spec = specs[src_seqid]

            decorations = []
            if not used:
                decorations.append('style=dashed')
            if dst_spec['type'] == 'Sync' or src_spec['type'] == 'Sync':
                decorations.append('constraint=false')
            if len(decorations) > 0:
                decoration = '[{0}]'.format(' '.join(decorations))
            else:
                decoration = ''

            print('  {0} -> {1} {2}'.format(src_seqid, dst_seqid, decoration))

    print('}')

    return True

cmds = {
    'dump-connections':     handle_dump_connections,
    'graph-connections':    handle_graph_connections,
}

def print_help(reason=""):
    if len(reason) > 0:
        print(reason)
        print("")
    print("$ {} CDEV MODE".format(argv[0]))
    print("  CDEV: path to special file for Linux Firewire character device (e.g. /dev/fw1)")
    print("  MODE: {}".format(" | ".join(cmds)))

if len(argv) < 3:
    print_help("Missing arguments")
    exit(1)

path = Path(argv[1])
if not path.exists():
    print_help("{} is not found.¥n".format(path))
    exit(1)
elif not path.is_char_device():
    print_help("{} is not special file for character device.".format(path))
    exit(1)
fullpath = str(path)

node = Hinawa.FwNode.new()
try:
    node.open(fullpath, 0)
except Exception as e:
    print_help("Fail to operate {}: {}".format(fullpath, e))
    exit(1)

mode = argv[2]
if mode not in cmds:
    print_help("{} is not supported.".format(mode))
    exit(1)
op = cmds[mode]

ctx = GLib.MainContext.new()
_, src = node.create_source()
src.attach(ctx)

dispatcher = GLib.MainLoop.new(ctx, False)
th = Thread(target = lambda d: d.run(), args=(dispatcher,))
th.start()

try:
    fcp = Hinawa.FwFcp.new()
    _ = fcp.bind(node)
    op(fcp)
except Exception as e:
    print(e)
finally:
    fcp.unbind()

dispatcher.quit()
th.join()