pdf-icon

UIFlow Guide

UiFlow1 Blockly

Event

Unit

UiFlow1 Project

Unit MQTT

Example

from m5stack import *
from m5stack_ui import *
from uiflow import *
import unit

screen = M5Screen()
screen.clean_screen()
screen.set_screen_bg_color(0xFFFFFF)
mqtt_0 = unit.get(unit.MQTT_ETH, 1, unit.PORTC)

mqtt_0.MqttStop()
while not (mqtt_0.isConnectLAN()):
  print('waiting LAN connection')
mqtt_0.configMQTT('mqtt.m5stack.com', 1883, 'unit_mqtt_id', '', '', 30,)
while not (mqtt_0.subscribe(1, 'mqtt_unit_down', unit_mqtt_cb, 0)):
  print('waiting subcribe topic')
while not (mqtt_0.saveParam()):
  print('waiting save configure')
mqtt_0.MqttStart()
while not (mqtt_0.isConnectMQTT()):
  print('waiting MQTT connection')
print('MQTT connected!')
while True:
  mqtt_0.publish('mqtt_unit_up', 'hello', 0)
  print(mqtt_0.receive_mqtt_message(5))
  wait_ms(2)

API

mqtt_0.uart_port_id(1)
  • Set Host Sensor ID
mqtt_0.Mqtt_cb_Loop()
  • Callback Polling
mqtt_0.configMQTT('', 0, '', '', '', 0,)
  • Create MQTTUnit Object
  • client_id: The unique client ID string broker used when connecting
  • Server:The hostname or IP address of the remote proxy
  • port: The network port of the server host to be connected to
  • username: The username used for proxy authentication
  • password: The password used for proxy authentication
  • keepalive: The maximum allowed time period
print((str('LAN:') + str((mqtt_0.isConnectLAN()))))
  • Check LAN Connection Status
print((str('MQTT:') + str((mqtt_0.isConnectMQTT()))))
  • Check MQTT Connection Status
mqtt_0.publish('', '', 0)
  • Publish Message
print((str('parameter:') + str((mqtt_0.saveParam()))))
  • Save Configuration Parameters
mqtt_0.MqttStart()
  • Start MQTT Client
mqtt_0.MqttStop()
  • Stop MQTT Client
print((str('msg:') + str((mqtt_0.subscribe(1, '', unit_mqtt_cb, 0)))))
  • Retrieve Subscribed Messages
def unit_mqtt_cb(mq_topic, mq_payload):
  global mqtt_topic, mqtt_msg
  mqtt_topic = mq_topic
  mqtt_msg = mq_payload
  pass
  • Publish/Subscribe Message
On This Page