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

# SPDX-License-Identifier: MIT
# SPDX-FileCopyrightText: Copyright 2025 Sam Blenny
#
# See NOTES.md for documentation links and pinout info.
#
import board
from microcontroller import cpu
from micropython import const
import os
import time
import wifi

from sb_chardisplay import CharDisplay
from sb_ircbot import IRCBot


# ---------------------------------------------------------------------------
# Options for settings.toml
# - `WIFI_SSID = "your ssid"` set the SSID to use for your wifi
# - `WIFI_PASSWORD = "the password"` set password for your wifi
# - `IRC_SERVER = "<some IP address>"` set IP address for your IRC server
# - `IRC_NICK = "<nickname>"` set nickname to use for your IRC server
#
WIFI_SSID = None
if (val := os.getenv("WIFI_SSID")) is not None:
    WIFI_SSID = str(val)
WIFI_PASSWORD = None
if (val := os.getenv("WIFI_PASSWORD")) is not None:
    WIFI_PASSWORD = str(val)
IRC_SERVER = None
if (val := os.getenv("IRC_SERVER")) is not None:
    IRC_SERVER = str(val)
IRC_NICK = None
if (val := os.getenv("IRC_NICK")) is not None:
    IRC_NICK = str(val)
IRC_CHAN = None
if (val := os.getenv("IRC_CHAN")) is not None:
    IRC_CHAN = str(val)
# ---------------------------------------------------------------------------


def print_settings_banner():
    # Print startup banner showing if settings.toml is missing something
    heading = '# === IRC Dashboard Gadget Settings ==='
    m = 'missing'
    print()
    print(heading)
    print('# WIFI_SSID: [%s]' % (m if WIFI_SSID is None else 'ok'))
    print('# WIFI_PASSWORD: [%s]' % (m if WIFI_PASSWORD is None else 'ok'))
    print('# IRC_SERVER: [%s]' % (m if IRC_SERVER is None else 'ok'))
    print('# IRC_NICK: [%s]' % (m if IRC_NICK is None else 'ok'))
    print('# IRC_CHAN: [%s]' % (m if IRC_CHAN is None else 'ok'))
    print('# ' + ('=' * (len(heading)-2)))


def wifi_connect():
    # Try to connect to wifi
    if WIFI_SSID and WIFI_PASSWORD:
        try:
            wifi.radio.connect(ssid=WIFI_SSID, password=WIFI_PASSWORD)
            ip = wifi.radio.ipv4_address
            return ip
        except ConnectionError as e:
            print(e)
    return None


def run():
    # This has the main program logic. Putting these things inside a function
    # helps force me to avoid spaghetti code. When lots of stuff lives in the
    # global namespace, it's too easy to write functions that unintentionally
    # reference global variables.

    print_settings_banner()
    cd = CharDisplay()

    # 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

    # Wrap the radio stuff in a big retry loop
    RETRY_S = const(5)
    RETRY_S_MAX = const(180)
    WIFI_RETRY = const(20)
    READ_TIME = const(0.8) # give me time to read the screen
    irc_retry = RETRY_S
    while True:

        # Ensure Wifi is up
        if not wifi.radio.connected:
            while True:
                cd.set_status("WiFi Connect...")
                time.sleep(READ_TIME)
                ip = wifi_connect()
                if ip:
                    cd.set_status('WiFi IP is %s' % ip)
                    time.sleep(READ_TIME)
                    break
                else:
                    # Wifi problem: sleep for a bit then try again
                    cd.set_status('Wifi Down')
                    time.sleep(WIFI_RETRY)
                    continue
        else:
            cd.set_status("WiFi already connected")
            time.sleep(READ_TIME)

        # Ensure IRC is up, then start responding to IRC messages
        cd.set_status("IRC Connect...")
        time.sleep(READ_TIME)
        radio = wifi.radio
        with IRCBot(IRC_NICK, IRC_CHAN, IRC_SERVER, port=6667) as irc:
            while radio.connected and not irc.connected:
                # If initial connect failed, retry with exponential backoff
                time.sleep(irc_retry)
                irc_retry = min(RETRY_S_MAX, irc_retry * 2)
                cd.set_status('IRC Connect Retry...')
                time.sleep(READ_TIME)
                irc.connect()

            irc.register()

            # Main IRC bot loop
            last_rx = time.monotonic()
            PING_TIMEOUT = const(200)
            while radio.connected and irc.connected:
                line = irc.recv_line()
                if line is None:
                    if time.monotonic() - last_rx > PING_TIMEOUT:
                        # Something's wrong. It's too quiet. Close & reconnect
                        cd.set_status('IRC: connection timeout')
                        irc.close()
                        time.sleep(irc_retry)
                        irc_retry = min(RETRY_S_MAX, irc_retry * 2)
                        break
                    time.sleep(0.010)
                    continue
                last_rx = time.monotonic()
                (prefix, cmd, params) = line
                print(cmd, params)
                if cmd == '433':
                    # - 433 * <nick> :Nickname already in use
                    cd.set_status('IRC: Nick in use')
                    irc.close()
                    time.sleep(irc_retry)
                    irc_retry = min(RETRY_S_MAX, irc_retry * 2)
                    break
                elif cmd == 'PING':
                    irc.pong(params)
                elif cmd == 'JOIN':
                    my_nick = ':%s!' % IRC_NICK
                    if prefix and prefix.find(my_nick) == 0:
                        # Set status line to channel name when I join. Ignore
                        # join notifications about other users.
                        cd.set_status(params[1:])  # slice off leading ':'
                        # SUCCESS: Joined OK, so reset the retry interval
                        irc_retry = RETRY_S
                elif cmd == '332':
                    # Channel topic notification for JOIN
                    # Typical cmd+params format: `332 tftbot #sensors :!pre /`
                    nickchan = '%s %s :' % (IRC_NICK, IRC_CHAN)
                    pre = '!pre '
                    if not params.startswith(nickchan):
                        continue
                    text = params[len(nickchan):]
                    if text.startswith(pre):
                        # bot mode for displaying preformatted text with
                        # dynamically slectable line delimeters: first char of
                        # first word after the `!pre` is the delimeter that
                        # gets replaced with line breaks
                        text = text[len(pre):]
                        if len(text) >= 1:
                            delim = text[0]
                            text = text[1:].replace(delim, '\n')
                            cd.set_topic(text, wrap='pre')
                    else:
                        # Default to hard wrapping lines
                        cd.set_topic(text, wrap='hard')
                elif cmd == 'TOPIC':
                    # Channel topic notifiction after JOIN
                    # Typical cmd+param format: `TOPIC #sensors :!pre /...`
                    chan = '%s :' % IRC_CHAN
                    pre = '!pre '
                    if not params.startswith(chan):
                        continue
                    text = params[len(chan):]
                    if text.startswith(pre):
                        # bot mode for displaying preformatted text with
                        # dynamically slectable line delimeters: first char of
                        # first word after the `!pre` is the delimeter that
                        # gets replaced with line breaks
                        text = text[len(pre):]
                        if len(text) >= 1:
                            delim = text[0]
                            text = text[1:].replace(delim, '\n')
                            cd.set_topic(text, wrap='pre')
                    else:
                        # Default to hard wrapping lines
                        cd.set_topic(text, wrap='hard')

# ---
# Main entry point
# ---
run()

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

# SPDX-License-Identifier: MIT
# SPDX-FileCopyrightText: Copyright 2025 Sam Blenny
#
# See NOTES.md for documentation links and pinout info.
#
import atexit
import board
import displayio
from fourwire import FourWire
import terminalio

from adafruit_display_text import label
from adafruit_ili9341 import ILI9341


class CharDisplay:
    def __init__(self, width=16):
        # Try to initialize 2.8" TFT display shield with ILI9341 chip
        displayio.release_displays()
        spi = board.SPI()
        tft_cs = board.D10
        tft_dc = board.D9
        display_bus = FourWire(spi, command=tft_dc, chip_select=tft_cs)
        display = ILI9341(display_bus, width=320, height=240,
            rotation=180, auto_refresh=False)
        group = displayio.Group()
        display.root_group = group
        display.refresh()
        # System status textbox: scale=1, one line, very top of display
        status = label.Label(font=terminalio.FONT, scale=1, color=0x80ef00)
        status.anchor_point = (1, 0)
        status.anchored_position = (316, 4)
        status.line_spacing = 1.0  # default is 1.25
        group.append(status)
        # IRC topic textbox: scale=3, 6 lines, this is for IRC notifications
        topic = label.Label(font=terminalio.FONT, scale=3, color=0xefef00)
        topic.anchor_point = (0, 0)
        topic.anchored_position = (8, 16)
        topic.line_spacing = 1.0  # default is 1.25
        group.append(topic)
        # Set an atexit handler to release the display once code.py
        # ends. This is an aesthetic filter to prevent CircuitPython's
        # supervisor from hijacking the display to show its own stuff.
        def atexit_shutdown_display():
            try:
                status.text = 'OFFLINE'
                display.refresh()
                displayio.release_displays()
            except AttributeError:
                pass

        atexit.register(atexit_shutdown_display)
        self.display = display
        self.status = status
        self.topic = topic
        self.width = width

    def hard_wrap(self, text):
        w = self.width
        return '\n'.join([text[i:i+w] for i in range(0, len(text), w)])

    def word_wrap(self, text):
        # Format a string to fit on a narrow screen by word-wrapping at spaces
        w = self.width
        lines = []
        start = 0
        prev_space = 0
        for _ in range(len(text)):
            space = text.find(' ', start)
            if len(text) - start <= w:
                # All the remaining text fits in one line without breaking
                lines.append(text[start:])
                start = len(text)
                break
            elif space == -1 or space - start > w:
                # No space in the first width characters, so hard wrap
                lines.append(text[start:start+w])
                start += w
            else:
                # There's at least one space suitable for a line break
                for _ in range(len(text)):
                    next_space = text.find(' ', space + 1)
                    if next_space == -1:
                        break
                    elif next_space - start <= w + 1:
                        space = next_space
                    else:
                        break
                lines.append(text[start:space])
                start = space + 1
        if start < len(text):
            lines.append(text[start:])
        return '\n'.join(lines)

    def set_status(self, txt):
        # Show the message text the top status line
        print("status =", txt)
        self.status.text = txt
        self.display.refresh()

    def set_topic(self, txt, wrap=None):
        # Show the message text in the 6x16 character display IRC topic area
        print("topic =", txt)
        if wrap == 'hard':
            wrapped = self.hard_wrap(txt)
        elif wrap == 'word' or wrap is None:
            wrapped = self.word_wrap(txt)
        elif wrap == 'pre':
            wrapped = txt
        self.topic.text = wrapped
        self.display.refresh()

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

# SPDX-License-Identifier: MIT
# SPDX-FileCopyrightText: Copyright 2025 Sam Blenny
#
# See NOTES.md for documentation links and pinout info.
#
import re
import socketpool
import wifi


class IRCBot:
    def __init__(self, nick, chan, server, port=6667):
        self.rx_buf = bytearray(512)
        self.line_buf = bytearray(512)
        self.pool = socketpool.SocketPool(wifi.radio)
        self.sock = self.pool.socket()     # defaults to IP + TCP
        self.nick = nick
        self.chan = chan
        self.server = server
        self.port = port
        self.connected = False
        self.registered = False
        self.connect_timeout = 10
        self._recv_line_iterator = None
        self.irc_re = re.compile(
            br'^((:\S+)\s+)?'      # optional prefix   (match group 2)
            br'(\S+)\s*'           # command or number (match group 3)
            br'(.*)'               # params            (match group 4)
        )

    def close(self):
        self.sock.close()
        self.connected = False
        self.registered = False

    def connect(self):
        try:
            self.sock.settimeout(self.connect_timeout)
            self.sock.connect((self.server, self.port))
            self._recv_line_iterator = self._recv_line_gen()
            self.connected = True
        except OSError as e:
            # Exceptions I've seen trigger this (bad port, bad IP, etc):
            # - OSError: [Errno 104] ECONNRESET
            # - OSError: [Errno 116] ETIMEDOUT
            # - OSError: [Errno 118] EHOSTUNREACH
            # - OSError: [Errno 120] EALREADY
            # - OSError: [Errno 128] ENOTCONN
            print('ERR connect: "%s", errno=%d', e, e.errno)
            self.connected = False

    def register(self, timeout=30):
        if not self.connected:
            return False
        self.sock.settimeout(timeout)
        try:
            msg = (
                'NICK {0}\r\n'
                'USER {0} 0 * :{0}\r\n'
                'JOIN {1}\r\n'
                ).format(self.nick, self.chan)
            self.sock.sendall(msg.encode())
            # This might fail, but for now assume it worked
            self.registered = True
        except OSError as e:
            print('ERR register: "%s", errno=%d', e, e.errno)
            return False
        return True

    def _recv_line_gen(self, timeout=1):
        # This generator wraps Socket.recv_into() to provide line buffering
        if not self.connected:
            return None
        self.sock.settimeout(timeout)
        buf = self.rx_buf
        line = self.line_buf
        bufmv = memoryview(buf)
        linemv = memoryview(line)
        max_line = len(line)
        buf_start = 0
        line_end = 0
        while True:
            # Try to get some TCP bytes (maybe a full line, maybe less)
            try:
                size = self.sock.recv_into(buf)
            except OSError as e:
                # Exceptions I've seen trigger this:
                # - OSError: [Errno 116] ETIMEDOUT
                if e.errno == 116:
                    pass  # timeout is fine
                else:
                    print('ERR recv_into: "%s", errno=%d', e, e.errno)
                # ---
                yield None                    # no data yet -> yield nothing
                # ---
                continue
            # BEGIN full line parsing
            # Assume there may be multiple CRLF terminated lines in rx buffer
            # and that line buffer may already have some accumulated data...
            while (crlf := buf.find(b'\r\n', buf_start, size)) > -1:
                line_buf_space = max_line - line_end
                line_len = crlf - buf_start
                # Got a CRLF line ending
                if line_len < line_buf_space:
                    # Got newline and line will fit in the line buffer
                    line[line_end:line_end+line_len] = bufmv[buf_start:crlf]
                    line_end += line_len
                else:
                    # Got newline but line is too long, so truncate it
                    limit = min(0, max_line - line_end)
                    line[line_end:max_line] = bufmv[buf_start:buf_start+limit]
                    line_end = max_line
                # ---
                yield bytes(linemv[:line_end]).decode()       # Yield line
                # ---
                line_end = 0                       # Clear line buffer
                buf_start = crlf + 2               # Adjust rx buffer pointer
            if buf_start >= size:
                # RX buffer is empty, so start over with new packet
                buf_start = 0
                continue
            # END full line parsing
            # BEGIN partial line parsing
            line_buf_space = max_line - line_end
            part_len = size - buf_start
            if part_len < line_buf_space:
                # Partial line fits in line buffer -> copy it
                line[line_end:line_end+part_len] = bufmv[buf_start:size]
                line_end += part_len
            else:
                # Partial line is too long -> truncate it
                limit = min(0, max_line - line_end)
                line[line_end:max_line] = bufmv[buf_start:buf_start+limit]
                line_end += limit
            # Clear the RX buffer
            buf_start = 0
            # END partial line parsing

    def recv_line(self):
        # Return a line from the TCP stream buffer (includes auto-PONG)
        #
        # CAUTION!!! group(n) returns matches as strings, not type bytes.
        # If you want to use them as bytes, you'll need to do .encode().
        #
        it = self._recv_line_iterator
        if it and (line := next(it)):
            # Got a line, so parse it with regex match
            m = self.irc_re.match(line)
            prefix  = m.group(2)
            command = m.group(3)
            params  = m.group(4)
            return (prefix, command, params)
        return None

    def pong(self, params):
        # Send a PONG to keep the connection alive (params type is string)
        try:
            msg = 'PONG {}\r\n'.format(params)
            self.sock.sendall(msg.encode())
        except OSError as e:
            print('ERR pong: "%s", errno=%d', e, e.errno)

    # Context handlers

    def __enter__(self):
        self.connect()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()
        return False  # don't suppress exceptions