|
| 1 | +# This file is part of the msgpack-rpc module. |
| 2 | +# Copyright (c) 2023 Arduino SA |
| 3 | +# This Source Code Form is subject to the terms of the Mozilla Public |
| 4 | +# License, v. 2.0. If a copy of the MPL was not distributed with this |
| 5 | +# file, You can obtain one at https://mozilla.org/MPL/2.0/. |
| 6 | +# |
| 7 | +# MessagePack RPC protocol implementation for MicroPython. |
| 8 | +# https://github.com/msgpack-rpc/msgpack-rpc/blob/master/spec.md |
| 9 | + |
| 10 | +import logging |
| 11 | +import openamp |
| 12 | +import umsgpack |
| 13 | +from micropython import const |
| 14 | +from io import BytesIO |
| 15 | +from time import sleep_ms, ticks_ms, ticks_diff |
| 16 | + |
| 17 | +_MSG_TYPE_REQUEST = const(0) |
| 18 | +_MSG_TYPE_RESPONSE = const(1) |
| 19 | +_MSG_TYPE_NOTIFY = const(2) |
| 20 | + |
| 21 | + |
| 22 | +def log_level_enabled(level): |
| 23 | + return logging.getLogger().isEnabledFor(level) |
| 24 | + |
| 25 | + |
| 26 | +class Future: |
| 27 | + def __init__(self, msgid, msgbuf, fname, fargs): |
| 28 | + self.msgid = msgid |
| 29 | + self.msgbuf = msgbuf |
| 30 | + self.fname = fname |
| 31 | + self.fargs = fargs |
| 32 | + |
| 33 | + def join(self, timeout=0): |
| 34 | + if log_level_enabled(logging.INFO): |
| 35 | + logging.info(f"join {self.fname}()") |
| 36 | + |
| 37 | + if timeout > 0: |
| 38 | + t = ticks_ms() |
| 39 | + |
| 40 | + while self.msgid not in self.msgbuf: |
| 41 | + if timeout > 0 and ticks_diff(ticks_ms(), t) > timeout: |
| 42 | + raise Exception("Join timeout") |
| 43 | + sleep_ms(100) |
| 44 | + |
| 45 | + obj = self.msgbuf.pop(self.msgid) |
| 46 | + if obj[2] is not None: |
| 47 | + raise (OSError(obj[2])) |
| 48 | + |
| 49 | + if log_level_enabled(logging.DEBUG): |
| 50 | + logging.debug(f"call {self.fname}({self.fargs}) => {obj}") |
| 51 | + return obj[3] |
| 52 | + |
| 53 | + |
| 54 | +class MsgPackIO: |
| 55 | + def __init__(self): |
| 56 | + self.stream = BytesIO() |
| 57 | + |
| 58 | + def feed(self, data): |
| 59 | + offset = self.stream.tell() |
| 60 | + self.stream.write(data) |
| 61 | + self.stream.seek(offset) |
| 62 | + |
| 63 | + def readable(self): |
| 64 | + if self.stream.read(1): |
| 65 | + offset = self.stream.tell() |
| 66 | + self.stream.seek(offset - 1) |
| 67 | + return True |
| 68 | + return False |
| 69 | + |
| 70 | + def truncate(self): |
| 71 | + if self.readable(): |
| 72 | + offset = self.stream.tell() |
| 73 | + self.stream = BytesIO(self.stream.getvalue()[offset:]) |
| 74 | + |
| 75 | + def __iter__(self): |
| 76 | + return self |
| 77 | + |
| 78 | + def __next__(self): |
| 79 | + offset = self.stream.tell() |
| 80 | + try: |
| 81 | + obj = umsgpack.unpack(self.stream) |
| 82 | + self.truncate() |
| 83 | + return obj |
| 84 | + except Exception: |
| 85 | + self.stream.seek(offset) |
| 86 | + raise StopIteration |
| 87 | + |
| 88 | + |
| 89 | +class MsgPackRPC: |
| 90 | + def __init__(self, fwaddr=None): |
| 91 | + self.epts = [] |
| 92 | + self.msgid = 0 |
| 93 | + self.msgbuf = {} |
| 94 | + self.msgio = MsgPackIO() |
| 95 | + self.servers = [] |
| 96 | + |
| 97 | + # Initialize OpenAMP. |
| 98 | + openamp.init(ns_callback=self._bind_callback) |
| 99 | + |
| 100 | + # Keep a reference to the remote processor object, to stop the GC from collecting |
| 101 | + # it, which would call the finaliser and shut down the remote processor while it's |
| 102 | + # still being used. |
| 103 | + self.rproc = openamp.RProc(fwaddr) |
| 104 | + self.rproc.start() |
| 105 | + |
| 106 | + def _bind_callback(self, src_addr, name): |
| 107 | + if log_level_enabled(logging.INFO): |
| 108 | + logging.info(f'New service announcement src: {src_addr} name: "{name}"') |
| 109 | + self.epts.append(openamp.RPMsg(name, dst_addr=src_addr, callback=self._recv_callback)) |
| 110 | + self.epts[-1].send(b"\x00") |
| 111 | + |
| 112 | + def _recv_callback(self, src_addr, data): |
| 113 | + if log_level_enabled(logging.DEBUG): |
| 114 | + logging.debug(f"Received message on endpoint: {src_addr} data: {bytes(data)}") |
| 115 | + |
| 116 | + self.msgio.feed(data) |
| 117 | + for obj in self.msgio: |
| 118 | + if obj[0] == _MSG_TYPE_RESPONSE: |
| 119 | + self.msgbuf[obj[1]] = obj |
| 120 | + elif obj[0] == _MSG_TYPE_REQUEST: |
| 121 | + self._dispatch(obj[1], obj[2], obj[-1]) |
| 122 | + if log_level_enabled(logging.DEBUG): |
| 123 | + logging.debug(f"Unpacked {type(obj)} val: {obj}") |
| 124 | + |
| 125 | + def _send_msg(self, msgid, msgtype, fname, fargs, **kwargs): |
| 126 | + timeout = kwargs.pop("timeout", 1000) |
| 127 | + self.epts[msgtype].send(umsgpack.packb([msgtype, msgid, fname, fargs]), timeout=timeout) |
| 128 | + if msgtype == _MSG_TYPE_REQUEST: |
| 129 | + self.msgid += 1 |
| 130 | + return Future(msgid, self.msgbuf, fname, fargs) |
| 131 | + |
| 132 | + def _dispatch(self, msgid, fname, fargs): |
| 133 | + func = None |
| 134 | + retobj = None |
| 135 | + error = None |
| 136 | + for obj in self.servers: |
| 137 | + if callable(obj) and obj.__name__ == fname: |
| 138 | + func = obj |
| 139 | + elif hasattr(obj, fname): |
| 140 | + func = getattr(obj, fname) |
| 141 | + if func is not None: |
| 142 | + break |
| 143 | + |
| 144 | + if func is not None: |
| 145 | + retobj = func(*fargs) |
| 146 | + else: |
| 147 | + error = "Unbound function called %s" % (fname) |
| 148 | + |
| 149 | + self._send_msg(msgid, _MSG_TYPE_RESPONSE, error, retobj) |
| 150 | + |
| 151 | + def ready(self): |
| 152 | + return len(self.epts) == 2 |
| 153 | + |
| 154 | + def serve(self, obj): |
| 155 | + self.servers.append(obj) |
| 156 | + |
| 157 | + def call(self, fname, *args, **kwargs): |
| 158 | + return self.call_async(fname, *args, *kwargs).join() |
| 159 | + |
| 160 | + def call_async(self, fname, *args, **kwargs): |
| 161 | + return self._send_msg(self.msgid, _MSG_TYPE_REQUEST, fname, list(args), *kwargs) |
0 commit comments