-
Notifications
You must be signed in to change notification settings - Fork 16
/
mqtt-disp7.py
76 lines (58 loc) · 1.59 KB
/
mqtt-disp7.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
# basic example: octopusLAB - ESP32 - WiFi - MQTT
print("mqtt-disp7.py > mqtt example")
from time import sleep
from utils.wifi_connect import read_wifi_config, WiFiConnect
from utils.mqtt import MQTT
from utils.pinout import set_pinout
from components.led import Led
from utils.octopus import disp7_init
from gc import mem_free
bd = bytes.decode
print("--- RAM free ---> " + str(mem_free()))
pinout = set_pinout()
led = Led(pinout.BUILT_IN_LED)
d7 = disp7_init() # 8 x 7segment display init
sleep(2)
for i in range(9):
d7.show(10-i)
sleep(0.2)
d7.show("")
def simple_blink():
led.value(1)
sleep(0.5)
led.value(0)
sleep(0.5)
def mqtt_handler(topic, msg):
print("MQTT handler {0}: {1}".format(topic, msg))
if "led" in topic:
print("led:", end='')
data = bytes.decode(msg)
if data[0] == '1': # on
print("-> on")
led.value(1)
elif data[0] == '0': # off
print("-> off")
led.value(0)
if "disp7" in topic:
print("disp7:")
data = bd(msg)
d7.show(data)
print("--- wifi_connect >")
net = WiFiConnect()
net.connect()
print("--- mqtt_connnect >")
# c = mqtt_connect_from_config(esp_id)
m = MQTT.from_config()
c = m.client
c.set_callback(mqtt_handler)
c.connect()
c.subscribe("octopus/device/{0}/#".format(m.client_id))
print("testing blink")
simple_blink()
print("send alive message")
c.publish("octopus/device", m.client_id) # topic, message (value) to publish
print("--- RAM free ---> " + str(mem_free()))
print("--- main loop >")
while True:
c.check_msg()
sleep(0.1)