samples: clean up a sample to allocate/deallocate isochronous resourc…

…es by blocking API

Signed-off-by: Takashi Sakamoto <o-takashi@sakamocchi.jp>

samples/iso-resource

@@ -1,104 +1,163 @@
 #!/usr/bin/env python3
 
-import gi
-gi.require_versions({'GLib': '2.0', 'Hinoko': '1.0'})
-from gi.repository import GLib, Hinoko
+from pathlib import Path
+from sys import argv, exit
+import traceback
 
 from threading import Thread
-from time import sleep
-
-bytes_per_payload = (2 + 8 * 3) * 4
-use_channel = 10
-use_bandwidth = Hinoko.FwIsoResource.calculate_bandwidth(bytes_per_payload,
-                                                         Hinoko.FwScode.S400)
-
-ctx = GLib.MainContext.new()
-dispatcher = GLib.MainLoop.new(ctx, False)
-th = Thread(target=lambda d: d.run(), args=(dispatcher,))
-th.start()
-
-def handle_event(res, channel, bandwidth, exception, ev_name):
-    print('event:', ev_name)
-    if exception is None:
-        print('  ', channel, bandwidth, res.get_property('generation'))
-    elif exception.matches(Hinoko.fw_iso_resource_error_quark(), Hinoko.FwIsoResourceError.EVENT):
-        print('  ', exception)
-
-# For Once mode. The allocation is bound to current generation of the bus and
-# lost automatically by bus reset without notification.
-res = Hinoko.FwIsoResourceOnce.new()
-res.open('/dev/fw0', 0)
-res.connect('allocated', handle_event, 'allocated')
-res.connect('deallocated', handle_event, 'deallocated')
-
-_, src = res.create_source()
-src.attach(ctx)
-
-for i in range(2):
-    try:
-        res.allocate_wait((use_channel, ), use_bandwidth, 100)
-    except GLib.Error as e:
-        if e.matches(Hinoko.fw_iso_resource_error_quark(), Hinoko.FwIsoResourceError.EVENT):
-            print(e)
-    else:
-        print('result: allocate')
-        print('  ', use_channel, use_bandwidth, res.get_property('generation'))
+from contextlib import AbstractContextManager, contextmanager
 
-    sleep(2)
+import gi
 
-    try:
-        res.deallocate_wait(use_channel, use_bandwidth, 100)
-    except GLib.Error as e:
-        if e.matches(Hinoko.fw_iso_resource_error_quark(), Hinoko.FwIsoResourceError.EVENT):
-            print(e)
-    else:
-        print('result: deallocate')
-        print('  ', use_channel, use_bandwidth, res.get_property('generation'))
+gi.require_versions({"GLib": "2.0", "Hinoko": "1.0"})
+from gi.repository import GLib, Hinoko
 
-    sleep(2)
+BYTES_PER_PAYLOAD = (2 + 8 * 3) * 4
+USE_CHANNEL = 10
+USE_BANDWIDTH = Hinoko.FwIsoResource.calculate_bandwidth(
+    BYTES_PER_PAYLOAD, Hinoko.FwScode.S400
+)
+TIMEOUT_MS = 100
+
+
+# Just for printing events, no functional purpose.
+def handle_resource_signal(
+    res: Hinoko.FwIsoResource,
+    channel: int,
+    bandwidth: int,
+    error: GLib.Error,
+    label: str,
+):
+    print("event:", label)
+    if error is None:
+        print("  ", channel, bandwidth, res.get_property("generation"))
+    elif error.matches(
+        Hinoko.fw_iso_resource_error_quark(), Hinoko.FwIsoResourceError.EVENT
+    ):
+        print("  ", error)
+
+
+# For Once mode. The allocation is bound to current generation of the bus and lost automatically by
+# bus reset without notification.
+def main_once(res: Hinoko.FwIsoResourceAuto, ctx: GLib.MainContext):
+    _, src = res.create_source()
+    _ = src.attach(ctx)
+
+    handler_id_allocated = res.connect("allocated", handle_resource_signal, "allocated")
+    handler_id_deallocated = res.connect(
+        "deallocated", handle_resource_signal, "deallocated"
+    )
+
+    for i in range(2):
+        res.allocate_wait((USE_CHANNEL,), USE_BANDWIDTH, TIMEOUT_MS)
+        print("result: allocate")
+        print("  ", USE_CHANNEL, USE_BANDWIDTH, res.get_property("generation"))
+
+        res.deallocate_wait(USE_CHANNEL, USE_BANDWIDTH, TIMEOUT_MS)
+        print("result: deallocate")
+        print("  ", USE_CHANNEL, USE_BANDWIDTH, res.get_property("generation"))
+
+    res.disconnect(handler_id_deallocated)
+    res.disconnect(handler_id_allocated)
+    src.destroy()
+
+
+# For Auto mode. The allocation is kept by Linux FireWire subsystem as long as the lifetime of
+# file descriptor. The instance of object represents state machine.
+def main_auto(res: Hinoko.FwIsoResourceAuto, ctx: GLib.MainContext):
+    def print_props(res):
+        for prop in ("is-allocated", "channel", "bandwidth", "generation"):
+            print("  {}: {}".format(prop, res.get_property(prop)))
+
+    _, src = res.create_source()
+    _ = src.attach(ctx)
+
+    handler_id_allocated = res.connect("allocated", handle_resource_signal, "allocated")
+    handler_id_deallocated = res.connect(
+        "deallocated", handle_resource_signal, "deallocated"
+    )
 
-src.destroy()
-del res
+    print_props(res)
 
-# For Auto mode. The allocation is kept by Linux FireWire subsystem as long as
-# the lifetime of file descriptor. The instance of object represents state
-# machine.
-res = Hinoko.FwIsoResourceAuto.new()
-res.open('/dev/fw0', 0)
-res.connect('allocated', handle_event, 'allocated')
-res.connect('deallocated', handle_event, 'deallocated')
+    for i in range(2):
+        res.allocate_wait([USE_CHANNEL], USE_BANDWIDTH, TIMEOUT_MS)
+        print("result: allocate")
+        print_props(res)
 
-_, src = res.create_source()
-src.attach(ctx)
+        res.deallocate_wait(TIMEOUT_MS)
+        print("result: deallocate")
+        print_props(res)
 
-def print_props(res):
-    for prop in ('is-allocated', 'channel', 'bandwidth', 'generation'):
-        print('  {}: {}'.format(prop, res.get_property(prop)))
+    res.disconnect(handler_id_deallocated)
+    res.disconnect(handler_id_allocated)
+    src.destroy()
 
-for i in range(2):
-    print_props(res)
 
-    try:
-        res.allocate_wait([use_channel], use_bandwidth, 100)
-    except GLib.Error as e:
-        if e.matches(Hinoko.fw_iso_resource_error_quark(), Hinoko.FwIsoResourceError.EVENT):
-            print(e)
-    else:
-        print('result: allocate')
-        print_props(res)
+@contextmanager
+def run_dispatcher() -> AbstractContextManager[GLib.MainContext]:
+    ctx = GLib.MainContext.new()
+    dispatcher = GLib.MainLoop.new(ctx, False)
 
-    sleep(1)
+    th = Thread(target=lambda d: d.run(), args=(dispatcher,))
+    th.start()
 
     try:
-        res.deallocate_wait(100)
-    except GLib.Error as e:
-        if e.matches(Hinoko.fw_iso_resource_error_quark(), Hinoko.FwIsoResourceError.EVENT):
-            print(e)
-    else:
-        print('result: deallocate')
-        print_props(res)
+        yield ctx
+    finally:
+        dispatcher.quit()
+        th.join()
+
+
+def multithread_main(path: Path) -> int:
+    res_once = Hinoko.FwIsoResourceOnce.new()
+    res_once.open(str(path), 0)
 
-    sleep(1)
+    res_auto = Hinoko.FwIsoResourceAuto.new()
+    res_auto.open(str(path), 0)
 
-dispatcher.quit()
-th.join()
+    try:
+        with run_dispatcher() as ctx:
+            main_once(res_once, ctx)
+            main_auto(res_auto, ctx)
+    except GLib.Error as e:
+        error_domain_map = {
+            GLib.file_error_quark(): GLib.FileError,
+            Hinoko.fw_iso_resource_error_quark(): Hinoko.FwIsoResourceError,
+            Hinoko.fw_iso_resource_auto_error_quark(): Hinoko.FwIsoResourceAutoError,
+        }
+        quark = GLib.quark_from_string(e.domain)
+        if quark in error_domain_map:
+            code_nick = error_domain_map[quark](e.code).value_nick
+            print(
+                "GLib.Error exception: '{}' due to '{}' in '{}'".format(
+                    e.message, code_nick, e.domain
+                )
+            )
+            print()
+        traceback.print_exception(e)
+        return 1
+    except Exception as e:
+        traceback.print_exception(e)
+        return 1
+
+    return 0
+
+
+def main() -> int:
+    if len(argv) < 2:
+        print(
+            "One argument is required for path to special file of Linux FireWire character "
+            "device"
+        )
+        return 1
+
+    path = Path(argv[1])
+    if not path.exists() or not path.is_char_device() or path.name.find("fw") != 0:
+        print("{} is invalid for Linux FireWire character device".format(path))
+        return 1
+
+    return multithread_main(path)
+
+
+if __name__ == "__main__":
+    exit(main())