TrainFactoryから2台を並行して動かす

取得した2台へ同じ処理を並行して実行します。

必要な本体: 2台。先に見つかった2台を取得。
線路・準備: 2台が安全に走れる周回コース。
終了方法: 各台約10秒の処理後に購読解除・停止・切断。
保存する場所: matrain-python-async/examples/bleak/two_trains_with_trainlib.py
実行する場所: matrain-python-asyncフォルダー内。.venvのPythonを使います。

Mac：.venv/bin/python examples/bleak/two_trains_with_trainlib.pyWindows：.venv\Scripts\python.exe examples\bleak\two_trains_with_trainlib.py

通常終了の処理を原例のまま掲載しています。例外やCtrl+Cでの停止が保証されないため、異常時は本体のボタンを短く押して止めます。切断だけを停止とみなさないでください。

掲載版の調整

SDK 1.0.5ではTrainMsgがパッケージ直下から公開されないため、intelino.trainlib_async.messagesから読み込みます。型以外の実処理は変更していません。

このPythonファイルを保存共通ファイルも一式保存（ZIP）

コード全文

bleak/two_trains_with_trainlib.py

# Matrain publication modification: licensing/source notice added.
# Executable behavior is unchanged except for the adaptations documented below.
# Source: https://github.com/intelino-code/intelino-trainlib-async-py
# Source commit: 298e1fd4091e069180c30f338b4c285b13accbd0
# This software is made available by Innokind, Inc. DBA Intelino, under the
# terms of the Intelino Public License Agreement, Version 1.0 located at
# https://intelino.com/intelino-public-license. BY INSTALLING, DOWNLOADING,
# ACCESSING, USING OR DISTRIBUTING ANY OF THE SOFTWARE, YOU AGREE TO THE TERMS OF
# SUCH LICENSE AGREEMENT.
# Publication adaptation: SDK 1.0.5ではTrainMsgがパッケージ直下から公開されないため、intelino.trainlib_async.messagesから読み込みます。型以外の実処理は変更していません。

import asyncio
import random
from intelino.trainlib_async import Train
from intelino.trainlib_async.messages import TrainMsg
from intelino.trainlib_async.train_factory import TrainFactory
from intelino.trainlib_async.enums import SpeedLevel


async def train_program(train: Train):
    print(f"starting train program {train.id}")

    def callback(data: TrainMsg):
        print(train.id, data)

    subscription = train.notifications.subscribe(callback)

    try:
        print("drive medium speed forward")
        await train.drive_at_speed_level(SpeedLevel.LEVEL2)

        for _ in range(10):
            print("setting color...")
            color = [random.randint(0, 1) * 0xFF for i in range(3)]
            await train.set_headlight_color(front=color, back=color)
            await asyncio.sleep(1.0)

        subscription.dispose()

        print("stop movement")
        await train.stop_driving()

        print(f"disconnecting from {train.id}")
        await train.disconnect()

    except Exception as e:
        print(e)


async def main():
    # Note: in case we would need to connect to specific trains, we can do that too.
    # addresses = [
    #     "33AA542E-1452-4525-8DB8-7BC941A7FE49",
    #     "CD3C3305-36E2-4DE7-94F3-EC7FFF59E1F8",
    # ]
    # trains = [await TrainFactory.create_train(address, timeout=5.0) for address in addresses]
    # trains = list(filter(None, trains))

    # connect to the first 2 trains discovered
    trains = await TrainFactory.create_trains(count=2, timeout=5.0)

    await asyncio.gather(*(train_program(train) for train in trains))


if __name__ == "__main__":
    asyncio.run(main())

冒頭に配布条件と出典のコメントを追加しています。表示している全文と、個別保存・ZIP内のこのファイルは同じです。ライセンス・出典を読む

作品・遊びの一覧へ