通信ドライバーからTrainを組み立てる
探索した機器、BleakDriver、TrainBleDevice、Trainの関係を確かめます。
- 必要な本体
- 1台。addressを自分の本体へ変更。
- 線路・準備
- 周回コース。原例の識別子は見本です。
- 終了方法
- 各色3秒の処理後に停止の送信・切断。
- 保存する場所
matrain-python-async/examples/bleak/raw_trainlib_connection.py- 実行する場所
matrain-python-asyncフォルダー内。.venvのPythonを使います。
Mac：.venv/bin/python examples/bleak/raw_trainlib_connection.pyWindows：.venv\Scripts\python.exe examples\bleak\raw_trainlib_connection.py
通常終了の処理を原例のまま掲載しています。例外やCtrl+Cでの停止が保証されないため、異常時は本体のボタンを短く押して止めます。切断だけを停止とみなさないでください。
掲載版の調整
Python 3.14/3.15での起動に合わせ、末尾のget_event_loop()/run_until_complete()をasyncio.run(..., debug=True)へ置き換えました。走行・色・待ち時間は変更していません。
コード全文
bleak/raw_trainlib_connection.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: Python 3.14/3.15での起動に合わせ、末尾のget_event_loop()/run_until_complete()をasyncio.run(..., debug=True)へ置き換えました。走行・色・待ち時間は変更していません。
import sys
import asyncio
import logging
import time
from bleak import BleakScanner
from bleak.exc import BleakError
from intelino.trainlib_async import Train
from intelino.trainlib_async.messages import TrainMsg
from intelino.trainlib_async.train_ble_device import TrainBleDevice
from intelino.trainlib_async.drivers.bleak_driver import BleakDriver
async def run(address: str, debug=False):
log = logging.getLogger(__name__)
log.setLevel(logging.DEBUG if debug else logging.INFO)
h = logging.StreamHandler(sys.stdout)
h.setLevel(logging.DEBUG)
log.addHandler(h)
device = await BleakScanner.find_device_by_address(address, timeout=5.0)
if not device:
raise BleakError(f"A device with address {address} could not be found.")
driver = BleakDriver(device)
train_device = TrainBleDevice(driver)
train = Train(train_device)
await train.connect()
log.info(f"{time.time():.3f} Connected: {train.is_connected}")
def callback_handler(msg: TrainMsg):
log.debug(f"{time.time():.3f} Received {msg.raw_packet}")
train.notifications.subscribe(callback_handler)
log.info(f"{time.time():.3f} Drive medium speed forward")
await train.send_command(0xB8, [0x01, 0x02, 0x01])
log.info(f"{time.time():.3f} Setting color to RED...")
color = [0xFF, 0, 0]
await train.send_command(0xB4, [0x06, *color, *color])
log.debug(f"{time.time():.3f} Setting color to RED DONE")
await asyncio.sleep(3.0)
log.info(f"{time.time():.3f} Setting color to GREEN...")
color = [0, 0xFF, 0]
await train.send_command(0xB4, [0x06, *color, *color])
log.debug(f"{time.time():.3f} Setting color to GREEN DONE")
await asyncio.sleep(3.0)
log.info(f"{time.time():.3f} Setting color to BLUE...")
color = [0, 0, 0xFF]
await train.send_command(0xB4, [0x06, *color, *color])
log.debug(f"{time.time():.3f} Setting color to BLUE DONE")
await asyncio.sleep(3.0)
log.info(f"{time.time():.3f} Stopping the train")
await train.send_command(0xB9, [0x01])
log.info(f"{time.time():.3f} Stopping notifications and disconnecting")
await train.disconnect()
log.info(f"{time.time():.3f} Done")
if __name__ == "__main__":
address = "33AA542E-1452-4525-8DB8-7BC941A7FE49"
asyncio.run(run(address, True), debug=True)
冒頭に配布条件と出典のコメントを追加しています。表示している全文と、個別保存・ZIP内のこのファイルは同じです。ライセンス・出典を読む