MQTT ESP32測試

ESP32在Docker MQTT使用上沒有成功，但將MQTT直接裝在PI上就可以使用了
如果在ESP32 MQTT 成功會出現以下的訊息

Connection successful

IP: 192.168.0.198
Socket connected to 192.168.0.172 on port 11883

Successfully connected to MQTT server at 192.168.0.172 on port 11883

Disconnected from MQTT server

import network

from umqtt.simple import MQTTClient

import usocket

# Wi-Fi 設定

SSID = "Bili-Net"

PASSWORD = "0932388283"

HOSTNAME = "ESP32_Test"

# MQTT 設定

mqtt_server = '192.168.0.172'

mqtt_port = 11883  # 測試端口

client_id = 'ESP32_test_client'

mqtt_topic = 'myTopic'

def connect_to_wifi():

    station = network.WLAN(network.STA_IF)

    station.active(True)

    station.config(dhcp_hostname=HOSTNAME)

    if not station.isconnected():

        print('Connecting to WiFi...')

        station.connect(SSID, PASSWORD)

        while not station.isconnected():

            pass

    print('Connection successful')

    print('IP:', station.ifconfig()[0])

def test_mqtt_connection():

    try:

        mqtt_client = MQTTClient(client_id, mqtt_server, port=mqtt_port)

#        mqtt_client = MQTTClient('', mqtt_server, port=mqtt_port)

        mqtt_client.connect()

        print('Successfully connected to MQTT server at {} on port {}'.format(mqtt_server, mqtt_port))

        mqtt_client.disconnect()

        print('Disconnected from MQTT server')

    except Exception as e:

        print('Could not connect to MQTT server {}:{}. Error: {}'.format(mqtt_server, mqtt_port, str(e)))

def test_socket_connection():

    addr = usocket.getaddrinfo(mqtt_server, mqtt_port)[0][-1]

    s = usocket.socket()

    try:

        s.connect(addr)

        print('Socket connected to {} on port {}'.format(mqtt_server, mqtt_port))

        s.close()

    except Exception as e:

        print('Socket connection failed:', e)

# 啟動測試程式

connect_to_wifi()

test_socket_connection()

test_mqtt_connection()