TWELITE Wings API / MWings for Python

ライブラリの概要

TWELITE Wings API (以降 MWings) は、Python スクリプトから TWELITE を扱うためのライブラリです。

• GitHub リポジトリ
• GitHub リポジトリ（Lite版）

機能

ホストへ接続された TWELITE の親機を通じて、TWELITE の子機と通信できます。

• 受信データを解釈して、辞書や JSON のほか pandas データフレームへ変換※
• 辞書から生成したコマンドを親機へ送信

※ Lite版は pandas に非対応

用途例

例えば、次のようなシステムを実現できます。

• MONOSTICK で受信した温湿度データを JSON としてクラウドサーバへ送信
• MONOSTICK で受信した加速度データを CSV または Excel ファイルへ記録※
• PC から MONOSTICK を通じて TWELITE DIP へ接続された LED を制御

※ Lite版は、直接 CSV や Excel ファイルを出力できません

特徴

モダン Python の モダン Python による モダン Python のためのモジュールです。

• pip や poetry を使って導入できる
• 型ヒント へ対応している
• pydantic により、送受信データのバリデーションを実施している
• 基本的に PEP8 へ準拠している※

※ 例外あり。詳細は後述

インストール

PyPI から入手できます。

pip の場合

pip install mwings

poetry の場合

poetry add mwings

pip install mwingslite

poetry add mwingslite

最も簡単なサンプルスクリプト

わずか6行で超簡単！標準アプリ（App_Twelite）の受信データを JSON 形式で出力できます。

import mwings as mw
twelite = mw.Twelite(mw.utils.ask_user_for_port())
@twelite.on(mw.common.PacketType.APP_TWELITE)
def on_app_twelite(packet):
    print(packet.to_json())
twelite.start()

環境整備と動作確認

用意するもの

• PC
• MONOSTICK（親機・中継機アプリ／デフォルト設定）
• TWELITE DIP（超簡単！標準アプリ／デフォルト設定）
  • スイッチなどのペリフェラルを接続しておく（例：DI1 ポートと GND 間にタクトスイッチを接続）

環境整備

pyenv の導入

処理系のバージョンを管理するために、pyenv を導入します。

Linux

curl https://pyenv.run | bash

sudo apt update; sudo apt install build-essential libssl-dev zlib1g-dev libbz2-dev libreadline-dev libsqlite3-dev curl git libncursesw5-dev xz-utils tk-dev libxml2-dev libxmlsec1-dev libffi-dev liblzma-dev

yum install gcc make patch zlib-devel bzip2 bzip2-devel readline-devel sqlite sqlite-devel openssl-devel tk-devel libffi-devel xz-devel

macOS

brew update
brew install pyenv

/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"

brew install openssl readline sqlite3 xz zlib tcl-tk

Windows

Windows 向けの pyenv はありません。代わりに pyenv-win を使用します。

Invoke-WebRequest -UseBasicParsing -Uri "https://raw.githubusercontent.com/pyenv-win/pyenv-win/master/pyenv-win/install-pyenv-win.ps1" -OutFile "./install-pyenv-win.ps1"; &"./install-pyenv-win.ps1"

pyenv / pyenv-win による Python の導入

MWings が対応する Python 3.12 以降（Lite版は3.11以降）の処理系を導入します。

導入できるバージョンの一覧を取得するには、下記を実行します。

pyenv install -l

ここでは例として、Python 3.12.4 を導入し、システム全体へ適用します。

pyenv install 3.12.4
pyenv global 3.12.4

導入済みバージョンの一覧を取得するには、下記を実行してください。

pyenv versions

pipx の導入

poetry のようなコマンドラインツールを隔離された環境で管理するために、pipx を導入します。

Linux

Debian系

sudo apt update
sudo apt install pipx
pipx ensurepath

pipから（Raspberry Piはこちらを推奨）

python3 -m pip install --user pipx
python3 -m pipx ensurepath

Fedora系

sudo dnf install pipx
pipx ensurepath

参考：Installation - pipx

macOS

brew install pipx
pipx ensurepath

Windows

scoop install pipx
pipx ensurepath

Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser
Invoke-RestMethod -Uri https://get.scoop.sh | Invoke-Expression

poetry の導入

プロジェクトに対する処理系のバージョンやモジュールの依存関係を Node.js のような形で管理するために、poetry を導入します。

pipx install poetry

python3 -m pip install --user pipx
python3 -m pipx ensurepath

プロジェクトの作成

ここでは、プロジェクト名を mwtest とします。

プロジェクトを作成するディレクトリへ移動し、下記を実行してください。

poetry new mwtest

mwtest ディレクトリを生成します。

プロジェクトの設定

プロジェクトへ移動し、先ほど pyenv で導入した 処理系のバージョンを紐付けます。

poetry env use 3.12.4

MWings を導入します。

poetry add mwings

export PYTHON_KEYRING_BACKEND=keyring.backends.null.Keyring

最も簡単なサンプルスクリプトの作成

まずは、先ほど紹介したスクリプト を動かしてみましょう。

import mwings as mw
twelite = mw.Twelite(mw.utils.ask_user_for_port())
@twelite.on(mw.common.PacketType.APP_TWELITE)
def on_app_twelite(packet):
    print(packet.to_json())
twelite.start()

import mwingslite as mw
twelite = mw.Twelite("/dev/ttyS0")
@twelite.on(mw.common.PacketType.APP_TWELITE)
def on_app_twelite(packet):
    print(packet.to_json())
twelite.start()

上記の内容で poetry が生成した __init__.py と同じ階層に simple.py を作成します。

📁 mwtest
└ 📁 mwtest
  ├ 📄 __init__.py
  └ 📄 simple.py

最も簡単なサンプルスクリプトの実行

MONOSTICK を接続し、実行します。

poetry run python simple.py

複数のシリアルポートが存在する場合は、シリアルポートを選択してください。

JSON 文字列の出力を得ることができたら成功です。実際の出力例を下記に示します。

{
  "time_parsed": "2024-02-20T03:16:50.150386+00:00",
  "packet_type": "APP_TWELITE",
  "sequence_number": 13699,
  "source_serial_id": "810E0E23",
  "source_logical_id": 120,
  "lqi": 84,
  "supply_voltage": 3249,
  "destination_logical_id": 0,
  "relay_count": 0,
  "periodic": true,
  "di_changed": [
    true,
    true,
    false,
    false
  ],
  "di_state": [
    false,
    false,
    false,
    false
  ],
  "ai_voltage": [
    8,
    272,
    1032,
    112
  ],
  "mwings_implementation": "python",
  "mwings_version": "1.0.0",
  "hostname": "silverstone.local",
  "system_type": "Darwin"
}

実用的なスクリプトの作成

simple.py はあくまでも説明用のサンプルです。実用的なスクリプトではありません。

なぜなら、twelite.start() がデータを受信するためのスレッドを作成するものの、これを終了する手段を用意していないからです。非常に読みづらいスクリプトでもあります。

次はより実用的なスクリプトを作成しましょう。それを実行したのち、内容を解説します。

今回は、下記の3点を条件とします。

• Ctrl+C により、スレッドを正常終了できるようにすること
• PEP8 へ準拠すること（一部例外あり、詳細は後述）
• 型ヒント を導入すること

これらを適用した例を下記に示します。

# -*- coding:utf-8 -*-
# Written for Python 3.12
# Formatted with Black

# MWings example: Receive data, print JSON, typed

from zoneinfo import ZoneInfo

import mwings as mw


# Main function
def main() -> None:
    # Create a twelite object
    twelite = mw.Twelite(mw.utils.ask_user_for_port())

    # Use JST for received data
    twelite.set_timezone(ZoneInfo("Asia/Tokyo"))

    # Register an event handler
    @twelite.on(mw.common.PacketType.APP_TWELITE)
    def on_app_twelite(packet: mw.parsers.app_twelite.ParsedPacket) -> None:
        print(packet.to_json(verbose=True, spread=False))

    # Start receiving
    try:
        # Set as daemon thread
        twelite.daemon = True
        # Start the thread, Join to the main thread
        twelite.start()
        print("Started receiving")
        while True:
            twelite.join(0.5)
    except KeyboardInterrupt:
        # Stop the thread
        print("Flushing...")
        twelite.stop()
        print("Completed")


if __name__ == "__main__":
    # Call the main function
    main()

上記を practical.py として保存してください。

📁 mwtest
└ 📁 mwtest
  ├ 📄 __init__.py
  ├ 📄 simple.py
  └ 📄 practical.py

実用的なスクリプトの実行

下記のコマンドを実行すると、simple.py と同じようにして JSON 形式の出力を得ることができます。

poetry run python practical.py

ただし、今回はエラーを発生させずに Ctrl+C で終了できるほか、time_parsed は日本標準時になっているものと思います。

実用的なスクリプトの解説

コードの解説

practical.py を解説します。

import 文

practical.py では、2つのモジュールを import しています。

from zoneinfo import ZoneInfo

import mwings as mw

• zoneinfo.ZoneInfo 受信時刻のタイムゾーンを指定するために使用します。
• mwings MWings ライブラリです。mw に短縮して呼び出せるようにしています。Lite版の場合は mwingslite です。

オブジェクトの作成

mw.Twelite オブジェクトは、ホストへ接続された TWELITE の親機へアクセスするためのインタフェースとなります。

# Create a twelite object
twelite = mw.Twelite(mw.utils.ask_user_for_port())

mw.utils.ask_user_for_port() 関数は、ホストで利用できるシリアルポートの一覧を取得し、ユーザが選択したポートのファイルディスクリプタのパスや COM ポート名を返します。

# Linux
twelite = mw.Twelite("/dev/ttyUSBx")
# macOS
twelite = mw.Twelite("/dev/cu.usbserial-MWxxxxxx")
# Windows
twelite = mw.Twelite("COMx")

twelite = mw.Twelite(port=None)

タイムゾーンの設定

デフォルトでは、データの受信時刻を UTC として扱います。

practical.py では、これを JST としています。

# Use JST for received data
twelite.set_timezone(ZoneInfo("Asia/Tokyo"))

ZoneInfo には IANA のタイムゾーン識別子を渡してください。

受信ハンドラの登録

TWELITE の子機から送られるデータを処理するには、受信ハンドラを登録します。ここでは、超簡単！標準アプリ向けの受信ハンドラ内で受信したデータを JSON 形式に変換し、それを出力しています。

# Register an event handler
@twelite.on(mw.common.PacketType.APP_TWELITE)
def on_app_twelite(packet: mw.parsers.app_twelite.ParsedPacket) -> None:
    print(packet.to_json(verbose=True, spread=False))

受信ハンドラは、任意の関数へ twelite.on() デコレータ ? を適用することで登録できます。

@twelite.on(mw.common.PacketType.APP_TWELITE)
def foobar(packet: mw.parsers.app_twelite.ParsedPacket):
    # handle packets

@twelite.on(mw.common.PacketType.APP_IO)
def foobar(packet: mw.parsers.app_io.ParsedPacket):
    # handle packets

@twelite.on(mw.common.PacketType.APP_ARIA)
def foobar(packet: mw.parsers.app_aria.ParsedPacket):
    # handle packets

@twelite.on(mw.common.PacketType.APP_CUE)
def foobar(packet: mw.parsers.app_cue.ParsedPacket):
    # handle packets

@twelite.on(mw.common.PacketType.APP_CUE_PAL_EVENT)
def foobar(packet: mw.parsers.app_cue_pal_event.ParsedPacket):
    # handle packets

@twelite.on(mw.common.PacketType.APP_PAL_MOT)
def foobar(packet: mw.parsers.app_pal_mot.ParsedPacket):
    # handle packets

@twelite.on(mw.common.PacketType.APP_PAL_AMB)
def foobar(packet: mw.parsers.app_pal_amb.ParsedPacket):
    # handle packets

@twelite.on(mw.common.PacketType.APP_PAL_OPENCLOSE)
def foobar(packet: mw.parsers.app_pal_openclose.ParsedPacket):
    # handle packets

@twelite.on(mw.common.PacketType.APP_UART_ASCII)
def foobar(packet: mw.parsers.app_uart_ascii.ParsedPacket):
    # handle packets

@twelite.on(mw.common.PacketType.APP_UART_ASCII_EXTENDED)
def foobar(packet: mw.parsers.app_uart_ascii_extended.ParsedPacket):
    # handle packets

@twelite.on(mw.common.PacketType.ACT)
def foobar(packet: mw.parsers.act.ParsedPacket):
    # handle packets

受信ハンドラが受け取るデータクラス ParsedPacket は、JSON 形式の文字列へ変換する to_json()、辞書へ変換する to_dict()、pandas データフレームへ変換する to_df() といったメソッドを備えています。

ここでは、to_json() メソッドを使って JSON 形式の文字列へと変換しています。

print(packet.to_json(verbose=True, spread=False))

受信の開始と終了

mw.Twelite は、threading.Thread を継承しています 。practical.py では、twelite.start() が別のスレッドで受信処理を開始し、twelite.stop() がそれを停止しています。

# Start receiving
try:
    # Set as daemon thread
    twelite.daemon = True
    # Start the thread, Join to the main thread
    twelite.start()
    print("Started receiving")
    while True:
        twelite.join(0.5)
except KeyboardInterrupt:
    # Stop the thread
    print("Flushing...")
    twelite.stop()
    print("Completed")

twelite.daemon を True へ設定すると、受信処理を担うサブスレッドは デーモン化 されます。デーモンでない生存中のスレッドがすべて終了すると、 Python プログラム全体も終了します。ここではメインスレッド内で twelite.join() を繰り返し呼ぶことで、メインスレッドを待機させています。

メインスレッドは Ctrl+C の入力を検知すると except 節で twelite.stop() を呼び出し、受信処理を停止させます。twelite.stop() はサブスレッドが受信ハンドラの呼び出しを終えるまで待機します。

try:
    ...
    twelite.join()
except KeyboardInterrupt:
    ...

補足説明

PEP8 への対応

practical.py や MWings のソースコードは、PEP8 に対応したコードフォーマッタ Black を使ってフォーマットされています。

Black は一行の最大長を除き、いかなる設定も受け付けない頑固なフォーマッタですが、コーディング規約を策定・共有する手間を省くことができます。コーディングルールはしばしば物議を醸す議題ですが、そうした取るに足らない作業へ費やすリソースを有益な作業へ割り当てることこそが Python らしさの本質でしょう。

プロジェクトへ Black を追加するには、下記を実行します。

poetry add --group dev black

対象のファイルやディレクトリを指定して実行できます。

poetry run black mwtest

フォーマットせずに確認だけ行うこともできます。

poetry run black --check mwtest

型ヒントへの対応

practical.py や MWings のソースコードは、型ヒント へ対応しています。

MWings ライブラリは、Python 公式の静的型チェッカである mypy を使用しています。

プロジェクトへ mypy を追加するには、下記を実行します。

poetry add --group dev mypy

こちらも Black と同様に、対象のファイルやディレクトリを指定して実行できます。

poetry run mypy mwtest

関連情報

実用的なスクリプトの応用

practical.py をさらに発展させたスクリプトを公開しています。

mwings_python/examples at main

log_export.py

親機の出力を保存したテキストファイルを読み込み、解釈した結果を pandas データフレームへ保存し、最終的に CSV または Excel ファイルを出力します。Lite版は非対応。

poetry run python log_export.py -h
usage: log_export.py [-h] [-x] [-v] [-s] INPUT_FILE

Parse a log file for App_Wings

positional arguments:
  INPUT_FILE     text file contains logs from App_Wings

options:
  -h, --help     show this help message and exit
  -x, --excel    export an Excel file instead of CSV
  -v, --verbose  include system information
  -s, --sort     sort columns in the output

rx_export.py

接続された親機の出力を受け取り、解釈した結果を pandas データフレームへ保存し、最終的に CSV または Excel ファイルを出力します。Lite版は非対応。

• CSV ファイルを選択した場合、すべての結果を一つのファイルへ保存します。
• 代わりに Excel ファイルを選択した場合、パケットの種別ごとに別のシートへ保存します。

poetry run python rx_export.py -h
usage: rx_export.py [-h] [-x] [-v] [-s]

Log packets from App_Wings to csv or excel

options:
  -h, --help     show this help message and exit
  -x, --excel    export an Excel file instead of CSV
  -v, --verbose  include system information
  -s, --sort     sort columns in the output

rx_export_csv_durable.py

接続された親機の出力を受け取り、解釈した結果を送信元のシリアルIDごとに CSV ファイルへ追記していきます。Lite版は非対応。

poetry run python rx_export_csv_durable.py -h
usage: rx_export_csv_durable.py [-h] [-v] [-s]

Log packets from App_Wings to csv, line by line

options:
  -h, --help     show this help message and exit
  -v, --verbose  include system information
  -s, --sort     sort columns in the output

rx_print_df.py

接続された親機の出力を受け取り、単に解釈した結果を pandas データフレームへ変換し、文字列として出力します。Lite版は非対応。

rx_print_dict.py

接続された親機の出力を受け取り、単に解釈した結果を辞書へ変換して出力します。Lite版も対応。

rx_print_json.py

接続された親機の出力を受け取り、単に解釈した結果を JSON 文字列へ変換して出力します。Lite版も対応。

tx_binary_uart.py

接続されたシリアル通信アプリの親機を介して、TWELITE UART へバイナリデータ [0xBE, 0xEF] を送ります。Lite版も対応。

tx_blink_dip_led.py

接続された親機・中継機アプリの親機を介して、TWELITE DIP の DO1 ポートに接続された LED を点滅させます。Lite版も対応。

tx_blink_pal_notice.py

接続された親機・中継機アプリの親機を介して、通知パルの LED を各色で点灯させます。Lite版も対応。

API リファレンス

• 日本語版
• 英語版（Sphinx）