CircuitPython utility coroutines

statusbar.py

from asyncio import sleep

import displayio
import terminalio
from adafruit_display_text import label
from wifi import radio


async def maintain_status_bar(canvas: displayio.Group, anchor_point=(0, 0), anchored_position=(0, 0)):
    """
    Coroutine for maintaining a status bar at the top of the screen.

    `anchor_point` and `anchored_position` are passed to the label.Label. By default, it anchors to the top left corner.
    """

    status = label.Label(
        font=terminalio.FONT,
        text="Initializing...",
        color=0xFFFFFF,
    )
    status.anchor_point = anchor_point
    status.anchored_position = anchored_position
    canvas.append(status)

    while True:
        wifi_status = "Not connected"
        if radio.ap_info:
            wifi_status = f"{radio.ap_info.ssid} {radio.ipv4_address} {radio.ap_info.rssi}"

        status.text = f"{wifi_status}"
        await sleep(0)

wifi_management.py

"""
Wi-Fi Connection Manager
"""

from asyncio import sleep
from time import monotonic

from wifi import radio


class ScanNetworks():
    """
    Context manager for scanning for WiFi networks.

    The context variable is an iterable of available `wifi.Network`s as produced by
    `wifi.radio.start_scanning_networks()`.
    """

    active = False

    def __enter__(self):
        if ScanNetworks.active:
            raise RuntimeError(
                "Cannot start a second scan while one is already running!")
        ScanNetworks.active = True

        return radio.start_scanning_networks()

    def __exit__(self, *args):
        radio.stop_scanning_networks()

        ScanNetworks.active = False


async def network_manager(known_networks: dict[str, str], scan_frequency: int = 30):
    """
    Coroutine for keeping the device connected to the strongest known network.

    `known_networks`: A dictionary mapping SSIDs to passwords.
    """

    last_network_scan = monotonic()

    while True:
        await sleep(0)
        if radio.ap_info and monotonic() - last_network_scan < scan_frequency:
            continue

        with ScanNetworks() as detected_networks_iter:
            detected_networks = list(filter(
                lambda network: network.ssid in known_networks.keys(), detected_networks_iter))
            detected_networks.sort(
                key=lambda network: network.rssi, reverse=True)

            for network in detected_networks:
                if radio.ap_info and radio.ap_info.ssid == network.ssid:
                    break

                try:
                    radio.connect(
                        ssid=network.ssid,
                        password=known_networks[network.ssid],
                        channel=network.channel
                    )
                    break
                except ConnectionError:
                    pass

        if radio.ap_info != None:
            last_network_scan = monotonic()