🔲
ESP32 - MicroPythonで家電を制御しよう-その1(コンセプト、スイッチ)
ESP32を使って、IFTTTサービスにリクエストを送り、家電を制御します。その仕組みとコードの部分について説明します。
本記事は、コンセプトとスイッチスレッドについての説明です。
関連記事
コンセプト
何をつくるのか説明します。
全体像
以下が全体像となります。
- 物理スイッチを押すことでWEBリクエストを行う
- WEBリクエスト(Webhooks)をIFTTTサービスにトリガーをかける
- IFTTTと連携しているサービスを使い、家電を制御する
ソフト設計
以下が今回作るソフトウェア内の各スレッドの関係となります。
- スイッチ管理スレッドは、スイッチが押されるとイベント管理スレッドに通知する
- イベント管理スレッドは、通知を受けて必要なスレッドへ橋渡しをする
- ディスプレイ管理スレッドは、通知を受けてOLEDに文字列を表示する
- HTTPC管理スレッドは、通知を受けてIFTTTにリクエストを発行する
- LED管理スレッドは、通知を受けてLEDの点灯/消灯する
機材
機材名 | 備考 |
---|---|
ESP32評価ボード | ESP32-WROVER 開発ボード/ESP32-WROOM 開発ボードのどちらでも可 |
スイッチ | 最大8つまで |
LED | 最大3つまで |
抵抗 | スイッチ用プルアップ抵抗, LED用抵抗 |
液晶 | OLED128x64dot |
スイッチ管理スレッドの作成
スイッチ管理スレッドの作成と動作確認を行います。
開発環境の構築
開発環境の構築は、下記の記事を参考にしてください。
パーツ一覧
機材名 | 数量 | 備考 |
---|---|---|
ESP32評価ボード | 1 | ESP32-WROVER 開発ボード/ESP32-WROOM 開発ボードのどちらでも可 |
スイッチ | 3 | 3つで説明 |
抵抗 | 2 | 10kΩ ※R6, R7に該当(3つのスイッチで説明) |
回路図
コード(スイッチ管理スレッド)
sw.py
from machine import Pin
import _thread
import time
# from util import send_que
TACT_SW_DEFs = {
"no1": (35, "in", 0),
"no2": (33, "in", 0),
"no3": (19, "in_pullup", 0),
}
# 50msec
TACT_JUDGE_PRESS = [False, False, True, True]
# 200sec
TACT_JUDGE_LONG = [False, False,
True, True, True, True, True, True, True, True]
# released
TACT_JUDGE_RELEASE = [True, True, False, False]
class TackSwitch:
def __init__(self, pin, mode, push_logic):
if mode == "in":
self.pin = Pin(pin, Pin.IN)
pass
elif mode == "in_pullup":
self.pin = Pin(pin, Pin.IN, Pin.PULL_UP)
else:
raise Exception('')
self.store = []
self.long_store = []
self.push_logic = push_logic
self.long_evt = False
def read(self):
value = self.pin.value()
if value == self.push_logic:
return True
else:
return False
def read_poll(self):
logic = self.read()
# 短押し
self.store.append(logic)
if len(self.store) <= len(TACT_JUDGE_PRESS):
return False, None
self.store.pop(0)
if TACT_JUDGE_PRESS == self.store:
return True, "pressed"
# スイッチ離す
if TACT_JUDGE_RELEASE == self.store:
if self.long_evt:
self.long_evt = False
return False, None
return True, "released"
# 長押し
self.long_store.append(logic)
if len(self.long_store) <= len(TACT_JUDGE_LONG):
return False, None
self.long_store.pop(0)
if TACT_JUDGE_LONG == self.long_store:
self.long_evt = True
return True, "long"
# イベントなし
return False, None
class SwProc():
def __init__(self, lock=None, snd_que=None, rcv_que=None):
self._lock = lock
self._tsws = {k: TackSwitch(v[0], v[1], v[2]) for k, v in TACT_SW_DEFs.items()}
self._snd_que = snd_que
self._rcv_que = rcv_que
return
def _proc_poll(self):
print("_proc_poll - run")
while True:
time.sleep_ms(25)
for key, tact_sw in self._tsws.items():
sw_evt, sw_how = tact_sw.read_poll()
if sw_evt:
print("dst:pre,src:sw,cmd:sw" + ",type:" + str(key) + ",how:" + sw_how)
# send_que(self._lock, self._snd_que, ("dst:pre,src:sw,cmd:sw" + ",type:" + str(key) + ",how:" + sw_how))
def run(self):
_thread.start_new_thread(self._proc_poll, ())
def main():
lock = _thread.allocate_lock()
sw_proc = SwProc(lock)
sw_proc.run()
time.sleep(10)
if __name__ == "__main__":
main()
実行方法と結果
>>> import sw
>>> sw.main()
_proc_poll - run
dst:pre,src:sw,cmd:sw,type:no1,how:pressed <=スイッチ1を短押し
dst:pre,src:sw,cmd:sw,type:no1,how:released <=スイッチ1を離す
dst:pre,src:sw,cmd:sw,type:no2,how:pressed <=スイッチ2を短押し
dst:pre,src:sw,cmd:sw,type:no2,how:released <=スイッチ2を離す
dst:pre,src:sw,cmd:sw,type:no3,how:pressed <=スイッチ3を短押し
dst:pre,src:sw,cmd:sw,type:no3,how:released <=スイッチ3を離す
dst:pre,src:sw,cmd:sw,type:no2,how:long <=スイッチ2を長押し
ポイント
スイッチの検出
- スイッチの検出は、GPIOをインプット指定します
- Pin(pin, Pin.IN) ←外部抵抗でプルアップする場合(SW1, SW2)
- Pin(pin, Pin.IN, Pin.PULL_UP) ←内部抵抗でプルアップする場合(SW3)
- スイッチが押されている場合は0( pin.value() )となる
- スイッチが押されていない場合は1( pin.value() )となる
- スイッチの状態は、ポーリングにて25msecに一回状態を取得します
- 取得されるデータがTACT_JUDGE_***と一致している場合、イベントとして検出する
- pressed, released, longの3種類のイベントを検出する
- ※便宜上、longのイベントを検出した場合、releasedは検出しないようにしている
スレッドについて
- MicroPythonではthreadingではなく、_threadの対応です
- 低レイヤーの対応となる
- あとで使用するために、lock, queue(実体はリスト)を渡せるようにしている
参考URL
- MicroPython公式ドキュメント
- MicroPython - ESP32 用クイックリファレンス
さいごに
次は、IFTTTにリクエストを投げるところを記載する予定です。
Discussion