Atomic Socket

Example

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)

API

sock.pub_topic

• Retrieves the device’s data publishing topic.

sock.sub_topic_relay

• Retrieves the subscription topic for the device, which receives messages published to this topic.

sock.get_active_power()

• Gets the current active power of the device’s load.

sock.get_current()

• Retrieves the current load current of the device.

sock.get_voltage()

• Retrieves the current load voltage of the device.

sock.get_relay_status()

• Gets the current state (on/off) of the relay.

sock.remote_control(1)

• Sets the device to remote control mode, here configured for MQTT.

sock.server_loop()

• Enables a web server polling loop when the device is in AP mode.

sock.set_relay_state(1)

• Sets the relay state to "on".

sock.wait_update_data()

• Waits for the device to receive and process updated data.