LEGO MINDSTORMS Pythonでなんちゃってスレッド処理

LEGO MINDSTORMS Robot Inventor

こちらに紹介ページはあるのですが「今すぐ購入」ボタンを押すとページは存在せず公式サイトで購入することはできません。日本では教育用のSPIKEプライムセットしか購入できません。SPIKEは教育用なのでパーツを入れるプラスチックケース入りなのに対してInventorはホビー向けなのでケースは紙製となっています。ただ、標準でモータが4つついていたりLEGOパーツの数が多かったりと個人で遊ぶのにはInventorの方が良いので、米国Amazonで購入することにしました。

本体の色がSPIKEは黄色なの対してInventorは青系の色となっています。またプログラム作成用ツールもSPIKE用とInventor用は別になっています。SPIKE用は日本語化されているのですがInventor用は日本で発売されていないこともあってか日本語化はされていません。

最近は小学校でも英語教育が必須化されたりしていますが、遊びの中で英語に触れる（LEGOなので殆ど言葉を利用することはありませんがw）ことが自然と英語を学ぶ良い方法なのかと思ったりします。

プログラミング用言語としてはScratchベースのブロックプログラミングとPythonが利用できます。PythonはMicroPythonなのでthreadingは入っていませんし、ローレベルのAPIである_threadも利用することはできません。利用できるのは非同期 I/O スケジューラuasyncio（非同期処理を扱うための標準ライブラリasyncioのサブセット）となるのでコールチンを利用した非同期処理のコードを書くことは可能となります。

ただ、ここで大切なのはマルチタスクやマルチスレッド、非同期処理といった概念ではなく複数の処理があたかも同時に処理されているように見えることが子供達にとっては大切だということだと思います。

MINDSTORMSでPythonは必要か？

Scratchでプログラムを作成する場合、まずはゲーム的なプログラムを作ることが多いと思いますが、その場合それぞれのスプライト用にプログラムを作成する考え方は直感的ですし、それを動かす場合に非同期処理（ScratchはNode.jsをベースに開発されているのでシングルスレッド・シングルプロセスが基本）を意識せずにマルチスレッド的な動作が自然と実践できるScratchの考え方は非常に優れていると思います。

さてそれではMINDSTORMSをPythonでプログラムする場合マルチスレッドは必要になるのかな？という疑問が出てきます。

ロボット工学は畑違いですが、MINDSTORMSのプログラムを作っていて感じるのはマルチスレッドよりもステートマシンでプログラムした方がすっきりとできるかなという感じです。

MINDSTORMS用のScratchは必要なブロックもきめ細かく用意されているのであえてPythonでプログラムを作成する必要はないかなと思います。Pythonを使ってプログラムしたい方はプログラミング言語を勉強したい方が取り組むべきかなと感じました。

Pythonジェネレータベースのコールチン

ということでまずはジェネレータベースの非同期処理を作ってみました。

Scratchの場合はスプライトという単位でプログラムが分けられますが、MINDSTORMS用のPythonではファイルの分割は出来なそうです。ある程度の規模のプログラムになるとできればファイル分割できた方が管理がやりやすいのですがね。

ノンプリエンティブなマルチタスクになりますので適時「yield True」でCPUを解放するのがポイントになります。またMainの__done()は非同期処理の対象外になっているので、ここで長時間CPUを占有すると非同期処理に影響を与えるので極力小さい処理にする必要があります。

LEGOでの注意事項としてはモータ制御はstart、stopで行うようにしてmoveは（同期処理になるので、短い距離では問題ないと思いますが）極力使用しないように、またセンサーのwait系も使用しないようにします。

from mindstorms import MSHub, Motor, MotorPair, ColorSensor, DistanceSensor, App
from mindstorms.control import wait_for_seconds, wait_until, Timer
from mindstorms.operator import greater_than, greater_than_or_equal_to, less_than, less_than_or_equal_to, equal_to, not_equal_to
import math

# オブジェクトの定義
app = App()
hub = MSHub()
wheels = MotorPair(MSHub.PORT_B, MSHub.PORT_A)
arm = Motor(MSHub.PORT_C)
distance_sensor = DistanceSensor(MSHub.PORT_D)
color_sensor = ColorSensor(MSHub.PORT_E)

# ステートの定義
STATE_START = 1
STATE_EXIT = -1

# ジェネレータクラス
class Generator1:
    def __init__(self):
        self.__distance = -1
        self.__loop = True

    @property
    def distance(self):
        return self.__distance

    def stop(self):
        self.__loop = False

    # ジェネレータ関数本体
    def loop(self):
        while self.__loop:
            yield from self.__done()
        yield False

    def __done(self):
        self.__distance = distance_sensor.get_distance_cm()
        print(self.__distance)
        yield True
        if self.__distance and self.__distance < 15:
            print("The wall is near!")
            yield True

# メイン処理
class Main:
    def __init__(self):
        self.setup()
        self.loop()

    def setup(self):
        # ジェネレータの登録
        self.generators = []
        self.distance_sensor = Generator1()
        self.generators.append(self.distance_sensor)
        # ジェネレータ関数の登録
        self.generatorLoops = []
        for g in self.generators:
            self.generatorLoops.append(g.loop())

    def loop(self):
        while self.state != STATE_EXIT:
            if hub.right_button.is_pressed():
                for i, g in enumerate(self.generators):
                    g.stop()
                    while (next(self.generatorLoops[i])):
                        pass
                self.state = STATE_EXIT
            else:
                # ジェネレータの次の要素を取得（処理の再開）
                for loop in self.generatorLoops:
                    next(loop)
                # 自身の処理
                self.__done()
        
    def __done(self):
        print("Main keep alive!")

# プログラム開始指示を待つ
hub.left_button.wait_until_pressed()
hub.speaker.beep()
# プログラム開始
Main()
# プログラム終了（メニューに戻る）
raise SystemExit