anecdata/espnow_receiver.py

# SPDX-FileCopyrightText: 2023 anecdata
#
# SPDX-License-Identifier: MIT

import time
import traceback
import random
import supervisor
import os
import rtc
import espnow
import espidf
import wifi
import socketpool
import ssl
import adafruit_ntp
import adafruit_requests

from sekrets import *


### ESPNOW Sender
TZ_DEFAULT = -5
URL = "http://wifitest.adafruit.com/testwifi/index.html"
RCVR_CH = 0  # channel 1 (unless connected to an AP or acting as an AP)


def struct_time_to_iso_time():
    st = time.localtime()
    tz = TZ_DEFAULT
    return f"{st[0]:04d}-{st[1]:02d}-{st[2]:02d}T{st[3]:02d}:{st[4]:02d}:{st[5]:02d}{tz:+03}:00"

def connect():
    try:
        wifi.radio.connect(os.getenv("WIFI_SSID"), os.getenv("WIFI_PASSWORD"), bssid=AP_BSSID)  # use AP on channel <> 1
        time.sleep(1)  # wait for ap_info
        print(f"{struct_time_to_iso_time()} ipv4={wifi.radio.ipv4_address} channel={wifi.radio.ap_info.channel} rssi={wifi.radio.ap_info.rssi}")
    except ConnectionError as ex:
        traceback.print_exception(ex, ex, ex.__traceback__)

def ntp_to_rtc():
    wifi.radio.enabled = True
    connect()
    try:
        ntp = adafruit_ntp.NTP(pool, tz_offset=TZ_DEFAULT)
        rtc.RTC().datetime = ntp.datetime
        print(f"{struct_time_to_iso_time()} RTC time set with NTP time")
    except Exception as e:
        traceback.print_exception(e, e, e.__traceback__)
    wifi.radio.enabled = False  # lose the wifi channel

def do_monitor():
    c = random.randrange(1, 14)
    print(f"{'-'*25}\n{struct_time_to_iso_time()} Monitor starting on channel {c}...")
    m = wifi.Monitor(channel=c)
    time.sleep(1)
    received = m.packet()
    if received != {}:
        print(f"ch=qlen={m.queued()} lost={m.lost()} len={received[wifi.Packet.LEN]} ch={received[wifi.Packet.CH]} rssi={received[wifi.Packet.RSSI]}")
        print(f"raw={received[wifi.Packet.RAW]}")
    m.deinit()

def do_wifi():
    print(f"{'-'*25}\n{struct_time_to_iso_time()} Wi-Fi starting...")
    wifi.radio.enabled = True
    connect()
    try:
        with requests.get(URL) as r:
            print(f"{struct_time_to_iso_time()} {r.status_code} {r.reason.decode()} {r.headers}")
    except Exception as ex:
        traceback.print_exception(ex, ex, ex.__traceback__)
    wifi.radio.enabled = False  # lose the wifi channel

def do_espnow():
    print(f"{'-'*25}\n{struct_time_to_iso_time()} ESPNow sender starting...")
    with espnow.ESPNow() as e:
        e.set_pmk(SNDR_PMK)

        peers_report = ""
        for peer in peers:
            e.peers.append(peer)
            peers_report += f"mac={peer.mac} lmk={peer.lmk} ch={peer.channel} if={peer.interface} enc={peer.encrypted}\n"

        print(f"{struct_time_to_iso_time()} Sending...", end=" ")
        try:
            e.send(struct_time_to_iso_time().encode())
        except espidf.IDFError as ex:  # ESP-NOW error 0x306a
            traceback.print_exception(ex, ex, ex.__traceback__)
            supervisor.reload()
        print(f"send=[{e.send_success} {e.send_failure}] read=[{e.read_success} {e.read_failure}] buf={e.buffer_size} phy={e.phy_rate} peers={peers_report}", end="")


time.sleep(3)  # wait for serial
print(f"{'='*25}")
pool = socketpool.SocketPool(wifi.radio)
context = ssl.create_default_context()
requests = adafruit_requests.Session(pool, context)
ntp_to_rtc()

peers = [espnow.Peer(mac=RCVR_MAC, lmk=RCVR_LMK, encrypted=True, channel=RCVR_CH),]
while True:
    do_monitor()
    time.sleep(1)
    do_wifi()
    time.sleep(1)
    do_espnow()
    time.sleep(3)

    try:
        e.send(f"{struct_time_to_iso_time().encode()}")

        delivery_time = 0
        start = time.monotonic_ns()
        while not e.send_success and not e.send_failure:
            if (delivery_time := time.monotonic_ns() - start) > SEND_TIMEOUT:
                break

Adafruit CircuitPython 8.1.0-beta.0-80-g22636e056 on 2023-03-29; Adafruit Feather ESP32-S2 TFT with ESP32S2
>>> import time
>>> import espnow
>>> from sekrets import *
>>> 
>>> RCVR_CH = 0 
>>> 
>>> e = espnow.ESPNow()
>>> e.set_pmk(SYS_PMK)
>>> peer = espnow.Peer(mac=RCVR_MAC, lmk=LMK_181, encrypted=True, channel=RCVR_CH)
>>> e.peers.append(peer)
>>> 
>>> e.send(f"{'#'*250}")
>>> time.sleep(0.1)  # delivery confirmation takes non-zero time
>>> print(f"send=[{e.send_success} {e.send_failure}] read=[{e.read_success} {e.read_failure}]")
send=[1 0] read=[1 0]
>>> 
>>> e.send(f"{'#'*251}")
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
espidf.IDFError: ESP-NOW error 0x306a
>>> time.sleep(0.1)  # delivery confirmation takes non-zero time
>>> print(f"send=[{e.send_success} {e.send_failure}] read=[{e.read_success} {e.read_failure}]")
send=[1 0] read=[1 0]
>>> 
>>> e.send(f"{'#'*250}")
>>> time.sleep(0.1)  # delivery confirmation takes non-zero time
>>> print(f"send=[{e.send_success} {e.send_failure}] read=[{e.read_success} {e.read_failure}]")
send=[2 0] read=[1 0]
>>>