micropy-light/code/main.py

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.

"""
Eclairage à LED Neopixel
avec contrôle (encodeur rotatif)
"""
__author__ = "arofarn"
__version__ = 0.3

# Imports

import time
import machine
import esp
import neopixel
from encoder import Encoder
import ntptime
from umqtt.robust import MQTTClient
import light_modes
import config


# Déclaration des objets et initialisation des variables

NEOPIX_PIN = 2
NEOPIX2_PIN = 16
# Les deux broches suivantes doivent être capable d'interruption !!!
# sur ESP8266 => : 4, 5, 12, 13 et 14
ENC_PIN_A = 14      # N° de la 1ere broche de l'encodeur
ENC_PIN_B = 12      # N° de la 2e broche de l'encodeur
ENC_PIN_C = 13      # broche du clic central de l'encodeur

WIFI_LED_PIN = 0

# With ESP8266 (no timing parameter)
NPXL_STRIP = neopixel.NeoPixel(machine.Pin(NEOPIX_PIN), config.NB_PIX)
# NPXL_STRIP2 = neopixel.NeoPixel(machine.Pin(NEOPIX2_PIN), config.NB_PIX2)

# Eteint tout à l'initialisation
NPXL_STRIP.fill([0, 0, 0])
NPXL_STRIP.write()

# Encodeur rotatif
ENCODER = Encoder(ENC_PIN_A, ENC_PIN_B,
                  min_val=0, max_val=config.MAX_BRIGHT,
                  clicks=1)

# Bouton
ENC_BUT = machine.Pin(ENC_PIN_C, machine.Pin.IN)

# LED status WIFI_LED_PIN
WIFI_LED = machine.Pin(WIFI_LED_PIN, machine.Pin.OUT)

# RTC setup
rtc = machine.RTC()

# initialisation des variables d'état
BRIGHTN = 50            # Luminosité (0 - 100)
PWR = True              # Est-ce que l'éclairage est allumé ?
BUTTN_STATE = 1
CURRENT_MODE = 0
PWROFF_DELAY = 0
MQTT_DELAY = 2          # Délais en sec. avant d'envoyer les données au MQTT

# Connection au broker MQTT
CLIENT = MQTTClient(client_id=config.CLIENT_ID,
                    server=config.MQTT_HOST,
                    user=config.MQTT_USERNAME,
                    password=config.MQTT_PASSWD,
                    port=config.MQTT_PORT)
CLIENT.DEBUG = True
CLIENT.connect()

#############
# FUNCTIONS #
#############


def set_rtc():
    while True:
        try:
            return ntptime.settime()
        except OSError as err:
            print("OS error: {0}".format(err))
            if err.args[0] == 110:
                time.sleep(2)
        except OverflowError as err:
            print("OverflowError: {0}".format(err))
            time.sleep(2)
        except Exception as err:
            return print(err)


def now(clock):
    """Return a string representing date/time now."""
    return "{0:04d}/{1:02d}/{2:02d}_{4:02d}:{5:02d}:{6:02d}".format(*clock.datetime())


def mqtt_publish(topic, message, retain=False, qos=0, sleep=0):
    """MQTT publish helper."""
    global CLIENT

    CLIENT.publish("{location}/{device}/{topic}".format(location=config.LOCATION,
                                                        device=CLIENT.client_id,
                                                        topic=topic),
                   message,
                   retain=retain,
                   qos=qos)
    time.sleep_ms(sleep)


def mqtt_multipub(payload, sleep=0):
    """Send multiple data"""
    for pl in payload:
        mqtt_publish(pl[0], pl[1], retain=pl[2], qos=pl[3], sleep=sleep)


def mqtt_callback(topic, msg):
    """Callback en cas de changement de mode"""
    global CURRENT_MODE
    global BRIGHTN
    global PWR
    # print(topic, msg)
    t = topic.split(b'/')
    l = len(t)
    if t[l-1] == b'value' and msg.isdigit():
        msg = int(msg)
        # print(t[l-2], msg)
        if t[l-2] == b'mode' and msg in range(0, len(light_modes.MODES_LST)):
            # print("Nouveau mode: ", int(msg))
            CURRENT_MODE = msg
        if t[l-2] == b'brightness' and msg in range(0, 101):
            # print("Modif. luminosité :", int(msg))
            BRIGHTN = msg
            ENCODER.reset(BRIGHTN)
        if t[l-2] == b'power' and msg in (0, 1):
            PWR = bool(msg)
    else:
        print("Pas une valeur. Ignore...")



def power_cycle():
    """ON/OFF function"""

    PWR = not PWR
    print("Power :", PWR)
    if not PWR:
        # Extinction des LED
        NPXL_STRIP.fill((0, 0, 0))
        NPXL_STRIP.write()
    else:
        # Luminosité basse si on rallume avec une luminisité de 0
        if BRIGHTN == 0:
            BRIGHTN = 10

        # On remet l'encodeur à la dernière luminosité connue
        # pour ignorer les mouvements pendant l'extinction
        ENCODER.reset(BRIGHTN)

    # on attend que le bouton soit relâché
    while ENC_BUT.value() == 0:
        time.sleep_ms(10)

    # Envoi de la nouvelle info au MQTT
    mqtt_multipub([['power/date', now(rtc), True, 1],
                   ['power/value', "{:d}".format(PWR), True, 0]])


def update_button_c(button, button_state, pwr, mode):
    """Surveille le bouton C (clic central)"""
    global PWROFF_DELAY

    if button.value() == 0:
        # print("appui")
        if button_state == 1:
            # Deadline avant de changer l'état d'allumage
            PWROFF_DELAY = time.ticks_add(time.ticks_ms(), 2000)
            # print(PWROFF_DELAY)
        else:
            # Est-ce que la deadline est atteinte ? si oui on change l'état
            # d'allumage
            # print(PWROFF_DELAY, "/", time.ticks_ms())
            if time.ticks_diff(time.ticks_ms(), PWROFF_DELAY) >= 0:
                PWROFF_DELAY = 0
                power_cycle()
    else:
        if pwr and button_state == 0:
            # Si on a juste un clic rapide sur le bouton, on change de mode au
            # relachement du bouton
            mode = mode + 1
            print("Mode : {}/{}".format(CURRENT_MODE,
                                        len(light_modes.MODES_LST)))
            if mode >= len(light_modes.MODES_LST):
                mode = 0

            mqtt_multipub([['mode/date', now(rtc), True, 1],
                           ['mode/value', "{}".format(mode), True, 1]])

    return mode


def update_conn_led():
    """Refresh Wifi status LED"""
    WIFI_LED.value(not WLAN.isconnected())


#####################
# BOUCLE PRINCIPALE #
#####################

set_rtc()

# Envoi de l'état de départ au broker
data = [["last_boot", now(rtc), True, 1],
        ["location", config.LOCATION, True, 1],
        ["sys/wifi_ssid", WLAN.config('essid'), True, 0],
        ["sys/ip", WLAN.ifconfig()[0], True, 1],
        ["sys/esp8266_device_id", "{:d}".format(esp.flash_id()), True, 1],
        ['mode/date', now(rtc), True, 1],
        ["mode/value", "{}".format(CURRENT_MODE), True, 1],
        ['brightness/date', now(rtc), True, 1],
        ["brightness/value", "{}".format(BRIGHTN), True, 1],
        ['power/date', now(rtc), True, 1],
        ["power/value", "{:d}".format(PWR), True, 1],
        ]
mqtt_multipub(data)

# MQTT subscription
print("Set callback...")
CLIENT.set_callback(mqtt_callback)
print("OK")
print("Subcribe...")
CLIENT.subscribe(b"{location}/{device}/#".format(location=config.LOCATION,
                                                 device=CLIENT.client_id))
print("OK")
time.sleep(1)

BRIGHTN_DATA = None

while True:

    # Si on a un changement de valeur, on met à jour la luminosité
    if PWR:
        if BRIGHTN != ENCODER.value:
            print(ENCODER.value)
            BRIGHTN = ENCODER.value
            MQTT_TIMER = time.ticks_add(time.ticks_ms(), MQTT_DELAY * 1000)
            BRIGHTN_DATA = [['brightness/date', now(rtc), True, 1],
                            ['brightness/value', "{}".format(BRIGHTN), True, 0]]

        # Mise à jour des LED
        light_modes.update_neopixel(CURRENT_MODE,
                                    NPXL_STRIP,
                                    config.USR_COLOR,
                                    BRIGHTN)

    # Si on laisse appuyer 2 secondes: extinction
    CURRENT_MODE = update_button_c(ENC_BUT,
                                   BUTTN_STATE,
                                   PWR, CURRENT_MODE)

    BUTTN_STATE = ENC_BUT.value()

    update_conn_led()

    CLIENT.check_msg()

    # On n'envoie les données de luminosité qu'une fois stabilisée (MQTT_DELAY)
    if BRIGHTN_DATA is not None and time.ticks_diff(time.ticks_ms(), MQTT_TIMER) >= 0:
        mqtt_multipub(BRIGHTN_DATA)
        BRIGHTN_DATA = None

    time_to_test = rtc.datetime()

    if time_to_test[5] + time_to_test[6] == 0:    # (chaque heure quand min et sec = 0)
        print("Old Time: {0:04d}/{1:02d}/{2:02d}_{4:02d}:{5:02d}:{6:02d}.{7}".format(*time_to_test))
        set_rtc()
        print("Time set: {0:04d}/{1:02d}/{2:02d}_{4:02d}:{5:02d}:{6:02d}.{7}".format(*rtc.datetime()))