141 lines
4.1 KiB
Python
141 lines
4.1 KiB
Python
import machine
|
|
import neopixel
|
|
import config
|
|
import ntptime
|
|
import time
|
|
from umqtt.simple import MQTTClient
|
|
from machine import WDT
|
|
|
|
hour_color = config.HOUR_COLOR
|
|
minute_color = config.MINUTE_COLOR
|
|
brightness = 1.0
|
|
switch_state = True
|
|
|
|
def set_time(hour, minute):
|
|
np_hour = neopixel.NeoPixel(machine.Pin(23), 24, bpp=4)
|
|
np_hour[(hour % 12) * 2 + int(minute / 30)] = hour_color if switch_state == True else (0,0,0,0)
|
|
np_hour.write()
|
|
|
|
np_minute = neopixel.NeoPixel(machine.Pin(32), 30, bpp=4)
|
|
np_minute[int(minute / 2)] = minute_color if switch_state == True else (0,0,0,0)
|
|
np_minute.write()
|
|
|
|
def sync_ntptime():
|
|
try:
|
|
print("Syncing time...")
|
|
ntptime.settime()
|
|
except:
|
|
print("Could not sync time")
|
|
|
|
def check_daylight_saving(current_time):
|
|
for day in range(25, 32):
|
|
t = time.mktime((current_time[0], 3, day, 2, 0, 0, 0, 0))
|
|
if time.localtime(t)[6] == 6:
|
|
summer_time = t
|
|
break
|
|
|
|
for day in range(25, 32):
|
|
t = time.mktime((current_time[0], 10, day, 2, 0, 0, 0, 0))
|
|
if time.localtime(t)[6] == 6:
|
|
winter_time = t
|
|
break
|
|
|
|
comp_time = time.mktime(current_time)
|
|
if summer_time <= comp_time <= winter_time:
|
|
return True
|
|
return False
|
|
|
|
def check_wifi_connection():
|
|
import network
|
|
sta_if = network.WLAN(network.STA_IF)
|
|
return sta_if.isconnected
|
|
|
|
def mqtt_send_availability(mqtt_client):
|
|
mqtt_client.publish(config.MQTT_TOPIC + "/available", "online")
|
|
|
|
def mqtt_send_status(mqtt_client):
|
|
global brightness
|
|
global switch_state
|
|
mqtt_client.publish(config.MQTT_TOPIC + "/brightness", str(int(brightness * 100)))
|
|
|
|
if switch_state == True:
|
|
mqtt_client.publish(config.MQTT_TOPIC + "/switch", 'ON')
|
|
else:
|
|
mqtt_client.publish(config.MQTT_TOPIC + "/switch", 'OFF')
|
|
|
|
def mqtt_receive_message(topic, msg):
|
|
global hour_color
|
|
global minute_color
|
|
global brightness
|
|
global switch_state
|
|
if topic.decode('utf-8') == config.MQTT_TOPIC + "/brightness/set":
|
|
brightness = float(msg) / 100.0
|
|
hour_color = (
|
|
int(float(config.HOUR_COLOR[0]) * brightness),
|
|
int(float(config.HOUR_COLOR[1]) * brightness),
|
|
int(float(config.HOUR_COLOR[2]) * brightness),
|
|
int(float(config.HOUR_COLOR[3]) * brightness),
|
|
)
|
|
minute_color = (
|
|
int(float(config.MINUTE_COLOR[0]) * brightness),
|
|
int(float(config.MINUTE_COLOR[1]) * brightness),
|
|
int(float(config.MINUTE_COLOR[2]) * brightness),
|
|
int(float(config.MINUTE_COLOR[3]) * brightness),
|
|
)
|
|
elif topic.decode('utf-8') == config.MQTT_TOPIC + "/switch/set":
|
|
if msg.decode('utf-8').strip() == "ON":
|
|
switch_state = True
|
|
else:
|
|
switch_state = False
|
|
|
|
# Test LEDs before startup
|
|
for hour in range(0, 11):
|
|
set_time(hour, 0)
|
|
time.sleep(0.1)
|
|
for minute in range(0, 59):
|
|
set_time(0, minute)
|
|
time.sleep(0.05)
|
|
set_time(0, 0)
|
|
|
|
# Configure Watchdog
|
|
wdt = WDT(timeout=5000)
|
|
|
|
if not check_wifi_connection():
|
|
machine.reset()
|
|
|
|
# MQTT Setup
|
|
mqtt_client = MQTTClient(config.MQTT_ID, config.MQTT_SERVER, config.MQTT_PORT, config.MQTT_USER, config.MQTT_PASSWORD, 30)
|
|
mqtt_client.set_callback(mqtt_receive_message)
|
|
mqtt_client.connect()
|
|
mqtt_client.subscribe(config.MQTT_TOPIC + "/brightness/set")
|
|
mqtt_client.subscribe(config.MQTT_TOPIC + "/switch/set")
|
|
|
|
# Main Loop
|
|
while True:
|
|
|
|
wdt.feed()
|
|
|
|
if not check_wifi_connection():
|
|
machine.reset()
|
|
|
|
current_time = time.localtime()
|
|
|
|
# Sync time if we are on full hour
|
|
if current_time[4] == 0 and current_time[5] == 0 or current_time[0] == 2000:
|
|
sync_ntptime()
|
|
current_time = time.localtime()
|
|
|
|
if check_daylight_saving(current_time):
|
|
current_time = time.localtime(time.mktime(current_time) + config.TIMEZONE_OFFSET_DAYLIGHT)
|
|
else:
|
|
current_time = time.localtime(time.mktime(current_time) + config.TIMEZONE_OFFSET)
|
|
|
|
# Call MQTT
|
|
mqtt_client.check_msg()
|
|
mqtt_send_availability(mqtt_client)
|
|
mqtt_send_status(mqtt_client)
|
|
|
|
print("Set Time:", current_time)
|
|
set_time(current_time[3], current_time[4])
|
|
time.sleep(1)
|