Wiznet makers

Here we use two MicroPython MQTT protocol libraries from the pypi (Python Package Index) shared Python‑package platform: micropython‑umqtt.simple and micropython‑umqtt.robust. They provide MQTT‑client functionality for MicroPython, but differ in design goals and features.

micropython‑umqtt.simple A lightweight MQTT‑client library designed for MicroPython. Implements the original MQTT protocol and provides core MQTT‑client features for resource‑constrained devices.

• Offers basic MQTT‑client capabilities: connect to MQTT broker, publish messages and subscribe to topics
• Suitable for resource‑limited embedded devices; no extra complex features or redundant code
• Supports core functions of MQTT 3.1.1: connect, publish, subscribe and message reception

Project page: https://pypi.org/project/micropython‑umqtt.simple/#description。

micropython‑umqtt.robust An extended version based on micropython‑umqtt.simple. It improves client robustness under unstable‑network conditions and resolves issues in umqtt.simple such as network drop‑out and re‑connection handling.

• Enhanced support for unstable networks with improved error‑handling and reconnection mechanisms
• Detailed logging functions to assist debugging and MQTT‑client status monitoring
• Built‑in auto‑reconnect for network interruption or lost‑connection scenarios

Project page: https://pypi.org/project/micropython‑umqtt.robust/。

Source code for micropython‑umqtt.simple:

# Python env   : MicroPython v1.23.0 on Wiznet W5500
# -*- coding: utf-8 -*-        
# @Time    : 2024/8/9 11:01 AM   
# @Author  : Li Qingshui            
# @File    : umqttsimple.py       
# @Description : Implement a simple MQTT client class
# ======================================== Import related modules ========================================
# Import network‑connection module
import usocket as socket
# Import byte‑stream and binary‑data processing module
import ustruct as struct
# Import base‑convert module
from ubinascii import hexlify
# Import time‑related module
import time
# ======================================== Global variables ============================================
# ======================================== Function definitions ============================================
# ======================================== Custom classes ============================================
# Custom MQTT exception class for MQTT‑operation errors
class MQTTException(Exception):
    '''
    Custom MQTT exception class to handle errors during MQTT protocol operations.
    Inherits from Python built‑in Exception and raises dedicated errors for MQTT‑client operations.
    '''
    pass

# MQTT client class for basic MQTT‑protocol operations
class MQTTClient:
    """
    MQTTClient class implementing fundamental MQTT‑protocol operations: connect to broker, publish messages, subscribe to topics, process incoming messages.
    This class encapsulates MQTT‑client functions, enables communication with MQTT‑servers, and supports message publish‑subscribe and last‑will configuration.
    Supports SSL/TLS encrypted connections, configurable keep‑alive interval, username and password.
    Attributes:
        client_id (int): Client identifier.
        server (str): MQTT broker address.
        port (int): MQTT broker port, default 0 (auto‑select port).
        user (str): Username, default None.
        password (str): Password, default None.
        keepalive (int): Keep‑alive interval, default 0.
        ssl (bool): Enable SSL/TLS encrypted connection, default False.
        ssl_params (dict): SSL/TLS parameters, default empty dict.
        sock (socket): Socket for client‑broker communication.
        pid (int): Packet identifier for MQTT packets.
        cb (callable): Message callback function for received payloads.
        lw_topic (str): Last‑will topic, default None.
        lw_msg (str): Last‑will message content, default None.
        lw_qos (int): Last‑will QoS level, default 0.
        lw_retain (bool): Last‑will retain flag, default False.
    Methods:
        __init__(self, client_id, server, port=0, user=None, password=None, keepalive=0, ssl=False, ssl_params={}):
            Initialize MQTTClient instance.
        _send_str(self, s):
            Send string data to broker.
        _recv_len(self):
            Receive variable‑length integer for parsing MQTT packet length field.
        set_callback(self, f):
            Assign message callback function.
        set_last_will(self, topic, msg, retain=False, qos=0):
            Configure last‑will message.
        connect(self, clean_session=True):
            Connect to MQTT broker.
        disconnect(self):
            Disconnect from MQTT broker.
        ping(self):
            Send PINGREQ keep‑alive request.
        publish(self, topic, msg, retain=False, qos=0):
            Publish message to target topic.
        subscribe(self, topic, qos=0):
            Subscribe to specified topic.
        wait_msg(self):
            Block and process a single incoming MQTT message.
        check_msg(self):
            Check for pending incoming messages from broker.
    """
    def __init__(self, client_id: int, server: str, port: int = 0, user: str = None, password: str = None,
                 keepalive: int = 0, ssl: bool = False, ssl_params: dict = {}) -> None:
        '''
        Initialize MQTT‑client instance
        Args:
            client_id (int): Client identifier
            server (str): Broker address
            port (int): Broker port, default 0 for auto‑selection
            user (str): Username, default None
            password (str): Password, default None
            keepalive (int): Keep‑alive interval, default 0
            ssl (bool): Enable SSL, default False
            ssl_params (dict): SSL parameters, default empty dict
        Returns:
            None
        '''
        # Auto‑select port when port equals zero
        if port == 0:
            port = 8883 if ssl else 1883
        self.client_id = client_id
        self.sock = None
        self.server = server
        self.port = port
        self.ssl = ssl
        self.ssl_params = ssl_params
        self.pid = 0
        self.cb = None
        self.user = user
        self.pswd = password
        self.keepalive = keepalive
        self.lw_topic = None
        self.lw_msg = None
        self.lw_qos = 0
        self.lw_retain = False

    # Send string payload
    def _send_str(self, s: str) -> None:
        '''
        Send string‑type data
        Args:
            s (str): String to transmit
        Returns:
            None
        '''
        self.sock.write(struct.pack("!H", len(s)))
        self.sock.write(s)

    # Receive variable‑length integer for MQTT‑packet length parsing
    def _recv_len(self) -> int:
        '''
        Read variable‑length integer used for MQTT packet length field
        Returns:
            int: Parsed length value
        '''
        n = 0
        sh = 0
        while True:
            b = self.sock.read(1)[0]
            n |= (b & 0x7F) << sh
            if not b & 0x80:
                return n
            sh += 7

    # Assign message callback function
    def set_callback(self, f: callable) -> None:
        '''
        Set callback function for incoming messages
        Args:
            f (callable): Callback function reference
        Returns:
            None
        '''
        self.cb = f

    # Configure last‑will message
    def set_last_will(self, topic: str, msg: str, retain: bool = False, qos: int = 0) -> None:
        '''
        Define MQTT last‑will message
        Args:
            topic (str): Will‑message topic
            msg (str): Will‑message payload
            retain (bool): Retain flag, default False
            qos (int): QoS level, default 0
        Returns:
            None
        Raises:
            AssertionError: Invalid QoS value or empty topic
        '''
        assert 0 <= qos <= 2
        assert topic
        self.lw_topic = topic
        self.lw_msg = msg
        self.lw_qos = qos
        self.lw_retain = retain

    # Establish connection to MQTT broker
    def connect(self, clean_session: bool = True) -> int:
        '''
        Connect to MQTT broker
        Args:
            clean_session (bool): Clear previous session state, default True
        Returns:
            int: Connection‑result flag
        Raises:
            MQTTException: Connection rejected by broker
        '''
        self.sock = socket.socket()
        addr = socket.getaddrinfo(self.server, self.port)[0][-1]
        self.sock.connect(addr)
        if self.ssl:
            import ussl
            self.sock = ussl.wrap_socket(self.sock, **self.ssl_params)
        premsg = bytearray(b"\x10\0\0\0\0\0")
        msg = bytearray(b"\x04MQTT\x04\x02\0\0")
        sz = 10 + 2 + len(self.client_id)
        msg[6] = clean_session << 1
        if self.user is not None:
            sz += 2 + len(self.user) + 2 + len(self.pswd)
            msg[6] |= 0xC0
        if self.keepalive:
            assert self.keepalive < 65536
            msg[7] |= self.keepalive >> 8
            msg[8] |= self.keepalive & 0x00FF
        if self.lw_topic:
            sz += 2 + len(self.lw_topic) + 2 + len(self.lw_msg)
            msg[6] |= 0x4 | (self.lw_qos & 0x1) << 3 | (self.lw_qos & 0x2) << 3
            msg[6] |= self.lw_retain << 5
        i = 1
        while sz > 0x7F:
            premsg[i] = (sz & 0x7F) | 0x80
            sz >>= 7
            i += 1
        premsg[i] = sz
        self.sock.write(premsg, i + 2)
        self.sock.write(msg)
        self._send_str(self.client_id)
        if self.lw_topic:
            self._send_str(self.lw_topic)
            self._send_str(self.lw_msg)
        if self.user is not None:
            self._send_str(self.user)
            self._send_str(self.pswd)
        resp = self.sock.read(4)
        assert resp[0] == 0x20 and resp[1] == 0x02
        if resp[3] != 0:
            raise MQTTException(resp[3])
        return resp[2] & 1

    # Disconnect from MQTT broker
    def disconnect(self) -> None:
        '''
        Terminate MQTT connection
        Returns:
            None
        '''
        self.sock.write(b"\xe0\0")
        self.sock.close()

    # Transmit keep‑alive PINGREQ packet
    def ping(self) -> None:
        '''
        Send PINGREQ keep‑alive request
        Returns:
            None
        '''
        self.sock.write(b"\xc0\0")

    # Publish message to specified topic
    def publish(self, topic: str, msg: str, retain: bool = False, qos: int = 0) -> None:
        '''
        Publish MQTT message
        Args:
            topic (str): Target topic name
            msg (str): Message payload
            retain (bool): Retain flag, default False
            qos (int): Quality‑of‑service level (0,1,2), default 0
        Returns:
            None
        Raises:
            AssertionError: Exceed maximum MQTT‑packet size
        '''
        pkt = bytearray(b"\x30\0\0\0")
        pkt[0] |= qos << 1 | retain
        sz = 2 + len(topic) + len(msg)
        if qos > 0:
            sz += 2
        assert sz < 2097152
        i = 1
        while sz > 0x7F:
            pkt[i] = (sz & 0x7F) | 0x80
            sz >>= 7
            i += 1
        pkt[i] = sz
        self.sock.write(pkt, i + 1)
        self._send_str(topic)
        if qos > 0:
            self.pid += 1
            pid = self.pid
            struct.pack_into("!H", pkt, 0, pid)
            self.sock.write(pkt, 2)
        self.sock.write(msg)
        if qos == 1:
            while 1:
                op = self.wait_msg()
                if op == 0x40:
                    sz = self.sock.read(1)
                    assert sz == b"\x02"
                    rcv_pid = self.sock.read(2)
                    rcv_pid = rcv_pid[0] << 8 | rcv_pid[1]
                    if pid == rcv_pid:
                        return
        elif qos == 2:
            assert 0

    # Subscribe to target topic
    def subscribe(self, topic: str, qos: int = 0) -> None:
        '''
        Subscribe to MQTT topic
        Args:
            topic (str): Topic to subscribe
            qos (int): Requested QoS level, default 0
        Returns:
            None
        Raises:
            AssertionError: Callback function not assigned
            MQTTException: Broker returns subscribe‑failure code
        '''
        assert self.cb is not None, "Subscribe callback is not set"
        pkt = bytearray(b"\x82\0\0\0")
        self.pid += 1
        struct.pack_into("!BH", pkt, 1, 2 + 2 + len(topic) + 1, self.pid)
        self.sock.write(pkt)
        self._send_str(topic)
        self.sock.write(qos.to_bytes(1, "little"))
        while 1:
            op = self.wait_msg()
            if op == 0x90:
                resp = self.sock.read(4)
                assert resp[1] == pkt[2] and resp[2] == pkt[3]
                if resp[3] == 0x80:
                    raise MQTTException(resp[3])
                return

    # Block and process single incoming MQTT message
    # Received subscribed messages are passed to previously‑assigned callback function
    # Internal MQTT control‑packets are processed inside this method
    def wait_msg(self) -> int:
        '''
        Block for one incoming MQTT message and process it
        Returns:
            int: Operation code; return None when no message arrives
        Raises:
            OSError: Empty read result
            AssertionError: Invalid PINGRESP packet
        '''
        res = self.sock.read(1)
        if res is None:
            return None
        if res == b"":
            raise OSError(-1)
        if res == b"\xd0":
            sz = self.sock.read(1)[0]
            assert sz == 0
            return None
        op = res[0]
        if op & 0xF0 != 0x30:
            return op
        sz = self._recv_len()
        topic_len = self.sock.read(2)
        topic_len = (topic_len[0] << 8) | topic_len[1]
        topic = self.sock.read(topic_len)
        sz -= topic_len + 2
        if op & 6:
            pid = self.sock.read(2)
            pid = pid[0] << 8 | pid[1]
            sz -= 2
        msg = self.sock.read(sz)
        self.cb(topic, msg)
        if op & 6 == 2:
            pkt = bytearray(b"\x40\x02\0\0")
            struct.pack_into("!H", pkt, 2, pid)
            self.sock.write(pkt)
        elif op & 6 == 4:
            assert 0
        return op

    # Check for pending messages from broker
    def check_msg(self) -> int:
        '''
        Check for queued incoming MQTT messages
        Returns:
            int: Operation code; return None when no message arrives
        '''
        return self.wait_msg()
# ======================================== Initialization ==========================================
# ======================================== Main program ===========================================

The micropython‑umqtt.simple file defines two classes:

MQTTException class Custom MQTT exception class for handling errors during MQTT operations.

MQTTClient class Implements basic MQTT‑client operations including connect, disconnect, publish and subscribe.

The MQTTClient class inside micropython‑umqtt.simple supplies these main methods for MQTT data transmission:

The umqtt.robust file also provides an MQTTClient class. It inherits all features from umqttsimple.MQTTClient, adds a debug parameter and a log method to toggle debug print output for troubleshooting. It leverages Python exception handling and infinite loops to override parent‑class methods: reconnect, publish, wait_msg, check_msg. These modifications resolve dead‑lock or infinite‑recursion risks under weak‑network / disconnection conditions and guarantee normal data transmission.

Below is slightly‑modified umqtt.robust source code for cleaner implementation and better Python object‑oriented style:

# Python env   : MicroPython v1.23.0 on Wiznet W5500
# -*- coding: utf-8 -*-        
# @Time    : 2024/8/9 11:38 PM   
# @Author  : Li Qingshui            
# @File    : umqttrobust.py       
# @Description : Solve dead‑lock and infinite‑recursion issues of umqttsimple under poor‑network or disconnection
# ======================================== Import related modules ========================================
# Import time‑related module
import time
# Import umqttsimple module
import umqttsimple
# ======================================== Global variables ============================================
# ======================================== Function definitions ============================================
# ======================================== Custom classes ============================================
# Inherit from umqttsimple.MQTTClient to fix dead‑lock / infinite‑recursion under unstable network
class MQTTClient(umqttsimple.MQTTClient):
    """
    MQTTClient class inherited from umqttsimple.MQTTClient for enhanced MQTT‑client capability.
    Extensions added on base class:
    - Debug mode for detailed error‑log output.
    - Auto‑reconnect to MQTT broker under unstable‑network / disconnection scenarios.
    - Limited‑attempt message checking and retry mechanism to avoid infinite blocking.
    Attributes:
        client_id (int): Client identifier.
        server (str): MQTT broker address.
        port (int): MQTT broker port, default 0 (auto‑select port).
        user (str): Username, default None.
        password (str): Password, default None.
        keepalive (int): Keep‑alive interval, default 0.
        ssl (bool): Enable SSL/TLS encrypted connection, default False.
        debug (bool): Toggle debug log output, default False.
        ssl_params (dict): SSL/TLS parameters, default empty dict.
        sock (socket): Socket for client‑broker communication.
        pid (int): Packet identifier for MQTT packets.
        cb (callable): Message callback function for received payloads.
        lw_topic (str): Last‑will topic, default None.
        lw_msg (str): Last‑will message content, default None.
        lw_qos (int): Last‑will QoS level, default 0.
        lw_retain (bool): Last‑will retain flag, default False.
    Methods:
        __init__(self, client_id, server, port=0, user=None, password=None, keepalive=0, ssl=False, debug=False, ssl_params={}):
            Initialize MQTTClient instance.
        log(self, in_reconnect, e):
            Print error messages for debugging.
        reconnect(self):
            Re‑establish MQTT‑broker connection under unstable‑network conditions.
        publish(self, topic, msg, retain=False, qos=0):
            Publish message with auto‑reconnect support.
        wait_msg(self):
            Wait for incoming message with auto‑reconnect support.
        check_msg(self, attempts=2):
            Check pending messages with finite retry attempts.
    """
    def __init__(self, client_id: int, server: str, port: int = 0, user: str = None, password: str = None, keepalive: int = 0, ssl: bool = False, debug: bool = False, ssl_params: dict = {}) -> None:
        '''
        Initialize MQTT‑client instance
        Args:
            client_id (int): Client identifier
            server (str): Broker address
            port (int): Broker port, default 0 for auto‑selection
            user (str): Username, default None
            password (str): Password, default None
            keepalive (int): Keep‑alive interval, default 0
            ssl (bool): Enable SSL, default False
            debug (bool): Toggle debug output, default False
            ssl_params (dict): SSL parameters, default empty dict
        Returns:
            None
        '''
        super().__init__(client_id, server, port, user, password, keepalive, ssl, ssl_params, )
        self.debug = debug

    def log(self, in_reconnect: bool, e: Exception) -> None:
        '''
        Output debug error information
        Args:
            in_reconnect (bool): Whether error occurs during reconnection flow
            e (Exception): Exception instance
        Returns:
            None
        '''
        if self.debug:
            if in_reconnect:
                print("mqtt reconnect: %r" % e)
            else:
                print("mqtt: %r" % e)

    def reconnect(self) -> None:
        '''
        Reconnect to MQTT broker for unstable‑network scenarios
        Returns:
            None
        '''
        i = 0
        while True:
            try:
                return super().connect(False)
            except OSError as e:
                self.log(True, e)
                i += 1
                time.sleep(i)

    def publish(self, topic: str, msg: str, retain: bool = False, qos: int = 0) -> None:
        '''
        Publish message and handle publish‑failure under unstable‑network conditions
        Args:
            topic (str): Target topic name
            msg (str): Message payload
            retain (bool): Retain flag, default False
            qos (int): Quality‑of‑service level, default 0
        Returns:
            None
        '''
        while True:
            try:
                return super().publish(topic, msg, retain, qos)
            except OSError as e:
                self.log(False, e)
            self.reconnect()

    def wait_msg(self) -> int:
        '''
        Blocking wait‑for‑message function for unstable‑network error handling
        Returns:
            int: Operation code
        '''
        while True:
            try:
                return super().wait_msg()
            except OSError as e:
                self.log(False, e)
            self.reconnect()

    def check_msg(self, attempts: int = 2) -> int:
        '''
        Check incoming messages with finite retry attempts for unstable‑network scenarios
        Args:
            attempts (int): Maximum retry times, default 2
        Returns:
            int or None: Operation code; None if no message received
        '''
        while attempts:
            self.sock.setblocking(False)
            try:
                return super().wait_msg()
            except OSError as e:
                self.log(False, e)
            self.reconnect()
            attempts -= 1
# ======================================== Initialization ==========================================
# ======================================== Main program ===========================================

Modifications compared with original umqtt.robust:

Save micropython‑umqtt.simple and micropython‑umqtt.robust as umqttsimple.py and umqttrobust.py inside the same folder. Treat this folder as project root directory for multi‑file firmware flashing.

Relationship between the two source files and their MQTTClient classes: