実験の制御と可視化

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

簡易的な実装

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

 1from datetime import datetime
 2import pandas as pd
 3import matplotlib.pyplot as plt
 4from ebilab.visa import K34411A
 5
 6fig, ax = plt.subplots(1, 1)
 7plt.pause(0.01)
 8
 9multimeter = K34411A()
10data = []
11started_at = datetime.now()
12while True:
13    R = multimeter.measure_resistance()
14    t = (datetime.now() - started_at).total_seconds()
15    print(R)
16
17    # update plot
18    data.append({"t": t, "R": R})
19    df = pd.DataFrame(data)
20    ax.cla()
21    ax.plot(df["t"], df["R"])
22    ax.set_xlabel("Time")
23    ax.set_ylabel("Resistance")
24    ax.grid()
25    plt.pause(0.1)

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

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

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

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

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

ebilabを利用した実装

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

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

  1# sample of GUI app
  2import asyncio
  3import random
  4
  5from ebilab.api import BaseExperiment, BasePlotter, FloatField, SelectField
  6from ebilab.gui.controller import launch_gui
  7
  8
  9# class to decide steps of experiment
 10class RandomWalkExperiment(BaseExperiment):
 11    """
 12    Random Walk
 13
 14    This is example experiment.
 15    The interval is 0.2 sec.
 16    """
 17
 18    columns = ["v", "v2"]  # please specify columns to write csv file
 19    name = "random-walk"  # filename is suffixed by datetime
 20
 21    initial = FloatField(default=2.0)
 22    step = SelectField(choices=[1, 2, 4], default_index=1)
 23
 24    async def setup(self):
 25        self.v = self.initial
 26        self.logger.info(f"Random walk experiment started with initial value {self.v}.")
 27
 28    async def steps(self):
 29        self.logger.info("Starting random walk experiment.")
 30        while True:
 31            # you can use self.logger to log messages
 32            self.logger.debug(f"log: {self.v}")
 33
 34            # random walk logic
 35            step_value = self.step if random.random() < 0.5 else -self.step
 36            self.v += step_value
 37
 38            # The value you yield will be sent to GUI and saved to CSV file
 39            yield {"v": self.v, "v2": self.v * 2}
 40
 41            # use asyncio.sleep instead of time.sleep
 42            await asyncio.sleep(0.2)
 43
 44    async def cleanup(self):
 45        self.logger.info("Random walk experiment finished.")
 46
 47
 48#  class to decide how to plot during experiment
 49@RandomWalkExperiment.register_plotter
 50class TransientPlotter(BasePlotter):
 51    name = "transient"
 52
 53    def setup(self):
 54        # this method is executed before starting experiment
 55        # e.g. adding Axes to Figure
 56        # figure is stored in `self.fig`
 57        if self.fig:
 58            self._ax = self.fig.add_subplot(111)
 59
 60    def update(self, df):
 61        # this method is executed many times during experiment
 62        # df is pandas.DataFrame which has experiment data
 63        if hasattr(self, "_ax") and not df.empty:
 64            self._ax.clear()
 65
 66            self._ax.plot(df["t"], df["v"])
 67            self._ax.set_xlabel("Time")
 68            self._ax.set_ylabel("Voltage")
 69            self._ax.grid(True)
 70            self._ax.text(0.05, 0.95, f"step={self.experiment.step}", transform=self._ax.transAxes, verticalalignment='top')
 71
 72
 73#  class to decide how to plot during experiment
 74@RandomWalkExperiment.register_plotter
 75class HistgramPlotter(BasePlotter):
 76    name = "histgram"
 77    bins = FloatField(default=10.0)
 78
 79    def setup(self):
 80        # this method is executed before starting experiment
 81        # e.g. adding Axes to Figure
 82        # figure is stored in `self.fig`
 83        if self.fig:
 84            self._ax = self.fig.add_subplot(111)
 85
 86    def update(self, df):
 87        # this method is executed many times during experiment
 88        # df is pandas.DataFrame which has experiment data
 89        if hasattr(self, "_ax") and not df.empty and "v" in df.columns:
 90            self._ax.clear()
 91
 92            self._ax.hist(df["v"], bins=int(self.bins))
 93            self._ax.set_xlabel("Value")
 94            self._ax.set_ylabel("Count")
 95            
 96            # 実験インスタンスにアクセスしてパラメータを表示
 97            if self.experiment:
 98                title = f"Histogram (step={self.experiment.step}, initial={self.experiment.initial})"
 99                self._ax.set_title(title)
100                # ログに出力してテスト
101                self.experiment.logger.info(f"Plotter accessed experiment parameters: step={self.experiment.step}")
102            
103            self._ax.grid(True)
104
105
106if __name__ == "__main__":
107    # This is a sample code to run the experiment
108    # You can run this file directly to see the experiment in action
109    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 を継承して定義します。 @Experiment.register_plotter デコレータで実験クラスに関連付けます。

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

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

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

注釈

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