Python-SPIKE プライムで非同期処理 #Python

今回はLEGO SPIKEPrimeのPython環境でマルチタスク(非同期処理)を実装してみたので備忘録として残す。

SPIKE アプリケーション内のPythonエディターでコーディングしていくのだが、アプリケーション内のナレッジベースにはSPIKEPrimeを制御するために使うことができるメソッドについての解説は網羅されているが、MicroPythonから提供されているライブラリに関する情報はほとんど掲載されていないため、MicroPython環境で利用できそうな非同期処理用のライブラリを検索した。(かつpipなどによるインストールが不要なものである必要がある)

機体

今回のソースコードを動かすために用意した機体は以下のような簡易的なもの。
画像では車型になっているが、ここではフォースセンサーと距離センサーが使えればいいのでそれら２つが固定されていれば問題ない。

非同期処理について

今回非同期処理のためにuasyncioというライブラリを利用した。
公式のドキュメントはこちらから確認できる。

また、参考にした記事はこちら

ソースコード

以下が今回実装したソースコード。
フォースセンサーが押されたかどうかと、距離センサーが10cm以内に物体を検知したかどうかを同時に監視している。

from spike import PrimeHub, LightMatrix, Button, StatusLight, ForceSensor, MotionSensor, Speaker, ColorSensor, App, DistanceSensor, Motor, MotorPair
from spike.control import wait_for_seconds, wait_until, Timer
from math import *
import uasyncio


hub = PrimeHub()

f_sensor = ForceSensor('A')
d_sensor= DistanceSensor('D')

async def touch():
    if f_sensor.is_pressed():
        hub.light_matrix.show_image('HAPPY')
        print('touched')

async def dis():
    distance = d_sensor.get_distance_cm()
    if distance == None:
        distance = 200
    elif distance < 10:
        hub.light_matrix.show_image('SMILE')
        print('detected')

async def main():
    hub.light_matrix.show_image('ANGRY')
    uasyncio.create_task(touch())
    uasyncio.create_task(dis())
    await uasyncio.sleep(0.2)

while True:
    uasyncio.run(main())

実行結果

プログラムを実行すると、ライトマトリクスがANGRYの表示になり、フォースセンサーの監視をするタスクと距離センサーの監視をするタスクが非同期で動いている状態になる。

非同期処理なので、フォースセンサーと距離センサーのどちらに関するアクションを行っても問題ないが、フォースセンサーを押すとライトマトリクスの表示がHAPPYに変わる。

距離センサーに手をかざすなどして10cm以内の検知を行わせると、ライトマトリクスの表示がSMILEに変わる。

SPIKEのPython環境ではuasyncioが問題なく利用できることがわかった。