MQTT unable to connect and subscribe to messages

[prcutler]

Troubleshooting this with Tyeth and Justin in Discord. Using Justin's latest on his branch at https://github.com/justmobilize/Adafruit_CircuitPython_MiniMQTT/tree/no-retry-on-unauthorized.

I have a PyPortal Titano with latest Nina Firmware and running absolute latest CircuitPython: Adafruit CircuitPython 9.1.0-beta.2-6-ge0f745c14c on 2024-05-18; Adafruit PyPortal Titano with samd51j20

I have a program that listens for messages and then downloads an image when a message is received, however it never is able to connect, it fails on the line mqtt_client.loop(35) and I have MQTT set up as:

mqtt_client = MQTT.MQTT(
    broker=getenv("broker"),
    username=getenv("ADAFRUIT_AIO_USERNAME"),
    password=getenv("ADAFRUIT_AIO_KEY"),
    port=1883,
    socket_timeout=30,
    recv_timeout=35,
    socket_pool=pool,
    ssl_context=ssl_context,
    is_ssl=False,
)

# Setup the callback methods above
mqtt_client.on_connect = connected
mqtt_client.on_disconnect = disconnected
mqtt_client.on_message = message

# Connect the client to the MQTT broker.
mqtt_client.logger = logger

mqtt_client.connect()

while True:
    # Poll the message queue
    try:
        mqtt_client.loop(35)

    except RuntimeError or ConnectionError:
        time.sleep(10)
        mqtt_client.connect()
        mqtt_client.loop()

    # Send a new message
    time.sleep(5)

and the error is:

4761.377: DEBUG - Attempting to connect to MQTT broker (attempt #0)
4761.379: DEBUG - Attempting to establish MQTT connection...
4761.447: DEBUG - Sending CONNECT to broker...
4761.449: DEBUG - Fixed Header: bytearray(b'\x10>')
4761.453: DEBUG - Variable Header: bytearray(b'\x00\x04MQTT\x04\xc2\x00<')
4761.598: DEBUG - Receiving CONNACK packet from broker
4765.758: DEBUG - Got message type: 0x20 pkt: 0x20
Subscribing to prcutler/feeds/albumart
4765.764: DEBUG - Sending SUBSCRIBE to broker...
4765.768: DEBUG - Fixed Header: bytearray(b'\x82\x1c')
4765.789: DEBUG - Variable Header: b'\x00\x01'
4765.809: DEBUG - SUBSCRIBING to topic prcutler/feeds/albumart with QoS 0
4765.813: DEBUG - payload: b'\x00\x17prcutler/feeds/albumart\x00'
4765.908: DEBUG - Got message type: 0x90 pkt: 0x90
4765.920: DEBUG - Resetting reconnect backoff
4765.922: DEBUG - waiting for messages for 35 seconds
Traceback (most recent call last):
  File "/lib/adafruit_minimqtt/adafruit_minimqtt.py", line 993, in _wait_for_msg
  File "/lib/adafruit_minimqtt/adafruit_minimqtt.py", line 1083, in _sock_exact_recv
  File "adafruit_esp32spi/adafruit_esp32spi_socketpool.py", line 183, in recv_into
timeout: timed out

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "code.py", line 240, in <module>
  File "/lib/adafruit_minimqtt/adafruit_minimqtt.py", line 968, in loop
  File "/lib/adafruit_minimqtt/adafruit_minimqtt.py", line 998, in _wait_for_msg
TypeError: function takes 3 positional arguments but 2 were given

[prcutler]

Update after testing this afternoon with Tyeth and @justmobilize (who asked to be tagged).

CircuitPython version: absolute latest - adafruit-circuitpython-pyportal_titano-en_US-20240518-main-PR9247-e0f745c.uf2
Nina firmware version: 1.7.7

The bug according to Justin is that it's raising a string "timed out" instead of errno.ETIMEDOUT (though we're not sure if this is a CircuitPython bug or Nina firmware bug)

It looks like the root cause is running out of memory, removing a lot of the display code and setting CIRCUITPY_PYSTACK_SIZE=3072 in settings.toml got it to run.

[justmobilize]

@dhalbert any ideas why the timeout error would be a string?

[justmobilize]

Here was the code that was run.

@prcutler could you uncomment all the display code and let me know if it fails or not?

# SPDX-License-Identifier: MIT

import sys
is_microcontroller = sys.implementation.name == "circuitpython"

#from adafruit_pyportal import PyPortal
import adafruit_connection_manager
import adafruit_requests
#import displayio
import json
#import terminalio
#from adafruit_display_text import label
import adafruit_minimqtt.adafruit_minimqtt as MQTT
import time
import adafruit_logging as logging


def getenv(value):
    if is_microcontroller:
        from os import getenv as cp_getenv
        return cp_getenv(value)
    else:
        import tomllib
        with open("settings.toml", "rb") as toml_file:
            settings = tomllib.load(toml_file)
        return settings[value]


secrets = {
    "ssid": getenv("CIRCUITPY_WIFI_SSID"),
    "password": getenv("CIRCUITPY_WIFI_PASSWORD"),
    "broker": getenv("broker"),
    "user": getenv("ADAFRUIT_AIO_USERNAME"),
    "pass": getenv("ADAFRUIT_AIO_KEY"),
}
if secrets == {"ssid": None, "password": None}:
    try:
        # Fallback on secrets.py until depreciation is over and option is removed
        from secrets import secrets
    except ImportError:
        print("WiFi secrets are kept in settings.toml, please add them there!")
        raise

IMAGE_URL = "https://silversaucer.com/static/img/album-art/image_300.bmp"

# pyportal = PyPortal()
# pyportal.network.connect()

# Set up WIFI w/ConnectionManager

if is_microcontroller:
    import board
    import busio
    from digitalio import DigitalInOut
    from adafruit_esp32spi import adafruit_esp32spi

    esp32_cs = DigitalInOut(board.ESP_CS)
    esp32_ready = DigitalInOut(board.ESP_BUSY)
    esp32_reset = DigitalInOut(board.ESP_RESET)

    # Secondary (SCK1) SPI used to connect to WiFi board on Arduino Nano Connect RP2040
    if "SCK1" in dir(board):
        spi = busio.SPI(board.SCK1, board.MOSI1, board.MISO1)
    else:
        spi = busio.SPI(board.SCK, board.MOSI, board.MISO)
    esp = adafruit_esp32spi.ESP_SPIcontrol(spi, esp32_cs, esp32_ready, esp32_reset)
else:
    esp = adafruit_connection_manager.CPythonNetwork()

pool = adafruit_connection_manager.get_radio_socketpool(esp)
ssl_context = adafruit_connection_manager.get_radio_ssl_context(esp)
requests = adafruit_requests.Session(pool, ssl_context)

# Set up logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

if is_microcontroller:
    print("Connecting to AP...")
    while not esp.is_connected:
        try:
            esp.connect_AP(secrets["ssid"], secrets["password"])
        except OSError as e:
            print("could not connect to AP, retrying: ", e)
            continue
    print("Connected to", str(esp.ssid, "utf-8"), "\tRSSI:", esp.rssi)
    print("sleeping for 10 seconds to make sure wifi is good...")
    time.sleep(10)
else:
    print("Using CPython sockets...")

# Get the Album title and artist name from JSON
data_source = "https://silversaucer.com/album/data"

try:
    resp = requests.get(data_source)
    data = resp.json()
    print(data)

except RuntimeError:
    resp = requests.get(data_source)
    data = resp.json()
    print(data)

image_location = data["image_url"]
artist = data["artist"]
album = data["album"]

album_info = artist + " - " + album

# album_info = "Garbage - Garbage"

# Load image on disk and display it on first boot
#display = board.DISPLAY
#display.rotation = 90

#winamp = displayio.OnDiskBitmap(open("winamp256.bmp", "rb"))

# Create a TileGrid to hold the bitmap
#tile_grid_1 = displayio.TileGrid(winamp, pixel_shader=winamp.pixel_shader, x=0)

# Create a Group to hold the TileGrid
#group = displayio.Group()

# Add the TileGrid to the Group
#group.append(tile_grid_1)

#album_art = displayio.OnDiskBitmap("albumart.bmp")

#tile_grid_2 = displayio.TileGrid(album_art, pixel_shader=album_art.pixel_shader, y=120)
#group.append(tile_grid_2)

#font = terminalio.FONT
#color = 0x00E200

#text_area = label.Label(font, text=album_info, color=color)

#text_area.x = 130
#text_area.y = 42
#group.append(text_area)

# Add the Group to the Display
#display.root_group = group

# ------------- MQTT Topic Setup ------------- #
mqtt_topic = "prcutler/feeds/albumart"
#mqtt_topic = "justmobilize/feeds/albumart"


### Code ###
# Define callback methods which are called when events occur
# pylint: disable=unused-argument, redefined-outer-name
def connected(client, userdata, flags, rc):
    # This function will be called when the client is connected
    # successfully to the broker.
    print("Subscribing to %s" % (mqtt_topic))
    client.subscribe(mqtt_topic)


def disconnected(client, userdata, rc):
    # This method is called when the client is disconnected
    print("Disconnected from MQTT Broker!")


def message(client, topic, message):
    """Method callled when a client's subscribed feed has a new
    value.
    :param str topic: The topic of the feed with a new value.
    :param str message: The new value
    """
    if message == "New album picked!":
        print("New message on topic {0}: {1}".format(topic, message))
        response = None

        url = "https://silversaucer.com/static/img/album-art/image_300.bmp"

        response = requests.get(url)
        if response.status_code == 200:
            print("Starting image download...")
            with open("albumart.bmp", "wb") as f:
                for chunk in response.iter_content(chunk_size=32):
                    f.write(chunk)
                print("Album art saved")
            response.close()

            resp = requests.get(data_source)
            data = resp.json()
            print(data)

            # There's a few different places we look for data in the photo of the day
            image_location = data["image_url"]
            artist = data["artist"]
            album = data["album"]

            album_str = artist + " - " + album

            album_info = album_str[:30]

            #winamp = displayio.OnDiskBitmap(open("winamp256.bmp", "rb"))

            # Create a TileGrid to hold the bitmap
            #tile_grid_1 = displayio.TileGrid(winamp, pixel_shader=winamp.pixel_shader)

            # Create a Group to hold the TileGrid
            #group = displayio.Group()

            # Add the TileGrid to the Group
            #group.append(tile_grid_1)

            #album_art = displayio.OnDiskBitmap("albumart.bmp")

            #tile_grid_2 = displayio.TileGrid(album_art, pixel_shader=album_art.pixel_shader, y=120)
            #group.append(tile_grid_2)

            #font = terminalio.FONT
            #color = 0x00E200

            #text_area = label.Label(font, text=album_info, color=color)

            #text_area.x = 130
            #text_area.y = 42
            #group.append(text_area)

            # Add the Group to the Display
            #display.root_group = group

            time.sleep(10)

        else:
            print("Bad get request")


# Initialize MQTT interface with the esp interface
# pylint: disable=protected-access
# MQTT.set_socket(socket, pyportal.network._wifi.esp)

# Set up a MiniMQTT Client
mqtt_client = MQTT.MQTT(
    broker=getenv("broker"),
    username=getenv("ADAFRUIT_AIO_USERNAME"),
    password=getenv("ADAFRUIT_AIO_KEY"),
    port=1883,
    socket_timeout=5,
    recv_timeout=10,
    socket_pool=pool,
    ssl_context=ssl_context,
    is_ssl=False,
)

# Setup the callback methods above
mqtt_client.on_connect = connected
mqtt_client.on_disconnect = disconnected
mqtt_client.on_message = message

# Connect the client to the MQTT broker.
mqtt_client.logger = logger

mqtt_client.connect()

while True:
    # Poll the message queue
    try:
        mqtt_client.loop(5)

    except RuntimeError or ConnectionError:
        time.sleep(10)
        mqtt_client.connect()
        mqtt_client.loop()

    # Send a new message
    time.sleep(5)

[prcutler]

Thank you for the reminder to add my code. Here it is with the display stuff uncommented out.

It appears to be working, though it does crash from time to time, but nothing I can measure, it appears to be random. I apologize in advance for any ugliness in my code. :)

Thanks again @justmobilize for all the troubleshooting.

# SPDX-License-Identifier: MIT

import sys
is_microcontroller = sys.implementation.name == "circuitpython"

#from adafruit_pyportal import PyPortal
import adafruit_connection_manager
import adafruit_requests
import displayio
import json
import terminalio
from adafruit_display_text import label
import adafruit_minimqtt.adafruit_minimqtt as MQTT
import time
import adafruit_logging as logging


def getenv(value):
    if is_microcontroller:
        from os import getenv as cp_getenv
        return cp_getenv(value)
    else:
        import tomllib
        with open("settings.toml", "rb") as toml_file:
            settings = tomllib.load(toml_file)
        return settings[value]


secrets = {
    "ssid": getenv("CIRCUITPY_WIFI_SSID"),
    "password": getenv("CIRCUITPY_WIFI_PASSWORD"),
    "broker": getenv("broker"),
    "user": getenv("ADAFRUIT_AIO_USERNAME"),
    "pass": getenv("ADAFRUIT_AIO_KEY"),
}
if secrets == {"ssid": None, "password": None}:
    try:
        # Fallback on secrets.py until depreciation is over and option is removed
        from secrets import secrets
    except ImportError:
        print("WiFi secrets are kept in settings.toml, please add them there!")
        raise

IMAGE_URL = "https://silversaucer.com/static/img/album-art/image_300.bmp"

# pyportal = PyPortal()
# pyportal.network.connect()

# Set up WIFI w/ConnectionManager

if is_microcontroller:
    import board
    import busio
    from digitalio import DigitalInOut
    from adafruit_esp32spi import adafruit_esp32spi

    esp32_cs = DigitalInOut(board.ESP_CS)
    esp32_ready = DigitalInOut(board.ESP_BUSY)
    esp32_reset = DigitalInOut(board.ESP_RESET)

    # Secondary (SCK1) SPI used to connect to WiFi board on Arduino Nano Connect RP2040
    if "SCK1" in dir(board):
        spi = busio.SPI(board.SCK1, board.MOSI1, board.MISO1)
    else:
        spi = busio.SPI(board.SCK, board.MOSI, board.MISO)
    esp = adafruit_esp32spi.ESP_SPIcontrol(spi, esp32_cs, esp32_ready, esp32_reset)
    print(esp.firmware_version)
else:
    esp = adafruit_connection_manager.CPythonNetwork()

pool = adafruit_connection_manager.get_radio_socketpool(esp)
ssl_context = adafruit_connection_manager.get_radio_ssl_context(esp)
requests = adafruit_requests.Session(pool, ssl_context)

# Set up logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

if is_microcontroller:
    print("Connecting to AP...")
    while not esp.is_connected:
        try:
            esp.connect_AP(secrets["ssid"], secrets["password"])
        except OSError as e:
            print("could not connect to AP, retrying: ", e)
            continue
    print("Connected to", str(esp.ssid, "utf-8"), "\tRSSI:", esp.rssi)
    print("sleeping for 10 seconds to make sure wifi is good...")
    time.sleep(10)
else:
    print("Using CPython sockets...")

# Get the Album title and artist name from JSON
data_source = "https://silversaucer.com/album/data"

try:
    resp = requests.get(data_source)
    data = resp.json()
    print(data)

except RuntimeError:
    resp = requests.get(data_source)
    data = resp.json()
    print(data)

image_location = data["image_url"]
artist = data["artist"]
album = data["album"]

album_info = artist + " - " + album

# album_info = "Garbage - Garbage"

# Load image on disk and display it on first boot
display = board.DISPLAY
display.rotation = 90

winamp = displayio.OnDiskBitmap(open("winamp256.bmp", "rb"))

# Create a TileGrid to hold the bitmap
tile_grid_1 = displayio.TileGrid(winamp, pixel_shader=winamp.pixel_shader, x=0)

# Create a Group to hold the TileGrid
group = displayio.Group()

# Add the TileGrid to the Group
group.append(tile_grid_1)

album_art = displayio.OnDiskBitmap("albumart.bmp")

tile_grid_2 = displayio.TileGrid(album_art, pixel_shader=album_art.pixel_shader, y=120)
group.append(tile_grid_2)

font = terminalio.FONT
color = 0x00E200

text_area = label.Label(font, text=album_info, color=color)

text_area.x = 130
text_area.y = 42
group.append(text_area)

# Add the Group to the Display
display.root_group = group

# ------------- MQTT Topic Setup ------------- #
mqtt_topic = "prcutler/feeds/albumart"
#mqtt_topic = "justmobilize/feeds/albumart"


### Code ###
# Define callback methods which are called when events occur
# pylint: disable=unused-argument, redefined-outer-name
def connected(client, userdata, flags, rc):
    # This function will be called when the client is connected
    # successfully to the broker.
    print("Subscribing to %s" % (mqtt_topic))
    client.subscribe(mqtt_topic)


def disconnected(client, userdata, rc):
    # This method is called when the client is disconnected
    print("Disconnected from MQTT Broker!")


def message(client, topic, message):
    """Method callled when a client's subscribed feed has a new
    value.
    :param str topic: The topic of the feed with a new value.
    :param str message: The new value
    """
    if message == "New album picked!":
        print("New message on topic {0}: {1}".format(topic, message))
        response = None

        url = "https://silversaucer.com/static/img/album-art/image_300.bmp"

        response = requests.get(url)
        if response.status_code == 200:
            print("Starting image download...")
            with open("albumart.bmp", "wb") as f:
                for chunk in response.iter_content(chunk_size=32):
                    f.write(chunk)
                print("Album art saved")
            response.close()

            resp = requests.get(data_source)
            data = resp.json()
            print(data)

            # There's a few different places we look for data in the photo of the day
            image_location = data["image_url"]
            artist = data["artist"]
            album = data["album"]

            album_str = artist + " - " + album

            album_info = album_str[:30]

            winamp = displayio.OnDiskBitmap(open("winamp256.bmp", "rb"))

            # Create a TileGrid to hold the bitmap
            tile_grid_1 = displayio.TileGrid(winamp, pixel_shader=winamp.pixel_shader)

            # Create a Group to hold the TileGrid
            group = displayio.Group()

            # Add the TileGrid to the Group
            group.append(tile_grid_1)

            album_art = displayio.OnDiskBitmap("albumart.bmp")

            tile_grid_2 = displayio.TileGrid(album_art, pixel_shader=album_art.pixel_shader, y=120)
            group.append(tile_grid_2)

            font = terminalio.FONT
            color = 0x00E200

            text_area = label.Label(font, text=album_info, color=color)

            text_area.x = 130
            text_area.y = 42
            group.append(text_area)

            # Add the Group to the Display
            display.root_group = group

            time.sleep(10)

        else:
            print("Bad get request")


# Initialize MQTT interface with the esp interface
# pylint: disable=protected-access
# MQTT.set_socket(socket, pyportal.network._wifi.esp)

# Set up a MiniMQTT Client
mqtt_client = MQTT.MQTT(
    broker=getenv("broker"),
    username=getenv("ADAFRUIT_AIO_USERNAME"),
    password=getenv("ADAFRUIT_AIO_KEY"),
    port=1883,
    socket_timeout=5,
    recv_timeout=10,
    socket_pool=pool,
    ssl_context=ssl_context,
    is_ssl=False,
)

# Setup the callback methods above
mqtt_client.on_connect = connected
mqtt_client.on_disconnect = disconnected
mqtt_client.on_message = message

# Connect the client to the MQTT broker.
mqtt_client.logger = logger

mqtt_client.connect()

while True:
    # Poll the message queue
    try:
        mqtt_client.loop(5)

    except RuntimeError or ConnectionError:
        time.sleep(10)
        mqtt_client.connect()
        mqtt_client.loop()

    # Send a new message
    time.sleep(5)

[justmobilize]

@prcutler what crashes are you getting?

[prcutler]

It was a generic MQTT error like yesterday, so it's probably memory problems. Of course it's running perfectly for the last 15 minutes while I wait for something to happen. :)

As soon as it happens, I'll paste it in. Thanks!

[dhalbert]

I have updated the library with a couple of pending PR's, and released as 7.8.0, if you want to re-test with that.

[justmobilize]

@dhalbert did you see the issue with getting a string error back?

[dhalbert]

Yes, but that's separate, is that right? I thought I would try to get some fixes in.

[justmobilize]

He did need to add that valve to an error check to get it to work. I have no idea where it's coming from

[dhalbert]

Aha, it is not a string, it is an exception in adafruit_espspi_socketpool.py named timeout:

class timeout(TimeoutError):  # pylint: disable=invalid-name
    """TimeoutError class. An instance of this error will be raised by recv_into() if
    the timeout has elapsed and we haven't received any data yet."""

It is from this deprecated CPython exception:
https://docs.python.org/3/library/socket.html#socket.timeout
I think this could be removed, but we'd need to check the examples and the Learn Guide repo for any use of this.

EDIT: more details:

In adafruit_minimqtt.py, there is this code:

        # CPython socket module contains a timeout attribute
        if hasattr(self._socket_pool, "timeout"):
            try:
                res = self._sock_exact_recv(1)
            except self._socket_pool.timeout:
                return None
        else:  # socketpool, esp32spi
            try:
                res = self._sock_exact_recv(1, timeout=timeout)
            except OSError as error:
                if error.errno in (errno.ETIMEDOUT, errno.EAGAIN):
                    # raised by a socket timeout if 0 bytes were present
                    return None
                raise MMQTTException from error

And in adafruit_espspi_socketpool.py, there is currently this code:

            if self._timeout > 0 and time.monotonic() - last_read_time > self._timeout:
                raise timeout("timed out")

In CPython, the exception class socket.timeout was made a subclass of OSError in CPython 3.3, and was made an alias of TimeoutError (which itself is a subclass of OSError) in CPython 3.10. It was not deprecated until 3.10. Since we would want to support CPython versions earlier than 3.10, we need to keep this historical baggage for now.

[justmobilize]

@dhalbert oh man....

Two options I see:

1. change it to raise timeout(errno.ETIMEDOUT) in adafruit_esp32spi_socketpool.py
2. in adafruit_minimqtt.py have the if error.errno in (errno.ETIMEDOUT, errno.EAGAIN): also check or isinstance(error, TimeoutError))

Thoughts?

I think this one will fix a bunch of esp32spi mqtt issues.

[dhalbert]

But is if hasattr(self._socket_pool, "timeout"): going to be True when using the ESP32SPI socketpool? Then it should be taking that branch already? This is despite the comment about CPython.

[justmobilize]

Oh, your right, the code comment # socketpool, esp32spi threw me. But he was totally getting that error.

@prcutler could you grab the method _wait_for_msg from minimqtt from your device when you have a chance?

[prcutler]

I replaced lines 978-982 in adafruit_minimqtt.py per Justin with:

            except OSError as error:
                print(f"OSError on _wait_for_msg: {error.errno}")
                if error.errno in (errno.ETIMEDOUT, errno.EAGAIN, "timed out"):
                    # raised by a socket timeout if 0 bytes were present
                    return None
                raise MMQTTException from error