先頭役が読んだ色を3台で共有する

1台を先頭役にし、その色をほかの2台にも送ります。

必要な本体: 3台。先頭役の上部ライトは緑。
線路・準備: 先頭役に色を読ませるレール。コードからは走行を開始しません。
終了方法: Enterで先頭役の購読を解除し、3台を切断。コードからの停止操作はありません。
保存する場所: matrain-python-async/examples/color_change/2_multiple_trains_with_a_leader.py
実行する場所: matrain-python-asyncフォルダー内。.venvのPythonを使います。

Mac：.venv/bin/python examples/color_change/2_multiple_trains_with_a_leader.pyWindows：.venv\Scripts\python.exe examples\color_change\2_multiple_trains_with_a_leader.py

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

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

コード全文

color_change/2_multiple_trains_with_a_leader.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.

"""Color change event example (2): Single leader and multiple follower trains."""
import asyncio
import aioconsole
from typing import List
from intelino.trainlib_async import (
    TrainScanner,
    Train,
)
from intelino.trainlib_async.messages import (
    TrainMsg,
    TrainMsgEventFrontColorChanged,
)


async def color_change(leader: Train, followers: List[Train]):
    """
    Wait for color change events on the leader train and set
    headlights/taillights based on the seen color on all the trains.
    """

    def observer_func(msg: TrainMsg):
        if isinstance(msg, TrainMsgEventFrontColorChanged):
            print(f"New color: {msg.color}")
            color = msg.color.to_rgb_bytes()
            asyncio.create_task(leader.set_headlight_color(color, color))
            for follower in followers:
                asyncio.create_task(follower.set_headlight_color(color, color))

    # subscribe to the notification stream
    subscription = leader.notifications.subscribe(observer_func)

    # highlight the leader train
    await leader.set_top_led_color(0, 255, 0)

    # let the train program run until the 'enter' key is pressed
    await aioconsole.ainput("Press <Enter> to exit at any time...\n")

    # cleanup
    subscription.dispose()


async def main():
    # connect to the first 3 trains discovered
    trains = await TrainScanner().get_trains(count=3)

    # run our train program
    await color_change(trains[0], trains[1:])
    # after the program finished, disconnect from the trains
    for train in trains:
        await train.disconnect()


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

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

作品・遊びの一覧へ