Avid Adlenarine asynchronous protocol

avid-trx.py

#!/usr/bin/env python3

# SPDX-License-Identifier: GPL-3.0-or-later
# Copyright (c) 2020 Takashi Sakamoto

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

from threading import Thread, RLock, Condition

CMD_OFFSET = 0xfffff0e00000
RESP_OFFSET = 0xfffff0e00800
FRAME_SIZE = 0x800 # Approximately.
TIMEOUT_MS = 100

node = Hinawa.FwNode.new()
node.open('/dev/fw1')

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

while not dispatcher.is_running():
    pass

class AvidProto(Hinawa.FwResp):
    def __init__(self):
        super().__init__()
        self.cv = Condition(RLock())
        self.node = None
        self.cv = None
        self.resp_frame = None
    def bind(self, node):
        self.reserve(node, RESP_OFFSET, FRAME_SIZE)
        self.node = node
        self.cv = Condition(RLock())
    def unbind(self):
        self.release()
        self.node = None
        self.cv = None
    def transaction(self, cmd):
        req = Hinawa.FwReq.new()
        self.resp_frame = None
        req.transaction_sync(self.node, Hinawa.FwTcode.WRITE_BLOCK_REQUEST, CMD_OFFSET, len(cmd), cmd, TIMEOUT_MS)
        with self.cv:
            res = None
            while self.resp_frame == None:
                res = self.cv.wait(TIMEOUT_MS)
        if not res:
            raise IOError
        resp_frame = self.resp_frame
        self.resp_frame = None
        return resp_frame
    def do_requested2(self, rcode, offset, src, dst, card, generation, frame, length):
        with self.cv:
            self.resp_frame = bytearray(frame)
            self.cv.notify()
        return Hinawa.FwRcode.COMPLETE

proto = AvidProto()
proto.bind(node)

# For example.
resp = proto.transaction([0x0b, 0x0b, 0x00, 0x06, 0x40, 0x06, 0x00, 0x00, 0x00, 0x01])
# Expected: 0b 0b 00 04 40 06 00 00
for i, b in enumerate(resp):
    print('{:2d}: {:02x}'.format(i, b))

proto.unbind()

dispatcher.quit()
th.join()