【中級者向け】複数デバイスでのUDP計測 | 脳波デバイス・ニューロフィードバックのGoodBrain

こんにちは、GoodBrain Web管理の吉川です！

このチュートリアルでは、GoodBrain Focus アプリの有料機能「UDP計測」と、Pyhonライブラリの「python-osc」を使用して、複数のデバイスから同時にかつリアルタイムにデータを取得する方法を説明します。

最終的には、取得したデータをターミナル上にプログレスバーで可視化します。

UDP計測は有料機能となっております。興味のある方は、お気軽にお問い合わせください。

UDP計測の詳細

お問い合わせ

また、python-oscを用いたUDP計測データのリアルタイム取得については別の記事で詳しく解説していますので、併せてご覧ください。

【中級者向け】Python と UDP計測（GB Focus) で脳波計測をリアルタイム描画する方法

はじめに

本記事の目的

この記事では、複数の脳波デバイスから、GoodBrain Focusアプリ経由でOSCメッセージを受信し、それぞれのデータをプログレスバーで可視化します。

具体的には：

Attention（集中度）
Meditation（瞑想度）

これらの値を受信して、リアルタイムで表示します。プログレスバーは、rich というライブラリをし、ターミナル（コマンドプロンプト）に表示させます。

筆者の開発環境

FocusCalm™
GoodBrain Focus アプリ (iOS版)
MacBook Air M1 2020 (Tahoe 26.2)
Visual Studio Code
Python 3.12.3

事前準備

必要なパッケージのインストール

まず、必要なPythonパッケージをインストールします。

pip install rich pyhton-osc

python-osc: OSC通信を扱うためのライブラリ
rich: ターミナルにプログレスバーを表示するためのライブラリ

IP アドレスの確認

スマートフォンのGB FocusアプリとPythonプログラムでUDP通信をするのに、IPアドレスが必要です。Macでは、ターミナルで下記コマンドを入力すると取得することができます。

ipconfig getifaddr en0

VS Code のターミナルで実行すると、私は 192.168.0.55 でした。

基本的なOSCサーバー

最初のステップでは、シンプルなOSCサーバーを作成します。

まずは、好きな場所にフォルダを作成し、それをVS Code で開きます。そして、udp-multi.pyというファイルを作成します。

下記は、gb-udp-multiというフォルダにudp-multi.pyを作成した様子です。

次に、作成したファイルに下記コードをコピペしてください。

また、IPアドレスの部分をご自分のIPアドレスに差し替えてください。

from pythonosc.dispatcher import Dispatcher
from pythonosc.osc_server import ThreadingOSCUDPServer

# 設定 =========================

IP = "192.168.0.55" # ! 自分のIPアドレスに変更してください
PORT = 8000

# OSCアドレス
addresses = [
    "/Attention",
    "/Meditation"
]

# ハンドラー関数 =========================

def handler(address, *args):
    print(f"{address}: {args}")

# メインの処理 =========================

# アドレスと関数を紐付け 
disp = Dispatcher()
for address in addresses:
    disp.map(address, handler)

# サーバーを起動
print(f"OSCサーバ (ポート: {PORT}) ... Ctrl-Cで終了")
server = ThreadingOSCUDPServer((IP, PORT), disp) # サーバーの作成
server.serve_forever() # サーバーを起動（Ctrl+Cで終了）

処理の説明

設定部分：

IP: サーバーでリッスンするIPアドレス（先ほど取得したもの）
PORT: サーバーがリッスンするポート番号
addresses: 受信するOSCアドレスのリスト

ハンドラー関数：OSCメッセージを受信したときに呼び出される関数。アドレスと受信データを print する）

メインの処理：

Dispatcherを作成：OSCアドレスとハンドラー関数を紐付けるオブジェクト
各アドレスに対してハンドラー関数を登録
ThreadingOSCUDPServerを作成：スレッドベースのUDPサーバー
serve_forever()でサーバーを起動（Ctrl+Cで終了）

詳細はこちらのコラムで解説しています：【中級者向け】Python と UDP計測（GB Focus) で脳波計測をリアルタイム描画する方法

動作確認

早速、リアルタイムのUDP計測をしてみましょう。

まずは、FocusCalmの電源ボタン（下記画像）を5秒程度長押しし、接続モードにしたら（成功すると２回振動がきます）、頭に装着します。

次に、GB Focusアプリの設定をしていきます。

スマートフォンで GB Focus アプリを開き、UDP計測を選択
デバイス選択で FocusCalm™ を選択
「IP/ポートを追加」から、下記のような２つのポートを追加します。
- ポート１
  - ホスト：ご自分のIPアドレス（先ほど取得したもの）
  - 送信ポート：8000（Pythonプログラムに書いたものと一致させる）
  - データ共有設定：Attention（集中度）
- ポート２
  - ホスト：ご自分のIPアドレス
  - 送信ポート：8000
  - データ共有設定：Meditation（リラックス度）

ポートを作成できたら、「接続」を押して、デバイスを接続してください。

デバイスが見つからない場合は、FocusCalm™ の電源が切れている可能性があるので、再度接続モード（5秒長押し）にしてみてください。
「接続中」で画面が止まってしまった場合は、一度アプリを落として再接続してみてください（作成したポートの情報は保存されます）。

接続が成功すると、下記画像のような「中央に一時停止ボタンがある画面」に遷移します。

接続がうまくいったら、Pythonのプログラムを実行します。

python udp-multi.py

うまくいっていれば、ターミナルに受信された情報が出力されます。

これで、デバイス１台のデータをPythonで取得することができました。次章からは、デバイス複数台のデータを取得する方法を解説していきます。

非同期OSCサーバー

概要

前章では、１つのポートでOSCメッセージを受信できるサーバーを作成しました。

次のステップでは、複数のポートで同時にOSCメッセージを受信できる非同期サーバーを作成します。これにより、複数デバイスからのデータを同時に処理できます。

同期サーバーと非同期サーバーの違い

同期サーバー（ThreadingOSCUDPServer）: 1つのポートのみ
非同期サーバー（AsyncIOOSCUDPServer）: 複数のポートを同時に処理可能

コードの解説

まずは、変更したコードを示します。コピペする場合は、IPアドレスの変更を忘れないでください！

import asyncio

from pythonosc.dispatcher import Dispatcher
from pythonosc.osc_server import AsyncIOOSCUDPServer

# 設定 =========================

IP = "192.168.0.55"  # ! 自分のIPアドレスに変更してください
PORTS = [8000, 8001]

addresses = ["/Attention", "/Meditation"]

# メインの処理 =========================

def create_osc_server(ip, port, loop):
    # ハンドラー関数
    def handler(address, *args):
        print(f"{port}{address}: {args}")

    # ディスパッチャーの作成
    disp = Dispatcher()
    for address in addresses:
        disp.map(address, handler)

    # 非同期OSCサーバーの作成
    server = AsyncIOOSCUDPServer((ip, port), disp, loop)
    print(f"ポート {port} でOSCサーバーを作成しました。")
    return server


async def main():
    # イベントループの取得
    loop = asyncio.get_running_loop()
    assert isinstance(loop, asyncio.BaseEventLoop)

    servers = [create_osc_server(IP, port, loop) for port in PORTS]

    # サーバーの起動
    endpoints = [await server.create_serve_endpoint() for server in servers]

    try:
        # 待機
        await asyncio.Event().wait()
    finally:
        # サーバーの停止
        for endpoint in endpoints:
            transport, _ = endpoint
            transport.close()


if __name__ == "__main__":
    # メイン関数の実行
    asyncio.run(main())

変更点

設定：

asyncioを新たにインポート
ThreadingOSCUDPServerの代わりにAsyncIOOSCUDPServerをインポート
PORT（単数）からPORTS（複数）に変更し、複数のポート番号をリストで定義

サーバー作成関数（create_osc_server）：

各ポートに対してディスパッチャーを作成し、OSCサーバーを作成
AsyncIOOSCUDPServerを使用（非同期版）
loopパラメータで非同期イベントループを指定

非同期メイン関数：

asyncio.get_running_loop()で現在の非同期イベントループを取得
リスト内包表記で各ポートにサーバーを作成
create_serve_endpoint()で各サーバーのエンドポイントを作成（実際に通信開始）
asyncio.Event().wait()で無限に待機（Ctrl+Cまで）
終了時にすべてのトランスポートをクローズ

プログラムの起動：

asyncio.run()で非同期メイン関数を実行

動作確認

それでは、変更したプログラムで、複数ポートのデータを取得できるか試してみましょう。

FocusCalm™ 複数台で試したいところですが、筆者の手元には１台しかないため、GB Focus アプリでポートを追加して試したいと思います。（FocusCalm™ を複数台お持ちの方は、ぜひ複数台でお試しください！）

送信ポートを8001に設定した「ポート３」と「ポート４」を追加しましょう。

Pythonのプログラムを実行します。

python udp-multi.py

うまくいっていれば、8000と8001の両方でデータが受信されているのを確認できます。

ここまでで、すでに複数台の脳波デバイスでのUDP計測を実現できました！次章では、rich というライブラリを使って計測データを可視化する方法を解説します。

プログレスバーの実装

概要

最終ステップでは、受信したOSCデータをプログレスバーで可視化します。rich ライブラリを使用して、美しく見やすいプログレスバーを実装します。

新しい機能

各ポート・各アドレスごとに独立したプログレスバーを表示
リアルタイムでプログレスバーを更新
値の範囲は0〜100

コードの解説

変更したコードを下記に記載します。コピペする場合は、IPアドレスの変更を忘れないでください！

import asyncio

from pythonosc.dispatcher import Dispatcher
from pythonosc.osc_server import AsyncIOOSCUDPServer
from rich.progress import BarColumn, Progress, TaskID, TaskProgressColumn, TextColumn

# 設定 =========================

IP = "192.168.0.55"
PORTS = [8000, 8001]

# バーの最大値
PROGRESS_MAX = 100.0

# メインの処理 =========================

async def main():
    loop = asyncio.get_running_loop()

    # プログレスバーを作成
    progress = Progress(
        TextColumn("[bold]{task.description}"),
        BarColumn(),
        TaskProgressColumn(),
        TextColumn("{task.completed:>6.1f} / {task.total:.0f}"),
        refresh_per_second=20,
        transient=False,
    )

    # ハンドラー関数を作成する関数
    def make_handler(task_id: TaskID):
        # ハンドラー関数
        def _handler(address: str, *args):
            if not args:
                return

            progress.update(task_id, completed=args[0])

        # 作成した関数を返す
        return _handler

    def create_osc_server(ip, port, loop):
        # ディスパッチャーを作成
        disp = Dispatcher()

        # タスクを作成（attention, meditation)
        attention_task = progress.add_task(f"{port}: Attention", total=PROGRESS_MAX)
        meditation_task = progress.add_task(f"{port} Meditation", total=PROGRESS_MAX)

        # ディスパッチャに関数を登録
        disp.map("/Attention", make_handler(attention_task))
        disp.map("/Meditation", make_handler(meditation_task))

        # サーバーを起動
        server = AsyncIOOSCUDPServer((IP, port), disp, loop)
        # servers.append(server)

        return server

    # プログレスバーを描画
    with progress:
        # OSCサーバーを入れるリスト
        servers = []

        # ポートごとにサーバーを作成して起動
        for port in PORTS:
            server = create_osc_server(IP, port, loop)
            servers.append(server)

        print(f"Listening on {IP} ports: {PORTS}")

        # 後でUDP通信を閉じるためのオブジェクト
        endpoints = [await s.create_serve_endpoint() for s in servers]

        try:
            # Ctrl+C まで待機
            await asyncio.Event().wait()
        finally:
            for transport, _ in endpoints:
                transport.close()


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

主な変更点

インポート：

rich.progressから必要なコンポーネントをインポート
- Progress: プログレスバーのメインクラス
- BarColumn: プログレスバーの表示
- TextColumn: テキスト表示
- TaskProgressColumn: 進捗率の表示
- TaskID: タスクの識別子

ハンドラー関数、サーバー作成関数：

progressを使用するため、関数（make_handler と create_osc_server）は、メイン関数の中に配置
ハンドラー関数を作成する関数 make_handlerを新たに作成
create_osc_server内でmake_handlerを実行し、プログレスバーを更新するハンドラーを作成

動作確認

早速、プログラムを実行してみましょう。

python udp-multi.py

うまくいくと、プログレスバーが増減する様子が見ることができます。

これで、プログレスバーで集中度・リラックス度を可視化できました！

まとめ

このチュートリアルでは、以下の内容を学習しました：

基本的なOSCサーバー
- ThreadingOSCUDPServerを使った単一ポートのサーバー
- ディスパッチャーとハンドラーの関係
非同期OSCサーバー
- AsyncIOOSCUDPServerを使った複数ポートのサーバー
- asyncioによる非同期処理
- プログレスバーの実装
richライブラリによる集中・リラックス度の可視化

本記事で説明した内容を発展させれば、実験やゲーム制作にも応用することができます。ギャラリーにUDP計測を使用した事例がありますので、ぜひご覧ください。

Gallery

目次

はじめに

本記事の目的

筆者の開発環境

事前準備

必要なパッケージのインストール

IP アドレスの確認

基本的なOSCサーバー

処理の説明

動作確認

非同期OSCサーバー

概要

同期サーバーと非同期サーバーの違い

コードの解説

変更点

動作確認

プログレスバーの実装

概要

新しい機能

コードの解説

主な変更点

動作確認

まとめ