実験の制御と可視化

実験中には、測定値をグラフによって可視化したいことがあります。 これを実現するために、ebilabを用いることができます。

簡易的な実装

簡易的には、 matplotlib を用いて、以下のようなコードで実現することができます。 これは、マルチメーターを用いて抵抗を測定してプロットするプログラムです。

from datetime import datetime
import pandas as pd
import matplotlib.pyplot as plt
from ebilab.visa import K34411A

fig, ax = plt.subplots(1, 1)
plt.pause(0.01)

multimeter = K34411A()
data = []
started_at = datetime.now()
while True:
    R = multimeter.measure_resistance()
    t = (datetime.now() - started_at).total_seconds()
    print(R)

    # update plot
    data.append({"t": t, "R": R})
    df = pd.DataFrame(data)
    ax.cla()
    ax.plot(df["t"], df["R"])
    ax.set_xlabel("Time")
    ax.set_ylabel("Resistance")
    ax.grid()
    plt.pause(0.1)

実際にこのようなコードで運用するには、いくつか問題があります。

• 実験のロジックを定義するコードと可視化のためのコードが混在している

  • 外れ値をフィルタリングしてプロットしたい場合など、どこまでが実験のロジックでどこからが可視化のためのロジックなのかが分かりづらくなる。

• matplotlibの描画によってデータの取得がブロッキングされ、データの取得速度に影響する。

• プログラムファイルへのデータ保存やグラフでの可視化のコードに関して、プログラムを作成する際に同じようなコードを何度も書く必要がある。

ebilabを利用した実装

ebilabを用いることで、 これらの問題を解決することができます。

上記のコードは、以下のように書き変えることができます。

# sample of GUI app
import asyncio
import random

from ebilab.api import BaseExperiment, BasePlotter, FloatField, SelectField
from ebilab.gui.controller import launch_gui


# class to decide steps of experiment
class RandomWalkExperiment(BaseExperiment):
    """
    Random Walk

    This is example experiment.
    The interval is 0.2 sec.
    """

    columns = ["v", "v2"]  # please specify columns to write csv file
    name = "random-walk"  # filename is suffixed by datetime

    initial = FloatField(default=2.0)
    step = SelectField(choices=[1, 2, 4], default_index=1)

    async def setup(self):
        self.v = self.initial
        self.logger.info(f"Random walk experiment started with initial value {self.v}.")

    async def steps(self):
        self.logger.info("Starting random walk experiment.")
        while True:
            # you can use self.logger to log messages
            self.logger.debug(f"log: {self.v}")

            # random walk logic
            step_value = self.step if random.random() < 0.5 else -self.step
            self.v += step_value

            # The value you yield will be sent to GUI and saved to CSV file
            yield {"v": self.v, "v2": self.v * 2}

            # use asyncio.sleep instead of time.sleep
            await asyncio.sleep(0.2)

    async def cleanup(self):
        self.logger.info("Random walk experiment finished.")


#  class to decide how to plot during experiment
@RandomWalkExperiment.register_plotter
class TransientPlotter(BasePlotter):
    name = "transient"
    window_length = 10

    def setup(self):
        # this method is executed before starting experiment
        # e.g. adding Axes to Figure
        # figure is stored in `self.fig`
        if self.fig:
            self._ax = self.fig.add_subplot(111)

    def update(self, df):
        # this method is executed many times during experiment
        # df is pandas.DataFrame which has experiment data
        if hasattr(self, "_ax") and not df.empty:
            self._ax.clear()

            self._ax.plot(df["t"], df["v"])
            self._ax.set_xlabel("Time")
            self._ax.set_ylabel("Voltage")
            self._ax.grid(True)
            self._ax.text(0.05, 0.95, f"step={self.experiment.step}", transform=self._ax.transAxes, verticalalignment='top')


#  class to decide how to plot during experiment
@RandomWalkExperiment.register_plotter
class HistgramPlotter(BasePlotter):
    name = "histgram"
    bins = FloatField(default=10.0)

    def setup(self):
        # this method is executed before starting experiment
        # e.g. adding Axes to Figure
        # figure is stored in `self.fig`
        if self.fig:
            self._ax = self.fig.add_subplot(111)

    def update(self, df):
        # this method is executed many times during experiment
        # df is pandas.DataFrame which has experiment data
        if hasattr(self, "_ax") and not df.empty and "v" in df.columns:
            self._ax.clear()

            self._ax.hist(df["v"], bins=int(self.bins))
            self._ax.set_xlabel("Value")
            self._ax.set_ylabel("Count")
            
            # 実験インスタンスにアクセスしてパラメータを表示
            if self.experiment:
                title = f"Histogram (step={self.experiment.step}, initial={self.experiment.initial})"
                self._ax.set_title(title)
                # ログに出力してテスト
                self.experiment.logger.info(f"Plotter accessed experiment parameters: step={self.experiment.step}")
            
            self._ax.grid(True)


if __name__ == "__main__":
    # This is a sample code to run the experiment
    # You can run this file directly to see the experiment in action
    launch_gui([RandomWalkExperiment])

BaseExperiment クラス、 BasePlotter クラスを継承し、 実験のロジックと可視化のロジックをそれぞれ実装することで、実験を設計することができます。 これにより、ロジックを適切に分割し、読みやすいコードを実現することができます。 また、クラス単位で定義することにより、同じ実験で可視化の方法だけを変更したり、別の実験でも同一の可視化方法を用いるなど、 それぞれのコンポーネントを再利用しやすくなります。

ファイルへの保存やプロット用プログラムへのデータの受け渡しに関して考慮する必要はありません。 また、matplotlibの描画中にもデータの取得を継続するため、マルチスレッド処理を行なっていますが、その制御に関しては考慮する必要はありません。

実際の実装

ユーザーは以下のクラスを実装する必要があります。

Experiment クラス

実験のロジックは、BaseExperiment を継承して定義します。

class RandomWalkExperiment(BaseExperiment):
    """
    Random Walk

    This is example experiment.
    The interval is 0.2 sec.
    """

    columns = ["v", "v2"]  # please specify columns to write csv file
    name = "random-walk"  # filename is suffixed by datetime

    initial = FloatField(default=2.0)
    step = SelectField(choices=[1, 2, 4], default_index=1)

    async def setup(self):
        self.v = self.initial
        self.logger.info(f"Random walk experiment started with initial value {self.v}.")

    async def steps(self):
        self.logger.info("Starting random walk experiment.")
        while True:
            # you can use self.logger to log messages
            self.logger.debug(f"log: {self.v}")

            # random walk logic
            step_value = self.step if random.random() < 0.5 else -self.step
            self.v += step_value

            # The value you yield will be sent to GUI and saved to CSV file
            yield {"v": self.v, "v2": self.v * 2}

            # use asyncio.sleep instead of time.sleep
            await asyncio.sleep(0.2)

    async def cleanup(self):
        self.logger.info("Random walk experiment finished.")

基本的な構造

• name : 実験の名前。csvファイル名のベースになります。

• columns : 記録用のcsvファイルの列名を指定します。

• パラメータ定義: FloatField, SelectField, IntField, StrField, BoolField などのフィールドクラスを用いて、GUIから設定できるパラメータを定義します。

ライフサイクルメソッド

実験は以下の3つの async メソッドで構成されます。

• async def setup() : 実験の準備。デバイスへの接続などを行います。

• async def steps() : データを yield するメインループ。

• async def cleanup() : 正常終了・中断・エラー時に必ず呼ばれる後処理。

steps() メソッドでは、 yield を使用して測定データを記録します。 yield で返された辞書は、CSVファイルに保存され、GUIで可視化されます。

注釈

time.sleep() の代わりに await asyncio.sleep() を使用してください。 これにより、スリープ中でも実験を中断できます。

ロギング

実験中のログは self.logger を通じて出力できます。 出力したログは、GUIのログビューアに表示される他、ログファイルにも保存されます（デバッグモードでは保存されない）。

self.logger.info("情報メッセージ")
self.logger.debug("デバッグメッセージ")
self.logger.warning("警告メッセージ")
self.logger.error("エラーメッセージ")

Plotter クラス

可視化のロジックは、BasePlotter を継承して定義します。 登録方法は2通りあり、併用できます。

方法1: デコレータでクラスを登録

@Experiment.register_plotter デコレータで実験クラスに関連付けます。

#  class to decide how to plot during experiment
@RandomWalkExperiment.register_plotter
class TransientPlotter(BasePlotter):
    name = "transient"
    window_length = 10

    def setup(self):
        # this method is executed before starting experiment
        # e.g. adding Axes to Figure
        # figure is stored in `self.fig`
        if self.fig:
            self._ax = self.fig.add_subplot(111)

    def update(self, df):
        # this method is executed many times during experiment
        # df is pandas.DataFrame which has experiment data
        if hasattr(self, "_ax") and not df.empty:
            self._ax.clear()

            self._ax.plot(df["t"], df["v"])
            self._ax.set_xlabel("Time")
            self._ax.set_ylabel("Voltage")
            self._ax.grid(True)

• name : プロッターの識別名。GUIのドロップダウンに表示されます。

• setup() : プロットの初期設定。プロッターがアクティブになった際に一度だけ呼ばれます。

• update(df) : データが更新されるたびに呼ばれます。 df は pandas.DataFrame で、全ての実験データが含まれます。

方法2: インスタンスのリストとして宣言

軸名を変えただけのシンプルな可視化なら、 SimpleXYPlotter のようなビルトイン Plotter を plotters クラス変数にインスタンスとして並べるだけで登録できます。

from ebilab.api import BaseExperiment, SimpleXYPlotter

class MyExperiment(BaseExperiment):
    plotters = [
        SimpleXYPlotter("t", "v"),
        SimpleXYPlotter("t", ["v", "v2"], name="v and v2"),
    ]

y_column には単独のカラム名 (例: "v") も、複数カラム名のリスト (例: ["v", "v2"]) も渡せます。 複数指定すると同じ軸に重ねて描画され、凡例が表示されます。

各インスタンスは「テンプレート」として扱われ、実験開始のたびに copy.copy() で複製されてから使われます。

Plotterのパラメータ

Plotter にもフィールドを定義できます。これらはGUIから実験中でも変更できます。

@RandomWalkExperiment.register_plotter
class HistgramPlotter(BasePlotter):
    name = "histgram"
    bins = FloatField(default=10.0)  # GUIから変更可能なパラメータ

    def update(self, df):
        self._ax.hist(df["v"], bins=int(self.bins))

Experimentへのアクセス

self.experiment を通じて、実験インスタンスのパラメータにアクセスできます。

def update(self, df):
    # 実験パラメータを参照
    title = f"step={self.experiment.step}"
    self._ax.set_title(title)

また、 self.experiment.is_running で実験が実行中かどうかを確認できます。 これを使って、実行中と履歴表示でプロットの挙動を変えることができます。

実験の実行

定義したクラスを用いて、以下のように実験を実行できます。

if __name__ == "__main__":
    # This is a sample code to run the experiment
    # You can run this file directly to see the experiment in action
    launch_gui([RandomWalkExperiment])

注釈

複数の実験クラスを指定することができます。GUIのドロップダウンから選択できます。

フィールドの種類

パラメータ定義に使用できるフィールドクラスは以下の通りです。

フィールド	用途	パラメータ
FloatField	浮動小数点数	default, min, max
IntField	整数	default, min, max
StrField	文字列	default, allow_blank
BoolField	真偽値	default
SelectField	選択肢	choices, default_index

使用例:

class MyExperiment(BaseExperiment):
    # 浮動小数点数パラメータ（範囲指定あり）
    voltage = FloatField(default=1.0, min=0.0, max=10.0)

    # 整数パラメータ
    count = IntField(default=100, min=1)

    # 文字列パラメータ
    sample_name = StrField(default="sample1")

    # 真偽値パラメータ
    auto_save = BoolField(default=True)

    # 選択肢パラメータ
    mode = SelectField(choices=["fast", "normal", "slow"], default_index=1)

データの保存

• スクリプト実行時のカレントディレクトリ配下に data ディレクトリが作成され、csvファイルが保存されます。

• ファイル名は name プロパティで指定した名前に日時が付加されます。

• 自動的に t 列（実験開始からの経過秒数）と sync_t 列（syncボタンからの経過秒数）が追加されます。

ショートカットキー

GUIでは以下のショートカットキーが使用できます。

• F5 : 実験開始

• F6 : デバッグ実行（データを保存しない）

• F9 : 実験終了

• F12 : sync（タイムスタンプを記録）