Atomic Socket

案例程序

from m5stack import *
from m5ui import *
from uiflow import *
from base.Socket_Kit import Socket
import time

button_next = None

sock = Socket()

from numbers import Number

def buttonA_wasPressed():
  global button_next
  button_next = (button_next if isinstance(button_next, Number) else 0) + 1
  if button_next == 1:
    sock.set_relay_state(1)
  elif button_next == 2:
    sock.set_relay_state(0)
    button_next = 0
  pass
btnA.wasPressed(buttonA_wasPressed)

button_next = 0
sock.remote_control(1)
print(sock.pub_topic)
print((str('This is Subscribe Topic(Relay State), Use Remote User: ') + str((sock.sub_topic_relay))))
print((str('This is Subscribe Topic(Parameter Status), Use Remote User: ') + str((sock.sub_topic_data))))
while True:
  if sock.wait_update_data():
    print((str('Voltage(V): ') + str((sock.get_voltage()))))
    print((str('Current(mA): ') + str((sock.get_current()))))
    print((str('Power(W): ') + str((sock.get_active_power()))))
    sock.server_loop()
    wait(1)
  button_next = int((sock.get_relay_status()))
  if button_next:
    rgb.setColorAll(0x009900)
  else:
    rgb.setColorAll(0xff0000)
  wait_ms(2)

功能说明

sock.pub_topic

• 获取设备发送数据的主题（Topic）。

sock.sub_topic_relay

• 获取设备的订阅主题，设备会接收该主题下发布的消息。

sock.get_active_power()

• 获取设备当前的负载有功功率。

sock.get_current()

• 获取设备当前负载的电流。

sock.get_voltage()

• 获取设备当前负载的电压。

sock.get_relay_status()

• 获取设备继电器的当前状态（开/关）。

sock.remote_control(1)

• 设置设备的远程控制模式，这里设定为MQTT。

sock.server_loop()

• 在设备工作在AP模式下时，启用一个Web服务器轮询循环。

sock.set_relay_state(1)

• 设置继电器的状态为"开"（ON）。

sock.wait_update_data()

• 等待设备接收并处理更新的数据。