1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

# SPDX-License-Identifier: MIT
# SPDX-FileCopyrightText: Copyright 2025 Sam Blenny
#
# See NOTES.md for documentation links and pinout info.
#
import board
import digitalio
import os
import time

# Record powerup time for calculating delay until I2C power is stable. The
# Feather ESP32-S3 board draws a big current spike when I2C_POWER gets
# turned on by the supervisor prior to loading boot.py on boot or wake from
# deep sleep. So, when running on a small lipo, we must allow time to
# recover from a possible brownout on the I2C bus power rail and pullups.
t0 = time.monotonic()

# ---------------------------------------------------------------------------
# Base Station Mode Override Option for settings.toml
#
# Put `BASE_MODE = 1` in settings.toml to force base station mode without
# regard to board_id. This gives me an easy way to use a Feather ESP32-S3
# sensor build as an extra base station (see also: `LCD_ENABLE = 0`).
#
BASE_MODE = False
if (base := os.getenv("BASE_MODE")) is not None:
    BASE_MODE = bool(base)
# ---------------------------------------------------------------------------


# Each time board resets or wakes from deep sleep, select mode from board_id:
if BASE_MODE or board.board_id == 'adafruit_feather_rp2350':
    import base_mode
    base_mode.run()

elif board.board_id == 'adafruit_feather_esp32s3_nopsram':
    # Toggle A1 to mark the start of code.py for the power profiler
    a1 = digitalio.DigitalInOut(board.A1)
    a1.switch_to_output(value=True)

    # start sensor mode
    import sensor_mode
    sensor_mode.run(a1, t0)

else:
    raise ValueError("Unexpected board_id. Halting. (check code.py)")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

# SPDX-License-Identifier: MIT
# SPDX-FileCopyrightText: Copyright 2025 Sam Blenny
#
# LoRa Sensor Mode (wireless thermometer on Feather ESP32-S3)
#
import alarm
from board import A0, A1, D9, D10, SCL, SDA, SPI
import busio
from digitalio import DigitalInOut
from microcontroller import cpu
from micropython import const
import os
import struct
import sys
import time

from adafruit_max1704x import MAX17048
from adafruit_mcp9808 import MCP9808

from common import HMAC_KEY, HMAC_TRUNC, LORA_NODE, encode_, rfm9x_factory
from sb_hmac import hmac_sha1


TARGET_INTERVAL = const(540.0)   # approximate seconds between transmits
WAKE_SECONDS    = const(1.963)   # wake runtime as measured by power profiler
BROWNOUT_DELAY  = const(0.300)   # time to wait for I2C bus brownout to clear
SENSOR_DELAY    = const(0.067)   # time to wait for I2C sensors to initialize
LOW_VOLTS       = const(3.450)   # threshold to begin power conservation


def run(a1, t0):
    # Initialize and run in remote sensor hardware configuration (LoRa TX)

    # Use GPIO outs to mark progress of run() for power analyzer logic inputs
    a0 = DigitalInOut(A0)
    a0.switch_to_output(value=True)
    a1.value = not a1.value

    # Reduce CPU frequency to save power during I2C and LoRa IO delays. The
    # default frequency is 240 MHz. To avoid messing up time.monotonic(), don't
    # attempt to set this below 80 MHz.
    if cpu.frequency == 240_000_000:
        cpu.frequency = 80_000_000

    # Configure LoRa radio while we wait for I2C power to settle
    cs, rst = DigitalInOut(D10), DigitalInOut(D9)
    rfm95 = rfm9x_factory(SPI(), cs, rst)
    a1.value = not a1.value

    # Make sure we've waited long enough since boot or wake. The MCP9808
    # can get confused if you talk to it too soon after switching on I2C_POWER
    # when running on a small lipo (likely due to brownout).
    elapsed = time.monotonic() - t0
    if (diff := BROWNOUT_DELAY - elapsed) > 0:
        time.sleep(diff)
    a1.value = not a1.value

    # Configure I2C sensors
    i2c = busio.I2C(SCL, SDA, frequency=250_000)  # default bus clock is slow
    try:
        mcp98 = MCP9808(i2c)      # temperature sensor
        mcp98.resolution = 1      # 0.25°C, 65ms conversion time
    except ValueError:
        # MCP9808() raises ValueError if it doesn't find sensor on I2C bus.
        # If that happens, carry on and report bad temperature values to the
        # base station. The point of this is to avoid the firmware draining
        # the battery if there's a hardware issue on the I2C bus.
        mcp98 = None
    max17 = MAX17048(i2c)     # battery fuel gauge
    a1.value = not a1.value

    # Allow sensors some time for first temperature conversion, etc.
    time.sleep(SENSOR_DELAY)
    a1.value = not a1.value

    # Get truncated Unix timestamp from RTC to use as message sequence number
    tstamp = time.time() & 0xffffffff

    # Check sensors then build packet with message (node address, timestamp,
    # volts, temperature) + truncated HMAC of message (modeled on TOTP)
    v = max17.cell_voltage                        # lipo cell volts
    max17.hibernate()                             # put it in low power mode
    if mcp98:
        f = (mcp98.temperature * 9/5) + 32        # °C -> °F
    else:
        # If sensor unplugged, send error value
        f = -127
    msg = encode_(LORA_NODE, tstamp, v, f)        # pack everything into bytes
    msg += hmac_sha1(HMAC_KEY, msg)[:HMAC_TRUNC]  # append truncated HMAC

    if v < LOW_VOLTS:
        # For low volts, skip transmit to avoid damaging LiPo cell
        pass
    else:
        # Send packet on LoRa radio with configurable repeats for better
        # reliability. You can modify the array of power levels to experiment
        # with tx power vs. range. Note that the node address is included in
        # the message for the HMAC, but that address byte gets sent as part of
        # the header rather than in the payload.
        print('%d, %08x, %.2f, %.1f' % (LORA_NODE, tstamp, v, f))
        for tx_pow in [23, 23]:    # tx power range is 5..23 dB, default 13
            a1.value = not a1.value
            rfm95.tx_power = tx_pow
            rfm95.send(msg[1:], node=msg[0])
            time.sleep(0.005)
        a1.value = not a1.value

    # Prepare peripherals and pins for low power
    rfm95.sleep()
    a0.value, a1.value = False, False

    # Calculate target sleep interval
    if v < LOW_VOLTS:
        # For low volts, sleep longer. This is speculatively intended to
        # possibly recover from temporary voltage sag due to extreme cold.
        t1 = 3600
    else:
        # Otherwise, use the normal interval
        t1 = TARGET_INTERVAL - WAKE_SECONDS

    # Begin deep sleep with ±2% random jitter added to target sleep interval
    t2 = t1 + (t1 * 0.02 * os.urandom(1)[0] / 255)
    alarm.exit_and_deep_sleep_until_alarms(
        alarm.time.TimeAlarm(monotonic_time=time.monotonic() + t2))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260

# SPDX-License-Identifier: MIT
# SPDX-FileCopyrightText: Copyright 2025 Sam Blenny
#
# LoRa Base Station
#
# See NOTES.md for related documentation
#
import board
from board import SCL, SDA, SPI
import busio
from collections import namedtuple
from digitalio import DigitalInOut
from microcontroller import cpu
import os
import struct
import time

from adafruit_character_lcd.character_lcd_i2c import Character_LCD_I2C

from common import HMAC_TRUNC, HMAC_KEY, decode_, rfm9x_factory
from sb_hmac import hmac_sha1

# ---------------------------------------------------------------------------
# Options for settings.toml
# - `LCD_BACKLIGHT = 1` enable character LCD backlight
# - `ESPNOW_GATEWAY = 1` retransmit LoRa packets as ESP-NOW packets
# - `ESPNOW_RX = 1` use ESP-NOW radio on builds that don't have a LoRa radio
#
LCD_BACKLIGHT = False
if (backlight := os.getenv("LCD_BACKLIGHT")) is not None:
    LCD_BACKLIGHT = bool(backlight)
ESPNOW_GATEWAY = False
if (gateway := os.getenv("ESPNOW_GATEWAY")) is not None:
    ESPNOW_GATEWAY = bool(gateway)
ESPNOW_RX = False
if (display_ := os.getenv("ESPNOW_RX")) is not None:
    ESPNOW_RX = bool(display_)
# ---------------------------------------------------------------------------


Report = namedtuple("Report",
        ["node_id", "timestamp", "volts", "temp_f", "rssi", "snr"])

class SensorReports:
    # This tracks reports from multiple sensors to format them for a 2x16 LCD
    def __init__(self):
        self.ready_timestamp = time.monotonic()
        self.lora_nodes = {}
        self.last_report = None

    def new_report(self, node_id, volts, temp_f, rssi, snr):
        r = Report(node_id, time.monotonic(), volts, temp_f, rssi, snr)
        self.lora_nodes[node_id] = r
        self.last_report = r

    def freshness_tag(self, seconds):
        minutes = round(max(0, seconds)) // 60
        hours = seconds // 3600   # 60*60=3600 seconds/hour
        days = seconds // 86400   # 60*60*24=86400 seconds/day
        if days > 0:
            return '%dd' % days
        elif hours > 0:
            return '%dh' % hours
        else:
            return '%dm' % minutes

    def __str__(self):
        # Format a two line status string for display on the 2x16 LCD
        now = time.monotonic()
        if len(self.lora_nodes) == 0:
            tag = self.freshness_tag(now - self.ready_timestamp)
            return 'Ready %s' % tag
        # Display latest report info formatted for 2x16 LCD
        r = self.last_report
        centivolts = round(r.volts * 100)
        tag = self.freshness_tag(now - r.timestamp)
        return '%d %d %d %s\n r%d s%.1f' % (
            r.node_id, r.temp_f, centivolts, tag, r.rssi, r.snr)


def run():
    # Initialize and run in base station hardware configuration (LoRa RX)

    # Print a banner showing which modes are active from settings.toml
    modes = ['=== Base Station']
    if ESPNOW_GATEWAY:
        modes.append(' + ESP-NOW Gateway')
    if ESPNOW_RX:
        modes.append(' + ESP-NOW Display')
    modes.append(' ===')
    banner = ''.join(modes)
    hr = '=' * len(banner)
    print()
    print(hr)
    print(banner)
    print(hr)

    # Reduce CPU frequency so the board runs cooler. The default ESP32-S3
    # default frequency is 240 MHz. To avoid messing up time.monotonic(), don't
    # attempt to set this below 80 MHz.
    if 'esp32s3' in board.board_id:
        cpu.frequency = 80_000_000

    # Try to initialize LoRa raido on SPI unless ESPNOW_RX option is turned on
    rfm95 = None
    if not ESPNOW_RX:
        cs, rst = DigitalInOut(board.D10), DigitalInOut(board.D9)
        rfm95 = rfm9x_factory(SPI(), cs, rst)
        rfm95.node = 255               # receive from all stations (nodes)

    # Try to initialize an I2C Character LCD. This is an optional feature
    # to assist with placing the antenna for good RSSI & SNR. The main output
    # for logging and charting goes over USB serial to a Raspberry Pi.
    lcd = None
    try:
        i2c = busio.I2C(SCL, SDA, frequency=250_000)  # default clock is slow
        cols, rows = 16, 2
        lcd = Character_LCD_I2C(i2c, cols, rows)
        lcd.clear()
        if LCD_BACKLIGHT:
            lcd.backlight = True
        lcd.message = 'Ready 0m'
    except RuntimeError as e:
        # This can happen on QT Py if you don't have pullups connected:
        # - RuntimeError: No pull up found on SDA or SCL; check your wiring
        pass
    except ValueError as e:
        # This happens if no I2C LCD backpack is present at 0x20
        pass

    # If one of the ESP-NOW modes is active, try to initialize wifi radio
    espnow_obj = None
    if ESPNOW_GATEWAY or ESPNOW_RX:
        import espidf
        import espnow
        import wifi
        # Force wifi radio onto channel 6 because esp-now needs that
        wifi.radio.start_ap(" ", "", channel=6, max_connections=0)
        wifi.radio.stop_ap()
        # Initialize the ESP-NOW object in broadcast mode
        espnow_obj = espnow.ESPNow()
        peer = espnow.Peer(mac=b'\xff\xff\xff\xff\xff\xff', channel=6)
        espnow_obj.peers.append(peer)

    # -----
    # BEGIN Receiver Loop
    # -----
    reports = SensorReports()
    EXPECTED_SIZE = 4 + 6 + HMAC_TRUNC    # size of header + message + MAC
    seq_list = {}
    TIMEOUT = 60
    while True:
        # Listen for message from ESP-NOW or LoRa radio, depending on settings
        if ESPNOW_RX:
            # Radio for RX is ESP-NOW
            # Since espnow.read() is buffered async, put it in a loop to make
            # it act like a blocking read in the style of RFM9x.receive()
            stop_time = time.monotonic() + TIMEOUT
            data = None
            while time.monotonic() < stop_time:
                if len(espnow_obj) > 1:
                    packet = espnow_obj.read()
                    data = packet.msg
                    espnow_rssi = packet.rssi
                    # ESP-NOW packets have int16_t rssi and snr at the end of
                    # the payload so the ESP-NOW receiver can see how good the
                    # signal was at the LoRa receiver on the gateway
                    if len(data) == EXPECTED_SIZE + 4:
                        i = EXPECTED_SIZE
                        # pop the LoRa rssi and snr values
                        snr = struct.unpack('>h', data[i+2:i+4])[0]
                        rssi = struct.unpack('>h', data[i:i+2])[0]
                        data = data[:EXPECTED_SIZE]
                    else:
                        rssi = packet.rssi
                        snr = 0
                    break
                else:
                    time.sleep(0.01)
        else:
            # Radio for RX is LoRa
            data = rfm95.receive(with_header=True, timeout=TIMEOUT)
            rssi = rfm95.last_rssi                   # signal strength
            snr = rfm95.last_snr                     # signal to noise ratio

        # Handle the packet (check length, verify HMAC, etc).
        # Data should be None for timeout or a bytearry of header+packet.
        if data:
            # Skip wrong-size packets
            if EXPECTED_SIZE != len(data):
                continue

            # Re-assemble message including the node address byte from header.
            # Message format: node address, sequence number, volts, °F.
            msg = data[1:2] + data[4:-HMAC_TRUNC]
            msg_hash = data[-HMAC_TRUNC:]
            node, seq, v, f = decode_(msg)
            # Verify HMAC of message (configure nodes to share the same key)
            if msg_hash != hmac_sha1(HMAC_KEY, msg)[:HMAC_TRUNC]:
                continue

            # Check for monotonic sequence number (per node)
            # CAUTION: After boot, this accepts the first sequence number it
            # sees for each node address (values don't persist across reset)
            seq_check = False
            prev_seq = seq_list.get(node)
            if (prev_seq is None) or (prev_seq < seq):
                seq_check = True
                seq_list[node] = seq

            # Now packet is authenticated (HMAC) but possibly a duplicate.
            # Print decoded packet regardless, as it can be useful to see
            # RSSI and SNR for retry packets.
            check_tag = "OK" if seq_check else "DUP"
            radio_tag = "ESPNOW" if ESPNOW_RX else "LORA"
            print('%s: %d, %.1f, %d, %08x, %.2f, %.0f, %s' %
                (radio_tag, rssi, snr, node, seq, v, f, check_tag))

            # -----
            # END receive section
            # BEGIN display / repeater / gateway section
            # -----

            # Don't do any of the stuff below for duplicate packets that failed
            # the monotonic sequence number check
            if not seq_check:
                continue

            # Update the reports data structure
            reports.new_report(node, v, f, rssi, snr)

            # If there's a character LCD available, show report message
            if lcd:
                lcd.clear()
                lcd.message = str(reports)

            # If ESPNOW_GATEWAY mode is enabled, check the hop count then
            # retransmit the packet if hops are low enough.
            to    = data[0]
            from_ = data[1]
            id_   = data[2]
            hops  = data[3] & 0x0f  # Low 4 bits of flags field of LoRa header
            max_hops = 1
            # Slice payload out of header+payload
            payload = data[4:]
            # Sush RSSI & SNR at end of payload
            payload += struct.pack('>hh', max(-256, min(0, round(rssi))),
                max(-256, min(256, round(snr))))
            if ESPNOW_GATEWAY and seq_check and hops < max_hops:
                new_hop = min(16, hops+1)
                enow_msg = bytearray([to, from_, id_, new_hop]) + payload
                try:
                    espnow_obj.send(enow_msg, peer)
                except Exception as e:
                    print('E', e)  # Log the error and keep going

        else:
            # For receive timeout, update report age tags on LCD
            if lcd:
                lcd.message = str(reports)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

# SPDX-License-Identifier: MIT
# SPDX-FileCopyrightText: Copyright 2025 Sam Blenny
#
# See NOTES.md for related documentation
#
from micropython import const
import os
import struct

from adafruit_rfm9x import RFM9x


# -------------------------------------------------------------------------
# Config Options

# Sensor Float Range Limits (for encoding LoRa messages)
BATT_LO = const(3.2)
BATT_HI = const(4.4)
TEMP_LO = const(-128)
TEMP_HI = const(127)

HMAC_TRUNC = const(4)    # size in bytes of truncated hash (similar to HOTP)

# Defaults for optional settings.toml variables
HMAC_KEY = b'Please set your own HMAC_KEY in settings.toml'
LORA_NODE = 0     # LoRa node address (range 0..254)

# Read settings.toml
if key := os.getenv("HMAC_KEY"):
    HMAC_KEY = key
if lora_node := os.getenv("LORA_NODE"):
    LORA_NODE = int(lora_node) & 0xff
# -------------------------------------------------------------------------


def rfm9x_factory(spi, cs, rst):
    # Configure an RFM9x object for the 900 MHz RFM95W LoRa FeatherWing.
    r = RFM9x(spi, cs, rst, 915.0, baudrate=5_000_000, agc=False, crc=False)
    r.tx_power = 13          # range 5..23 dB, default: 13
    r.spreading_factor = 10  # default: 7
    r.preamble_length = 16   # range uint16, default: 8
    r.coding_rate = 8
    return r

def scale_to_byte(val, lo, hi):
    # Scale and convert float in range lo..hi to integer in range 0..255
    return min(255, max(0, round(255 * (val - lo) / (hi - lo))))

def scale_from_byte(b, lo, hi):
    # Scale and convert integer in range 0..255 to float in range lo..hi
    return (b * (hi - lo) / 255) + lo

def encode_(node, count_, volt, degree_F):
    # Compress message counter and float measurements into bytes value
    v = scale_to_byte(volt, BATT_LO, BATT_HI)
    t = scale_to_byte(degree_F, TEMP_LO, TEMP_HI)
    return struct.pack('>BLBB', node & 0xff, count_, v, t)

def decode_(data):
    # Expand bytes into float measurements with correct range
    node, count_, v_byte, t_byte = struct.unpack('>BLBB', data)
    v = scale_from_byte(v_byte, BATT_LO, BATT_HI)
    t = scale_from_byte(t_byte, TEMP_LO, TEMP_HI)
    return node, count_, v, t

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

# SPDX-License-Identifier: MIT
# SPDX-FileCopyrightText: Copyright 2024 Sam Blenny
#
# Related Docs:
# - https://datatracker.ietf.org/doc/html/rfc2104  (HMAC RFC)
# - https://nvlpubs.nist.gov/nistpubs/fips/nist.fips.198-1.pdf (FIPS PUB 198-1)
#
import hashlib


def hmac(K, text, hash_fn, B, L):
    # Return HMAC digest for key and text using the named hash algorithm.
    #
    # The non-pythonic parameter names correspond to the notation used in
    # RFC 2104 and FIPS 198-1. The point of doing it this way is to make
    # it easier to compare this implementation against the specs.
    # Params:
    #   K: secret key (UTF-8 string or bytes)
    #   text: hash function input data (UTF-8 string or bytes)
    #   hash_fn: name of hash function for use with hashlib.new()
    #   B: block size in bytes of hash function input
    #   L: byte-length of hash function output
    # FIPS 198-1 Crypto Math Notation:
    #   ⨁: plus sign in a circle means bitwise XOR
    #   ||: two vertical bars means concatenate bitstreams

    # Convert text from string to bytes if needed
    if isinstance(text, str):
        text = text.encode('utf-8')
    # Convert key from string to bytes if needed
    if isinstance(K, str):
        K = K.encode('utf-8')
    # Prepare key as block sized byte buffer (hash or pad as needed)
    if len(K) == B:
        K0 = K
    elif len(K) > B:
        H = hashlib.new('sha1')
        H.update(K)
        K0 = H.digest() + (b'\x00' * (B - L))
    elif len(K) < B:
        K0 = K + (b'\x00' * (B - len(K)))
    # Calculate inner hash: H((K₀ ⨁ ipad)||text)
    K0_xored = bytearray([b ^ 0x36 for b in K0])
    H = hashlib.new('sha1')
    H.update(K0_xored)
    H.update(text)
    inner_hash = H.digest()
    # Calculate outer hash: H((K₀ ⨁ opad)||inner_digest)
    K0_xored = bytearray([b ^ 0x5c for b in K0])
    H = hashlib.new('sha1')
    H.update(K0_xored)
    H.update(inner_hash)
    return H.digest()


def hmac_sha1(key, data):
    # HMAC-SHA1 digest for key and data
    return hmac(K=key, text=data, hash_fn='sha1', B=64, L=20)