実験の制御と可視化

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

Simple implementation

You can implement code like below with matplotlib. This program measures resistance by digital multimeter and plot it to a chart.

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)

This code has some problems.

This code includes both a part which describe a logic of the experiment and a part for visualization.
- 外れ値をフィルタリングしてプロットしたい場合など、どこまでが実験のロジックでどこからが可視化のためのロジックなのかが分かりづらくなる。
Data acquision rate is lowered becaues the rendering of matplotlib blocks data acquision.
You have to write similar code many times like saving data or visualizing.

ebilabを利用した実装

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

The preceding code can be refactored as below,

# 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で可視化されます。

Note

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

ロギング

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

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

Plotter クラス

可視化のロジックは、BasePlotter を継承して定義します。 @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 で、全ての実験データが含まれます。

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])

Note

複数の実験クラスを指定することができます。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（タイムスタンプを記録）