Python-LEGO Mindstormsのセンサーをスレッドで非同期処理する

今回はLEGO Mindstorms EV3(以降EV3)のセンサーを複数待機状態にして、それぞれがしきい値に達した際になにかしらのアクションを行わせる。センサーごとにスレッドを定義し、それぞれのスレッドを非同期に処理させていく。

次回ではライントレースも合わせて行っている。

EV3について

教育版 LEGO® MINDSTORMS EV3

本記事内での環境

• PC
  Windows10
  Python 3.7.3
  開発環境 VisualStudioCode

• EV3
  ev3dev
  APIリファレンス

環境構築やソースコードの作成、実行方法はこちら

EV3のモデル

今回は超音波センサーとタッチセンサーが付いている。
モーター等は利用しないが安定して設置するために付いている。

threading

Pythonにはいくつかの種類の非同期処理ライブラリがあるが、今回はthreadingを使って複数スレッドをたて、センサーの非同期処理をしていく。

スレッドとしては、タッチセンサーが押されたことを判定するスレッドと、超音波センサーが15cm以内に物体を検知したことを判定するスレッドを用意して、それぞれ非同期で処理を行う。

以下がコード。

thread_test1.py

from ev3dev2.sound import Sound
from ev3dev2.motor import LargeMotor, OUTPUT_B, OUTPUT_C
from ev3dev2.sensor import INPUT_1, INPUT_4
from ev3dev2.sensor.lego import TouchSensor, UltrasonicSensor
import time
import threading

sound = Sound()
sound.set_volume(30)
touch = TouchSensor()
ultrasonic = UltrasonicSensor()


def wtouch():    
    while True:
        if touch.is_pressed:
            print('Touched')
            sound.tone(1000, 200)
            time.sleep(1.0)
            break
        

def wsonic():
    while True:
        if ultrasonic.distance_centimeters < 15:
            print('Detected')
            sound.tone(500, 200)
            time.sleep(1.0)
            break
        

def main():
    th1 = threading.Thread(target = wtouch)
    th2 = threading.Thread(target = wsonic)

    th1.start()
    th2.start()
    print('Thread is started')


if __name__ == '__main__':
    main()

関数としてdef wtouchでタッチセンサーの判定を用意し、
def wsonicで超音波センサーの判定を用意している。
判定時にはprintでコンソールに文字を出力するのに加えそれぞれ異なるビープ音がEV3から鳴るようになっている。

またdef mainでそれぞれの関数をスレッドとして設定し、
スレッドを開始させている。開始時にはコンソールに文字を出力している。

実行

実行してみると非同期でセンサーの判定が行われているのがわかる。
以下では2回実行していて、1回目は超音波、タッチの順で判定してプログラムが終了しており、2回目ではタッチ、超音波の順で判定してプログラムが終了している。