Zenn
🔲

ESP32 - MicroPythonで家電を制御しよう－その２（WEBリクエスト）

2021/08/07に公開
Python
電子工作
ESP32
MicroPython
tech

ESP32を使って、IFTTTサービスにリクエストを送り、家電を制御します。その仕組みとコードの部分について説明します。
全体像は、その１(コンセプト、スイッチ)を参照してください。
本記事は、WEBリクエスト(HTTPC管理スレッド)についての説明です。

関連記事

HTTPC管理スレッドの作成

HTTPC管理スレッドの作成と動作確認を行います。

開発環境の構築

開発環境の構築は、下記の記事を参考にしてください。

https://zenn.dev/kotaproj/articles/d969fb39100da443f41f
https://qiita.com/kotaproj/items/b53006aef9d04053a5ee

パーツ一覧

機材名 数量 備考
ESP32評価ボード 1 ESP32-WROVER 開発ボード/ESP32-WROOM 開発ボードのどちらでも可

回路図

なし

事前準備(IFTTT)

Appletの登録

IFTTTの設定手順について説明します。

  • https://ifttt.com/home へアクセスしてアカウントを作成 or ログイン
  • Createにて、以下を追加
    • This(Trigger)
      • "Webhooks" - "Receive a web request"
        • "Event Name" : sw1
    • That(Action)
      • "Nature Remo" - "Turn on air conditioner"
        • Cool 26℃

tokenの確認

webhooksをキックするためにtokenを確認します。

  • Webhooks Settingsのページより確認する
  • 下記画像のマスクされている箇所に記載されている

コード（HTTPC管理スレッド）

httpc.pyとutil.pyの2つのファイルで構成されています。

httpc.py
import _thread
from usocket import socket, AF_INET, SOCK_DGRAM
import network
import urequests as requests
import time
import gc

# from settings import SettingsFile
from util import *

SSID = "XXXXXXXXXXXXX"
PASSWORD = "YYYYYYYYYYYYY"
TOKEN = "ZZZZZZZZZZZZZZZZZ"

class HttpcProc():
    
    def __init__(self, lock=None, snd_que=None, rcv_que=None):
        # param
        # settings_file = SettingsFile()
        self._ssid = SSID
        self._password = PASSWORD
        self._token = TOKEN
        self._lock = lock
        self._snd_que = snd_que
        self._rcv_que = rcv_que
        return


    def do_connect(self):

        wlan = network.WLAN(network.STA_IF)
        wlan.active(True)
        if not wlan.isconnected():
            print('connecting to network...')
            # wlan.connect('essid', 'password')
            wlan.connect(self._ssid, self._password)
            while not wlan.isconnected():
                pass
                # time.sleep(1)
                time.sleep(3)
                print(".")

        # send_que(self._lock, self._snd_que, ("dst:pre,src:httpc,cmd:dsp,type:clear"))
        # send_que(self._lock, self._snd_que, ("dst:pre,src:httpc,cmd:led,type:off_all"))
        print('network config:', wlan.ifconfig())
        # send_que(self._lock, self._snd_que, ("dst:pre,src:httpc,cmd:led,type:blink,name:green"))
        return


    def _proc_httpc(self):

        def act_httpc(msg):

            def do_webhook(eventid, value1="dummy1", value2="dummy2", value3="dummy3"):
                # care memory
                gc.collect()

                # valueに載せる情報
                payload = "value1="
                # payload += str(self._rtc.datetime())
                payload += value1
                payload += "&value2="
                payload += value2
                payload += "&value3="
                payload += value3

                url = "http://maker.ifttt.com/trigger/" + eventid + "/with/key/" + self._token

                ret = True
                try:
                    # response = requests.post(url, headers={"Content-Type": "application/x-www-form-urlencoded"}, data=payload)
                    # value指定が不要な場合は↓でok
                    response = requests.post(url)
                    # send_que(self._lock, self._snd_que, ("dst:pre,src:httpc,cmd:led,type:blink,name:blue"))
                    response.close()
                except:
                    # send_que(self._lock, self._snd_que, ("dst:pre,src:httpc,cmd:led,type:blink,name:red"))
                    ret = False
                return ret

            print("act_httpc:run")
            d = conv_msg2dict(msg)
            print(d)
            cmd, typ = d["cmd"], d["type"]

            if "sw" in cmd:
                how = d["how"]
                evt_id = conv_typ2eventid(typ, how)
                print(evt_id)
                if evt_id is not None:
                    # send_que(self._lock, self._snd_que, ("dst:pre,src:httpc,cmd:dsp,type:ifttt,how:"+evt_id+",sts:sending...,tmr:3000"))
                    do_webhook(evt_id)
                    # send_que(self._lock, self._snd_que, ("dst:pre,src:httpc,cmd:dsp,type:ifttt,how:"+evt_id+",sts:sended.,tmr:3000"))
            else:
                print("act_httpc:error, ", msg)
            return


        is_connect = False

        while True:

            if is_connect is False:
                self.do_connect()
                is_connect = True

            msg = recv_que(self._lock, self._rcv_que)
            if msg is None:
                time.sleep_ms(50)
                continue

            print("proc_httpc:msg - ", msg)
            ret = act_httpc(msg)
            if False == ret:
                # reconnect
                is_connect = False
        return


    def run(self):
        _thread.start_new_thread(self._proc_httpc, ())


def main():

    lock = _thread.allocate_lock()

    que_pre2httpc = []
    httpc_proc = HttpcProc(lock, snd_que=None, rcv_que=que_pre2httpc)
    httpc_proc.run()
    time.sleep_ms(5_000)
    send_que(lock, que_pre2httpc, ("dst:httpc,src:pre,cmd:sw,type:no1,how:released"))
    time.sleep_ms(10_000)
    return

if __name__ == "__main__":
    main()
util.py
# 通知とIFTTT-Webhookイベントの紐づけ
#  - cmd:sw,type:no1,how:released" => sw1をイベントIDとしてIFTTTにリクエスト
KEY_TO_EVENTID = {
    "no1" : {
        "pressed" : None,
        "released" : "sw1",
        "long" : None,
    },
}

def conv_msg2dict(msg):
    # input : "a:b,c:d"
    # output: {"a":"b","c":"d"}
    d = {}
    for item in msg.split(','):
        key, val = item.split(':')
        d[key] = val
    return d

def conv_typ2eventid(typ, how):
    if typ in KEY_TO_EVENTID:
        return KEY_TO_EVENTID[typ][how]
    return None

def send_que(lock, que, value):
    if que is None:
        print("Error:que is None.")
        return
    # lock
    lock_p = lock.acquire(1, -1) #wait forever
    if not lock_p:
        print("Error:task can not get the lock.")
    else:
        que.append(value)
        lock.release()
    # unlock
    return

def recv_que(lock, que):
    if que is None:
        print("Error:que is None.")
        return

    val = None

    # lock
    lock_p = lock.acquire(1, -1) #wait forever
    if not lock_p:
        print("Error:task can not get the lock.")
    else:
        if len(que) > 0:
            val = que[0]
            del que[0]
        lock.release()
    # unlock
    return val

実行方法と結果

>>> import httpc
>>> httpc.main()
connecting to network...
.
network config: ('192.168.XX.xx', '255.255.255.0', '192.168.XX.1', '192.168.XX.1')
proc_httpc:msg -  dst:httpc,src:pre,cmd:sw,type:no1,how:released
act_httpc:run
{'src': 'pre', 'dst': 'httpc', 'cmd': 'sw', 'how': 'released', 'type': 'no1'}
sw1
>>>

→エアコンをオンすることができました。

ポイント

SSID/PASSWORD/TOKENの指定

  • SSID/PASSWORDには、使用するアクセスポイントの設定となります
    • 2.4GHz帯のみ使用、5GHzは使用不可
  • TOKEN
    • 事前準備で確認したIFTTT-WebhooksのTokenを指定

WiFiの接続

  • do_connect()でWiFiへの接続を行う
    • 接続するまで本処理は抜けない

IFTTTへのリクエスト／メモリケア／例外

  • do_webhook()にてリクエストを行う
    • メモリケア
      • gc.collect()にて能動的にガベレージコレクションを実施
      • response.close()にてクローズ（メモリの開放）
        • 評価ボードがSPI-RAMなし版の場合、メモリがプア
        • HTTPリクエスト時はメモリが多く消費されるため行っている
    • 例外処理 - try,except
      • WiFiの切断している場合などは例外があがる
      • 本例外は、切断していると判断しWiFiへの再接続を行う

IFTTTへのリクエストでvalueを指定

  • 今回は使用していないが、valueをのせる場合は、headerの指定が必要
    • response = requests.post(url, headers={"Content-Type": "application/x-www-form-urlencoded"}, data=payload)
    • urequestsモジュール内で、ヘッダが固定文字になっているため

スレッド間通信

  • MicroPythonには、Queueモジュールが搭載されていない
  • 代用として通常のリストをキューとして使用している
    • キューの送信 : send_que()
    • キューの受信 : recv_que()

参考URL

  • MicroPython公式ドキュメント

https://micropython-docs-ja.readthedocs.io/ja/latest/

  • MicroPython - ESP32 用クイックリファレンス

https://micropython-docs-ja.readthedocs.io/ja/latest/esp32/quickref.html

さいごに

次は、LED/Displayの制御を記載する予定です。

GitHubで編集を提案

Discussion