Hitaki.SndDice sample program for Python 3 runtime

sample

#!/usr/bin/env python3

import gi
gi.require_versions({'GLib': '2.0', 'Hitaki': '0.0'})
from gi.repository import GLib, Hitaki

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

def main() -> int:
    if len(argv) < 2:
        print('At least one argument is required for full path of ALSA HwDep cdev')
        return 1

    path = Path(argv[1]).resolve()
    if not path.exists():
        print(f'Not found: {path}')
        return 1
    
    unit = Hitaki.SndDice.new()
    _ = unit.open(str(path), 0)
    unit.connect('notified', handle_notified)

    # For asynchronous transaction, need instantiation of Hinawa.FwNode and
    # Hinawa.FwReq. The sysname of FireWire cdev can be retrieved by
    # Hitaki.AlsaFirewire:node-device. The source can be retrieved by
    # Hinawa.FwNode.create_source().

    ctx = GLib.MainContext.new()
    _, src = unit.create_source()
    src.attach(ctx)
    
    dispatcher = GLib.MainLoop.new(ctx, False)
    th = Thread(target=lambda d: d.run(), args=(dispatcher,))
    th.start()
    
    sleep(100)
    
    dispatcher.quit()
    th.join()

    return 0

FLAGS = {
    'RX_CFG_CHG': 0x00000001,
    'TX_CFG_CHG': 0x00000002,
    'LOCK_CHG': 0x00000010,
    'CLOCK_ACCEPTED': 0x00000020,
    'EXT_STATUS': 0x00000040,
}

def handle_notified(unit: Hitaki.SndDice, msg: int):
    print('Notified:')
    for name, flag in FLAGS.items():
        if msg & flag > 0:
            print(f'  {name}')

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