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

2つの接続をgatherで並行させ、各台のライトを10回変えます。

必要な本体: 2台。addressesの2つを自分の本体へ変更。
線路・準備: 2台が安全に走れる周回コース。
終了方法: 各台約10秒の処理後、通知解除・停止パケットを送り、接続を終了。
保存する場所: matrain-python-async/examples/bleak/two_trains_with_bleak.py
実行する場所: matrain-python-asyncフォルダー内。.venvのPythonを使います。

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

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

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

コード全文

bleak/two_trains_with_bleak.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.

from bleak import BleakClient
import asyncio
import random


COMMAND_CHARACTERISTIC = "40c540d0-344c-4d0d-a1da-9cc260b82d43"
RESPONSE_CHARACTERISTIC = "a4b80869-a84c-4160-a3e0-72fa58ff480e"


async def train_program(address: str):
    print(f"starting train program {address}")
    async with BleakClient(address, timeout=5.0) as client:

        def callback(sender: int, data: bytearray):
            print(address, data)

        try:
            await client.start_notify(RESPONSE_CHARACTERISTIC, callback)

            print(f"drive medium speed forward")
            cmd = bytearray([0xB8, 0x03, 0x01, 0x02, 0x01])
            await client.write_gatt_char(COMMAND_CHARACTERISTIC, cmd, True)

            for i in range(10):
                print(f"setting color...")
                color = [random.randint(0, 1) * 0xFF for i in range(3)]
                cmd = bytearray([0xB4, 0x07, 0x06, *color, *color])
                await client.write_gatt_char(COMMAND_CHARACTERISTIC, cmd, True)
                await asyncio.sleep(1.0)

            await client.stop_notify(RESPONSE_CHARACTERISTIC)

            print(f"stop movement")
            cmd = bytearray([0xB9, 0x01, 0x01])
            await client.write_gatt_char(COMMAND_CHARACTERISTIC, cmd, True)

            print(f"disconnecting from {address}")

        except Exception as e:
            print(e)


async def main():
    addresses = [
        "33AA542E-1452-4525-8DB8-7BC941A7FE49",
        "CD3C3305-36E2-4DE7-94F3-EC7FFF59E1F8",
    ]
    await asyncio.gather(*(train_program(address) for address in addresses))


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

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

作品・遊びの一覧へ