Bluetooth LE with aioble, and ESP-NOW

ble/gatt_temp.py

"""A temperature sensor as a Bluetooth LE peripheral: it advertises the Environmental Sensing service, and notifies
its Temperature characteristic when the reading has changed by 0.1 degree or more. A phone connects at 3 s."""
import struct

import mpy_aioble as aioble  # on a board: import aioble
import mpy_asyncio as asyncio  # on a board: import asyncio
import mpy_bluetooth as bluetooth  # on a board: import bluetooth
import mpy_time as time  # on a board: import time

ENV_SENSING = bluetooth.UUID(0x181A)  # Environmental Sensing service
TEMPERATURE = bluetooth.UUID(0x2A6E)  # Temperature characteristic
GENERIC_THERMOMETER = 0x0300  # appearance
READINGS = [23.51, 23.53, 23.58, 23.71, 23.74, 23.69, 24.02, 24.05, 23.80, 23.82]  # one every 2 s


def encode(temp_c):
    """sint16 in hundredths of a degree, least significant byte first, as the GATT Specification Supplement says."""
    return struct.pack("<h", round(temp_c * 100))


service = aioble.Service(ENV_SENSING)
temperature = aioble.Characteristic(service, TEMPERATURE, read=True, notify=True)
aioble.register_services(service)


def phone_got(value):
    print("%5d ms  phone   notified: %s = %.2f degrees" % (time.ticks_ms(), value.hex(" "), struct.unpack("<h", value)[0] / 100))


phone = aioble.bench_phone(3000, phone_got)


async def sensor():
    last = None
    for reading in READINGS:
        if phone.connected and (last is None or abs(reading - last) >= 0.1):
            temperature.write(encode(reading), send_update=True)
            last = reading
        else:
            temperature.write(encode(reading))  # kept for a read, but no notification
        await asyncio.sleep_ms(2000)


async def peripheral():
    await aioble.advertise(250_000, name="plant-1", services=[ENV_SENSING], appearance=GENERIC_THERMOMETER)
    print("%5d ms  board   a central connected" % time.ticks_ms())


async def main():
    asyncio.create_task(peripheral())
    await sensor()
    print("%d readings, %d notifications of 2 bytes each; a read now gives %.2f degrees" % (
        len(READINGS), temperature.sent, struct.unpack("<h", temperature.read())[0] / 100))


asyncio.run(main())

advertising every 250 ms, 20 bytes: 02 01 06 03 03 1a 18 08 09 70 6c 61 6e 74 2d 31 03 19 00 03
 3000 ms  board   a central connected
 4030 ms  phone   notified: 36 09 = 23.58 degrees
 6030 ms  phone   notified: 43 09 = 23.71 degrees
12030 ms  phone   notified: 62 09 = 24.02 degrees
16030 ms  phone   notified: 4c 09 = 23.80 degrees
10 readings, 4 notifications of 2 bytes each; a read now gives 23.82 degrees

mpy_aioble.py (the model) · ble/mpy_aioble.py

"""A model of aioble, from the micropython-lib that MicroPython 1.29 ships, for a peripheral that advertises one
service and notifies one characteristic, on the asyncio model of mpy_asyncio.py.

What it keeps from aioble/peripheral.py and aioble/server.py of that version:
  - advertise() builds the advertising data from its arguments, each field as its length, its type and its data: the
    flags, then the 16-bit service UUIDs, the name and the appearance; a field that does not fit the 31 bytes of the
    advertising data goes to the scan response, and one that fits neither raises ValueError("Advertising payload
    too long"); advertise() returns the connection once a central has connected;
  - Characteristic(service, uuid, read=True, notify=True) keeps a value; write(data, send_update=True) stores it and
    sends a notification to each connected central that subscribed, write(data) only stores it;
  - register_services() comes before advertising.
The radio is the bench's: bench_phone() is a central that connects at a set time, subscribes, and receives each
notification 30 ms after it is sent. No simulator models the air, so range, timing and loss need a board.
Not MicroPython: a model for computers.
"""
import struct

import mpy_asyncio as asyncio
import mpy_time as time

_MAX = 31
_phone = None
_services = []


class Service:
    def __init__(self, uuid):
        self.uuid = uuid
        self.characteristics = []


class Characteristic:
    def __init__(self, service, uuid, read=False, write=False, notify=False, indicate=False, initial=None):
        self.uuid, self.notify = uuid, notify
        self._value = initial or b""
        self.sent = 0  # notifications sent, for the bench
        service.characteristics.append(self)

    def read(self):
        return self._value

    def write(self, data, send_update=False):
        self._value = bytes(data)
        if send_update and self.notify and _phone is not None and _phone.subscribed:
            self.sent += 1
            value = self._value
            time.at(time.ticks_ms() + 30, lambda: _phone.on_notify(value))


def register_services(*services):
    _services.extend(services)


def _append(adv, resp, kind, value):
    field = struct.pack("BB", len(value) + 1, kind) + value
    if len(adv) + len(field) < _MAX:
        adv += field
        return resp
    if len(resp) + len(field) < _MAX:
        resp += field
        return resp
    raise ValueError("Advertising payload too long")


class _Connection:
    async def disconnected(self, timeout_ms=None):
        await asyncio.sleep_ms(10 ** 9 if timeout_ms is None else timeout_ms)


async def advertise(interval_us, name=None, services=None, appearance=0, timeout_ms=None):
    adv, resp = bytearray(), bytearray()
    resp = _append(adv, resp, 0x01, b"\x06")  # flags: general discoverable, no classic Bluetooth
    if services:
        resp = _append(adv, resp, 0x03, b"".join(bytes(u) for u in services))  # complete list of 16-bit UUIDs
    if name:
        resp = _append(adv, resp, 0x09, name.encode())  # complete local name
    if appearance:
        resp = _append(adv, resp, 0x19, struct.pack("<H", appearance))
    bench_advertising(bytes(adv), bytes(resp), interval_us)
    while _phone is None or _phone.at > time.ticks_ms():
        await asyncio.sleep_ms(10)
    _phone.connected = True
    return _Connection()


def bench_advertising(adv, resp, interval_us):
    print("advertising every %d ms, %d bytes: %s" % (interval_us // 1000, len(adv), adv.hex(" ")))
    if resp:
        print("scan response, %d bytes: %s" % (len(resp), resp.hex(" ")))


class _Phone:
    def __init__(self, at, on_notify):
        self.at, self.on_notify = at, on_notify
        self.connected = False
        self.subscribed = True  # it subscribes as soon as it connects


def bench_phone(at_ms, on_notify):
    global _phone
    _phone = _Phone(at_ms, on_notify)
    return _phone

mpy_bluetooth.py (the model) · ble/mpy_bluetooth.py

"""bluetooth.UUID of MicroPython 1.29, as far as the programs of this lesson use it: a 16-bit UUID whose bytes are
sent least significant byte first, as Bluetooth sends every field. Not MicroPython: a model for computers."""


class UUID:
    def __init__(self, value):
        self.value = value

    def __bytes__(self):
        return self.value.to_bytes(2, "little")

    def __repr__(self):
        return "UUID(0x%04x)" % self.value

mpy_asyncio.py (the model) · ble/mpy_asyncio.py

"""A model of MicroPython 1.29's asyncio, on the simulated clock of mpy_time.py.

It follows extmod/asyncio of v1.29.0 (core.py, funcs.py, event.py, lock.py) wherever a program can see it:
  - one loop runs one task at a time, and a task gives the loop back only at an await that has to wait;
  - sleep_ms() and sleep() return one shared object, not a coroutine, so that a sleep allocates nothing; create_task(),
    gather() and wait_for() refuse it with TypeError("coroutine expected");
  - run() returns as soon as its own task ends; the other tasks stay queued, nothing cancels them, and a later run()
    goes on with them; new_event_loop() empties the queue;
  - a task that ends with an exception while nothing waits for it prints "Task exception wasn't retrieved" (a board
    prints it to standard error, with the traceback) and the other tasks go on;
  - CancelledError derives from BaseException and TimeoutError from Exception; wait_for() cancels the awaitable when
    its time is up and waits for it to finish; gather() returns the results in the order of its arguments, or raises
    the first exception while the other tasks go on;
  - Event.set() wakes every waiting task; Lock hands itself to the next waiting task; ThreadSafeFlag is one bit that
    an interrupt handler may set and wait() clears, with one waiting task.
The clock moves on only while no task can run, or inside a blocking call (mpy_time.sleep_ms()), so every run prints
the same times. Not MicroPython: a model for computers.
"""
import atexit
import heapq
import inspect

import mpy_time as _time


class CancelledError(BaseException):
    pass


class TimeoutError(Exception):
    pass


_queue = []  # (due, order, task, value): tasks that run once the clock reaches due
_order = 0
_ended = 0
_current = None
_polled = []  # objects the loop polls before each task: a ThreadSafeFlag, a stream with a reader waiting
_ERRNO = {19: "ENODEV", 104: "ECONNRESET", 110: "ETIMEDOUT", 111: "ECONNREFUSED", 113: "EHOSTUNREACH"}



atexit.register(lambda: _close_unstarted([entry[2] for entry in _queue]))


class _Request:
    """What a task hands the loop when it has to wait."""

    def __init__(self, *what):
        self.what = what

    def __await__(self):
        return (yield self.what)


class Task:
    """A coroutine that the loop runs. create_task() makes one; awaiting it waits for its end."""

    def __init__(self, coro):
        self.coro = coro
        self.name = getattr(coro, "__name__", "coro")
        self._done = False
        self._result = None
        self._error = None
        self._ended = 0
        self._waiters = []  # tasks waiting for this one to end
        self._throw = None  # raised inside the task when it next runs: cancel()
        self._value = None  # what the loop sends in when it next runs
        self._joined = None  # the task it awaits, to which cancel() passes on
        self._token = 0  # its live entry in the queue
        self._undo = []  # what to undo when it stops waiting
        self._watched = 0  # gather() or wait_for() collects its result

    def done(self):
        return self._done

    def cancel(self):
        """Raises CancelledError inside the task at the await where it waits. A task that awaits another task passes
        the cancel on to that one, as in MicroPython, and receives its CancelledError when it ends."""
        if self._done:
            return False
        if self is _current:
            raise RuntimeError("can't cancel self")
        target = self
        while target._joined is not None:
            target = target._joined
        target._throw = CancelledError()
        _wake(target)
        return True

    def __await__(self):
        if not self._done:
            yield ("join", self)
        if self._error is not None:
            raise self._error
        return self._result

    def __repr__(self):
        return "<Task>"


def _push(task, due, value=None):
    global _order
    _order += 1
    task._token = _order
    heapq.heappush(_queue, (due, _order, task, value))


def _unpark(task):
    task._joined = None
    undo, task._undo = task._undo, []
    for step in undo:
        step()


def _wake(task, value=None):
    _unpark(task)
    _push(task, _time._now, value)


def _drop(waiting, task):
    if task in waiting:
        waiting.remove(task)


def _unpoll(obj, task):
    if obj._waiter is task:
        obj._waiter = None
    if obj in _polled and obj._waiter is None:
        _polled.remove(obj)


def _park(task, what):
    kind = what[0]
    if kind == "sleep":
        _push(task, what[1])
    elif kind in ("join", "join-any", "join-or-time"):
        others = what[1] if kind == "join-any" else [what[1]]
        for other in others:
            other._waiters.append(task)
            task._undo.append(lambda other=other: _drop(other._waiters, task))
        if kind == "join":
            task._joined = what[1]
        if kind == "join-or-time":
            _push(task, what[2], False)  # still waiting at that time: the time is up
    elif kind == "wait":  # an Event or a Lock: its list of waiting tasks
        what[1].append(task)
        task._undo.append(lambda: _drop(what[1], task))
    elif kind == "poll":  # one slot for a waiting task: a second waiter takes the first one's place
        obj = what[1]
        obj._waiter = task
        if obj not in _polled:
            _polled.append(obj)
        task._undo.append(lambda: _unpoll(obj, task))
    else:
        raise RuntimeError("the model cannot wait for %r" % (what,))


def _text(error):
    if isinstance(error, OSError) and error.args and isinstance(error.args[0], int):
        return "[Errno %d] %s" % (error.args[0], _ERRNO.get(error.args[0], ""))
    return str(error)


def _finish(task, result=None, error=None):
    global _ended
    _ended += 1
    task._done, task._result, task._error, task._ended = True, result, error, _ended
    waiters, task._waiters = task._waiters, []
    for waiter in list(waiters):
        _wake(waiter, True)
    if error is not None and not waiters and not task._watched and not isinstance(error, CancelledError):
        print("Task exception wasn't retrieved")
        print("future: <Task> coro= <generator object '%s'>" % task.name)
        print("%s: %s" % (type(error).__name__, _text(error)))


def _step(task):
    global _current
    _current = task
    try:
        if task._throw is not None:
            error, task._throw = task._throw, None
            what = task.coro.throw(error)
        else:
            what = task.coro.send(task._value)
    except StopIteration as stop:
        _finish(task, result=stop.value)
        return
    except BaseException as error:  # CancelledError too: the task has ended
        _finish(task, error=error)
        return
    finally:
        _current = None
    _park(task, what)


def _poll():
    for obj in list(_polled):
        if obj._waiter is not None and obj._ready():
            _wake(obj._waiter)


def _run_until_complete(main):
    while True:
        _poll()
        if _queue and _queue[0][2]._token != _queue[0][1]:
            heapq.heappop(_queue)  # an entry left behind when the task was woken another way
            continue
        if not _queue or _queue[0][0] > _time._now:
            due = _queue[0][0] if _queue else None
            nxt = _time.next_event()
            if due is None and nxt is None:
                raise RuntimeError("model: every task waits for something that never comes")
            _time._advance(min(t for t in (due, nxt) if t is not None))
            continue
        _, _, task, value = heapq.heappop(_queue)
        task._token = 0
        _unpark(task)
        task._value = value
        _step(task)
        if task is main and main._done:
            if main._error is not None:
                raise main._error
            return main._result


def _promote(aw):
    return aw if isinstance(aw, Task) else create_task(aw)


class _Sleep:
    """What sleep_ms() and sleep() return: one object for every sleep, as in MicroPython. It is not a coroutine."""

    def __init__(self):
        self.due = None

    def __await__(self):
        due, self.due = self.due, None
        yield ("sleep", due)


_sleep = _Sleep()


def sleep_ms(t):
    _sleep.due = _time._now + max(0, int(t))
    return _sleep


def sleep(t):
    return sleep_ms(int(t * 1000))


def create_task(coro):
    if not hasattr(coro, "send"):
        raise TypeError("coroutine expected")
    task = Task(coro)
    _push(task, _time._now)
    return task


def current_task():
    if _current is None:
        raise RuntimeError("no running event loop")
    return _current


def run(coro):
    return _run_until_complete(create_task(coro))


def _close_unstarted(tasks):
    # A coroutine that never started holds no code to run; closing it keeps CPython from warning about it, which a
    # board never does. Coroutines that started are left as they are.
    for task in tasks:
        if inspect.iscoroutine(task.coro) and inspect.getcoroutinestate(task.coro) == inspect.CORO_CREATED:
            task.coro.close()


def new_event_loop():
    """Empties the loop: tasks still queued from an earlier run() are forgotten."""
    global _queue, _polled
    _close_unstarted([entry[2] for entry in _queue])
    _queue, _polled = [], []


async def gather(*aws, return_exceptions=False):
    tasks = [_promote(aw) for aw in aws]
    for t in tasks:
        t._watched += 1
    try:
        while True:
            failed = [t for t in tasks if t._done and t._error is not None]
            if failed and not return_exceptions:
                raise min(failed, key=lambda t: t._ended)._error
            running = [t for t in tasks if not t._done]
            if not running:
                return [t._result if t._error is None else t._error for t in tasks]
            try:
                await _Request("join-any", running)
            except CancelledError:
                for t in running:
                    t.cancel()
                raise
    finally:
        for t in tasks:
            t._watched -= 1


async def _wait_for(aw, timeout):
    task = _promote(aw)
    if timeout is None:
        return await task
    task._watched += 1
    try:
        try:
            ended = await _Request("join-or-time", task, _time._now + max(0, timeout))
        except CancelledError:
            task.cancel()
            raise
        if not ended:
            task.cancel()
            try:
                await task
            except BaseException:  # its CancelledError, or whatever its clean-up raised
                pass
            raise TimeoutError
        return await task
    finally:
        task._watched -= 1


def wait_for(aw, timeout):
    """Returns a coroutine, as in MicroPython: its argument is promoted to a task only when it runs."""
    return _wait_for(aw, None if timeout is None else int(timeout * 1000))


def wait_for_ms(aw, timeout):
    return _wait_for(aw, timeout)


class Event:
    def __init__(self):
        self._set = False
        self._waiting = []

    def is_set(self):
        return self._set

    def set(self):
        """Wakes every waiting task. Call it from a task only, never from an interrupt or another thread."""
        waiting, self._waiting = self._waiting, []
        for task in list(waiting):  # waking a task takes it out of the list
            _wake(task)
        self._set = True

    def clear(self):
        self._set = False

    async def wait(self):
        if not self._set:
            await _Request("wait", self._waiting)
        return True


class ThreadSafeFlag:
    def __init__(self):
        self._state = 0
        self._waiter = None

    def set(self):
        """Safe from an interrupt handler: it only sets a bit, which the loop polls before each task."""
        self._state = 1

    def clear(self):
        self._state = 0

    def _ready(self):
        return self._state == 1

    async def wait(self):
        if not self._state:
            await _Request("poll", self)
        self._state = 0


class Lock:
    def __init__(self):
        self._state = 0  # 0 free, 1 held, or the task it has been handed to
        self._waiting = []

    def locked(self):
        return self._state == 1

    def release(self):
        if self._state != 1:
            raise RuntimeError("Lock not acquired")
        if self._waiting:
            task = self._waiting.pop(0)
            self._state = task
            _wake(task)
        else:
            self._state = 0

    async def acquire(self):
        if self._state != 0:
            try:
                await _Request("wait", self._waiting)
            except CancelledError:
                if self._state is _current:  # handed to this task, then cancelled: pass it on
                    self._state = 1
                    self.release()
                raise
        self._state = 1
        return True

    async def __aenter__(self):
        return await self.acquire()

    async def __aexit__(self, *exc):
        self.release()



def __getattr__(name):
    """Loads the streams the first time a program asks for them, as MicroPython's asyncio loads its stream module."""
    if name in ("open_connection", "start_server", "StreamReader", "StreamWriter"):
        import mpy_stream

        return getattr(mpy_stream, name)
    raise AttributeError(name)

mpy_time.py (the model) · ble/mpy_time.py

"""The time functions of MicroPython 1.29 that asyncio uses, on a simulated clock (extmod/modtime.c).

asyncio, machine.Timer and these functions share one clock of whole milliseconds, as they share time.ticks_ms() on a
board. ticks_ms() wraps at 2**30 on the rp2 and esp32 ports, like these functions; ticks_diff() and ticks_add() work
across the wrap. sleep_ms() and sleep() block: the clock moves on and nothing else runs except interrupts (the hard
timer callbacks of machine.py), as a blocking call holds a board's whole program. Not MicroPython: a model for
computers. at() is the bench's own: something that happens at a set time, an interrupt or a packet arriving.
"""
import heapq

TICKS_PERIOD = 1 << 30
_now = 0  # milliseconds since the bench started
_events = []  # (time, order, function): what the bench makes happen at a set time
_order = 0


def ticks_ms():
    return _now % TICKS_PERIOD


def ticks_add(ticks, delta):
    return (ticks + delta) % TICKS_PERIOD


def ticks_diff(end, start):
    half = TICKS_PERIOD // 2
    return ((end - start + half) % TICKS_PERIOD) - half


def at(when, function):
    """The bench calls function() when the clock reaches when (a time in milliseconds since the start)."""
    global _order
    _order += 1
    heapq.heappush(_events, (when, _order, function))


def next_event():
    return _events[0][0] if _events else None


def _advance(to):
    """Let the clock run to `to`, firing every interrupt and delivery due on the way, in time order."""
    global _now
    while _events and _events[0][0] <= to:
        when, _, function = heapq.heappop(_events)
        _now = max(_now, when)
        function()
    _now = max(_now, to)


def sleep_ms(ms):
    """Blocks for ms milliseconds: no task runs meanwhile, only interrupts."""
    if ms > 0:
        _advance(_now + ms)


def sleep(seconds):
    sleep_ms(int(seconds * 1000))


def reset():
    global _now, _events, _order
    _now, _events, _order = 0, [], 0


def bench_busy(ms):
    """The bench's own: ms milliseconds of work that never gives the loop back (a computation, a blocking driver)."""
    sleep_ms(ms)

espnow/espnow_send.py

"""What send() tells a sensor about four receivers: a gateway that reads, a gateway whose program has stopped, a
board on another channel, and nobody at all for a broadcast."""
import mpy_espnow as espnow  # on a board: import espnow, after network.WLAN(network.WLAN.IF_STA).active(True)
import mpy_time as time  # on a board: import time

SENSOR, GATEWAY, STOPPED, FAR = b"\x02\x00\x00\x00\x00\x01", b"\x02\x00\x00\x00\x00\x02", b"\x02\x00\x00\x00\x00\x03", b"\x02\x00\x00\x00\x00\x04"
BROADCAST = b"\xff" * 6
espnow.bench_board(SENSOR, channel=6)
gateway = espnow.bench_board(GATEWAY, channel=6)
stopped = espnow.bench_board(STOPPED, channel=6)  # its Wi-Fi is on, but its program has crashed
espnow.bench_board(FAR, channel=11)  # joined to an access point on channel 11

e = espnow.ESPNow(SENSOR)
e.active(True)
for peer in (GATEWAY, STOPPED, FAR, BROADCAST):
    e.add_peer(peer)

for name, mac in (("gateway that reads", GATEWAY), ("gateway that stopped", STOPPED), ("board on channel 11", FAR), ("broadcast", BROADCAST)):
    print("send() to the %-21s returned %s" % (name, e.send(mac, b"moisture=41")))
time.sleep_ms(10)

g = espnow.ESPNow(GATEWAY)
mac, msg = g.recv(100)
print("the gateway that reads received %s from %s" % (msg, mac.hex(":")))
print("the stopped gateway holds %d unread packet(s), and nothing will ever read them" % len(stopped.inbox))
print("espnow.MAX_DATA_LEN on this bench: %d bytes (ESP-NOW v1; a board with v2 reports 1470)" % espnow.MAX_DATA_LEN)

send() to the gateway that reads    returned True
send() to the gateway that stopped  returned True
send() to the board on channel 11   returned False
send() to the broadcast             returned True
the gateway that reads received b'moisture=41' from 02:00:00:00:00:01
the stopped gateway holds 2 unread packet(s), and nothing will ever read them
espnow.MAX_DATA_LEN on this bench: 250 bytes (ESP-NOW v1; a board with v2 reports 1470)

mpy_espnow.py (the model) · espnow/mpy_espnow.py

"""A model of MicroPython 1.29's espnow module on the ESP32, for the programs of this lesson.

What it keeps from the espnow documentation of v1.29.0:
  - active(True) needs an active WLAN interface first, station or access point, but no access point to join;
  - add_peer(mac) registers a peer, at most 20 of them and at most 6 with encryption; a broadcast address must be added
    too; MAX_DATA_LEN is 250 bytes on ESP-NOW v1 and 1470 on v2, which 1.29 brought to the ESP32;
  - send(mac, msg) returns True when the peer's radio acknowledged the packet, which a peer does when its Wi-Fi
    interface is active on the same channel, whether or not its program reads anything; a broadcast always returns True;
  - recv(timeout_ms) returns [mac, msg], or (None, None) when the time runs out;
  - the receive buffer, rxbuf, holds two packets of the largest size by default; packets that arrive when it is full
    are dropped, and are still acknowledged to the sender.
The air is the bench's: bench_board() puts a board on a channel; a packet arrives 2 ms after it is sent.
Not MicroPython: a model for computers.
"""
import mpy_time as _time

MAX_DATA_LEN = 250
MAX_TOTAL_PEER_NUM = 20
MAX_ENCRYPT_PEER_NUM = 6
_boards = {}  # mac: _Board


class _Board:
    def __init__(self, mac, channel, rx_packets):
        self.mac, self.channel, self.rx_packets = mac, channel, rx_packets
        self.wifi_active = True
        self.inbox = []
        self.dropped = 0


def bench_board(mac, channel=1, rx_packets=2):
    """A board on the air: its MAC address, its channel, and how many packets its receive buffer holds."""
    board = _Board(mac, channel, rx_packets)
    _boards[mac] = board
    return board


class ESPNow:
    def __init__(self, mac):
        self._me = _boards[mac]  # on a board, ESPNow() takes no argument: the bench says which board this is
        self._peers = []
        self._active = False

    def active(self, flag=None):
        if flag is None:
            return self._active
        self._active = bool(flag)

    def add_peer(self, mac):
        if len(self._peers) >= MAX_TOTAL_PEER_NUM:
            raise OSError("ESP_ERR_ESPNOW_FULL")
        self._peers.append(mac)

    def send(self, mac, msg, sync=True):
        if len(msg) > MAX_DATA_LEN:
            raise ValueError("ESP_ERR_ESPNOW_ARG")
        if mac not in self._peers:
            raise OSError("ESP_ERR_ESPNOW_NOT_FOUND")
        if mac == b"\xff" * 6:
            for board in _boards.values():
                if board is not self._me and board.channel == self._me.channel:
                    self._deliver(board, bytes(msg))
            return True
        peer = _boards.get(mac)
        if peer is None or not peer.wifi_active or peer.channel != self._me.channel:
            return False
        self._deliver(peer, bytes(msg))
        return True  # the radio acknowledged it, kept or not

    def _deliver(self, board, msg):
        sender = self._me.mac

        def arrive():
            if len(board.inbox) >= board.rx_packets:
                board.dropped += 1  # dropped, though the sender was told it arrived
            else:
                board.inbox.append((sender, msg))

        _time.at(_time._now + 2, arrive)

    def recv(self, timeout_ms=300_000):
        waited = 0
        while not self._me.inbox:
            if waited >= timeout_ms:
                return (None, None)
            _time.sleep_ms(1)
            waited += 1
        return list(self._me.inbox.pop(0))

mpy_time.py (the model) · espnow/mpy_time.py

"""The time functions of MicroPython 1.29 that these programs use, on a simulated clock (extmod/modtime.c).

ticks_ms() wraps at 2**30 on the rp2 and esp32 ports, like these functions; ticks_diff() and ticks_add() work across
the wrap. sleep_ms() blocks: the clock moves on and nothing else runs, as on a board. at() is the bench's own: it
makes something happen at a set time (the radio finishing a step of a join). bench_stop_at() ends a program that
would wait for ever, so that the page can show how long it had waited. Not MicroPython: a model for computers.
"""
import heapq

TICKS_PERIOD = 1 << 30
_now = 0  # milliseconds since the bench started
_events = []  # (time, order, function)
_order = 0
_stop = None  # the bench stops a program that is still waiting at this time


class BenchStopped(Exception):
    """Raised by sleep_ms() when the clock passes bench_stop_at(): the program would still be waiting."""


def ticks_ms():
    return _now % TICKS_PERIOD


def ticks_add(ticks, delta):
    return (ticks + delta) % TICKS_PERIOD


def ticks_diff(end, start):
    half = TICKS_PERIOD // 2
    return ((end - start + half) % TICKS_PERIOD) - half


def at(when, function):
    global _order
    _order += 1
    heapq.heappush(_events, (when, _order, function))


def _advance(to):
    global _now
    while _events and _events[0][0] <= to:
        when, _, function = heapq.heappop(_events)
        _now = max(_now, when)
        function()
    _now = max(_now, to)


def sleep_ms(ms):
    if _stop is not None and _now + ms > _stop:
        _advance(_stop)
        raise BenchStopped(_stop)
    if ms > 0:
        _advance(_now + ms)


def sleep(seconds):
    sleep_ms(int(seconds * 1000))


def reset():
    global _now, _events, _order, _stop
    _now, _events, _order, _stop = 0, [], 0, None


def bench_stop_at(ms):
    global _stop
    _stop = ms