5秒ごとに速さを表示する

本体から受け取った設定速度と実際の速度を表示します。

必要な本体: 1台。
線路・準備: 分岐のない周回コース。
終了方法: Enterで購読解除・停止後に切断。
保存する場所: matrain-python-async/examples/data_streams/movement_stream.py
実行する場所: matrain-python-asyncフォルダー内。.venvのPythonを使います。

Mac：.venv/bin/python examples/data_streams/movement_stream.pyWindows：.venv\Scripts\python.exe examples\data_streams\movement_stream.py

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

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

コード全文

data_streams/movement_stream.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.

"""Data stream example (1): Print speed every 5 seconds."""
import asyncio
import aioconsole
from intelino.trainlib_async import TrainScanner, Train
from intelino.trainlib_async.messages import TrainMsgMovement
from intelino.trainlib_async.helpers import AsyncObserver


async def data_streams(train: Train):
    """
    Print current (and desired) speed every 5 seconds.
    """

    async def movement_observer_func(msg: TrainMsgMovement):
        print(
            f"Set speed: {msg.desired_speed_cmps} cm/s",
            f"Current speed: {msg.speed_cmps} cm/s",
        )

    # enable the movement notification stream every 5 seconds
    # Note: since we use it only for one purpose, we can do this
    movement_stream = await train.movement_notification_stream(
        streaming=True,
        interval=5.0,
    )
    subscription = movement_stream.subscribe(AsyncObserver(movement_observer_func))

    # start driving
    await train.drive_at_speed(35)

    # 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()
    await train.stop_driving()


async def main():
    async with TrainScanner() as train:
        # run our train program
        await data_streams(train)


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

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

作品・遊びの一覧へ