MicroPython Forum (Archive)

Code: Select all

import socket
import time
try:
    import binascii
except ImportError:
    import ubinascii as binascii

import config


class MQTT(object):
    s = None
    addr = None
    prefix = ""

    def __init__(self, host, port=1883, prefix=config.MQTT_PREF):
        self.addr = socket.getaddrinfo(host, port)[0][4]
        self.prefix = prefix
        self.reset_socket()

    def reset_socket(self):
        self.s = socket.socket()

    def connect(self):
        address = self.addr
        self.s.connect(address)
        self.s.send(self.mtpConnect(config.DEVICE_NAME))

    def disconnect(self):
        self.s.send(self.mtpDisconnect())
        self.s.close()

    def publish(self, topic, data, sleep=1):
        self.s.send(self.mtpPub(self.prefix+topic, bytes(str(data), 'ascii')))
        if sleep:
            time.sleep(sleep)

    def recv(self, length=4096):
        return binascii.hexlify(self.s.recv(length))

    @staticmethod
    def mtStr(s):
        return bytes([len(s) >> 8, len(s) & 255]) + s.encode('utf-8')

    @staticmethod
    def mtPacket(cmd, variable, payload):
        return bytes([cmd, len(variable) + len(payload)]) + variable + payload

    def mtpConnect(self, name):
        return self.mtPacket(
            0b00010000,
            self.mtStr("MQTT") +  # protocol name
            b'\x04' +  # protocol level
            b'\x00' +  # connect flag
            b'\xFF\xFF',  # keepalive
            self.mtStr(name)
        )

    @staticmethod
    def mtpDisconnect():
        return bytes([0b11100000, 0b00000000])

    def mtpPub(self, topic, data):
        return self.mtPacket(0b00110001, self.mtStr(topic), data)