Programmable LED 〜Slackから点灯パターンを設定できるクリスマスツリー用のLEDデバイスを作ってみました〜

import os
import time
import json
import machine
import network
from simple import MQTTClient

state = "OK"

wifi_ssid = "XXXX"
wifi_password = "XXXXXXXX"

aws_endpoint = b"xxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com"
private_key = "private.pem.key"
private_cert = "cert.pem.crt"


pattern_list = ["1111100000", "1111100000", "1111100000", "1111100000"]

with open(private_key, "r") as f:
    key = f.read()
with open(private_cert, "r") as f:
    cert = f.read()

device_id = "0001"
topic_sub = "christmas-tree/" + device_id
ssl_params = {"key": key, "cert": cert, "server_side": False}

led_list = []
led_list.append(machine.Pin(12, machine.Pin.OUT))  # GPIO12 L1 WHITE
led_list.append(machine.Pin(14, machine.Pin.OUT))  # GPIO14 L2 RED
led_list.append(machine.Pin(27, machine.Pin.OUT))  # GPIO27 L3 YELLOW
led_list.append(machine.Pin(26, machine.Pin.OUT))  # GPIO26 L4 BLUE

wlan = network.WLAN(network.STA_IF)
wlan.active(True)
if not wlan.isconnected():
    print("Connecting to network...")
    wlan.connect(wifi_ssid, wifi_password)
    while not wlan.isconnected():
        pass
    print("Connection successful")
    print("Network config:", wlan.ifconfig())


def mqtt_connect(client=device_id, endpoint=aws_endpoint, sslp=ssl_params):
    mqtt = MQTTClient(
        client_id=client,
        server=endpoint,
        port=8883,
        keepalive=1200,
        ssl=True,
        ssl_params=sslp,
    )
    print("Connecting to AWS IoT...")
    mqtt.connect()
    print("Connected.")
    return mqtt


def mqtt_subscribe(topic, msg):
    print("on received...")
    print("msg.decode(): {}".format(msg.decode()))
    global pattern_list

    for i, color in enumerate(["white", "red", "yellow", "blue"]):
        pattern_list[i] = json.loads(msg.decode())[color]

    # 一番長いパターンに合わせるため、不足分に"0"を追加する
    length_list = []
    for i in range(len(pattern_list)):
        length_list.append(len(pattern_list[i]))
    max_length = max(length_list)
    print("length_list:{} max_length: {}".format(length_list, max_length))
    for i in range(len(pattern_list)):
        for _ in range(max_length - length_list[i]):
            pattern_list[i] = pattern_list[i] + "0"

    for i in range(len(pattern_list)):
        print("pattern_list[{}]: {}".format(i, pattern_list[i]))


try:
    mqtt = mqtt_connect()
    mqtt.set_callback(mqtt_subscribe)
    mqtt.subscribe(topic_sub)
except:
    print("Unable to connect to MQTT.")


index = 0

while True:
    try:
        mqtt.check_msg()
    except:
        print("Unable to check for messages.")

    time.sleep(0.1)

    index += 1
    if index >= len(pattern_list[0]):
        index = 0

    for n in range(4):
        pattern = pattern_list[n][0 + index : 1 + index]
        led_list[n].value(int(pattern))

    if index % 20 == 0:
        print("pattern_list: {}".format(pattern_list))

import json
import requests # requests を使用するためにLayerが必要
import boto3

url = "https://hooks.slack.com/services/xxxxxxxx/xxxxxxxx/xxxxxxxxxxxxDERSV1MxZ"
respone_message = "設定しました"


def text_to_pattern(text):
    lines = text.split("\n")
    pattern = {"white": "", "red": "", "yellow": "", "blue": ""}
    for line in lines:
        tmp = line.split(":")
        pattern[tmp[0].strip()] = tmp[1].strip()
    return pattern


def send_pattern_to_mqtt(pattern):
    topic = "christmas-tree/0001"
    iot = boto3.client("iot-data")
    payload = pattern
    iot.publish(topic=topic, qos=1, payload=json.dumps(payload, ensure_ascii=False))


def send_response_to_slack(pattern):
    data = {"text": "{}\n{}".format(respone_message, json.dumps(pattern))}
    response = requests.post(url, data=json.dumps(data))
    print("slack response: {} {}".format(response.status_code, response.text))


def job(text):
    pattern = text_to_pattern(text)
    send_response_to_slack(pattern)
    send_pattern_to_mqtt(pattern)


def lambda_handler(event, context):
    # Enable Eventsの認証時に使用
    # return {
    #     'statusCode': 200,
    #     'body': json.dumps(event["body"])
    # }

    # Lambdaが投稿したrespone_messageが含まれていないものを処理対象としている
    if not respone_message in event:
        if "body" in event:
            body = json.loads(event["body"])
            if "event" in body:
                event = body["event"]
                if "text" in event:
                    job(event["text"])