実験の設計

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

簡易的な実装

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

 1from datetime import datetime
 2import pandas as pd
 3import matplotlib.pyplot as plt
 4from ebilab.experiment.devices 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(v)
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.experiment モジュールを利用した実装

ebilab.experiment モジュールを用いることで、 これらの問題を解決することができます。

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

 1# sample of multi-thread measurement / plotting using Plotter / Experiment class
 2import matplotlib.pyplot as plt
 3
 4from ebilab.experiment import (
 5    ExperimentProtocol,
 6    ExperimentPlotter,
 7    ExperimentContext,
 8    launch_experiment,
 9)
10from ebilab.experiment.devices import K34411A
11
12
13class ResistancePlotter(ExperimentPlotter):
14    name = "simple"
15
16    def prepare(self):
17        _, self._ax = plt.subplots(1, 1, num=self.fig.number)
18
19    def update(self, df):
20        df = df.query("R < 1e20")
21
22        self._ax.cla()
23        self._ax.plot(df["t"], df["R"])
24        self._ax.set_xlabel("Time / s")
25        self._ax.set_ylabel("Resistance / Ohm")
26        self._ax.grid()
27
28
29class ContinuousResistanceMesurement(ExperimentProtocol):
30    name = "resistance"
31    columns = ["R"]
32    plotter_classes = [ResistancePlotter]
33
34    def steps(self, ctx: ExperimentContext):
35        multimeter = K34411A()
36        while True:
37            r = multimeter.measure_resistance()
38            ctx.send_row({"R": r})
39            ctx.loop()
40
41
42if __name__ == "__main__":
43    launch_experiment([ContinuousResistanceMesurement])

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

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

実際の実装

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

ExperimentPlotter クラス

可視化のロジックは、以下のように定義されます。

    ExperimentContext,
    launch_experiment,
)
from ebilab.experiment.devices import K34411A


class ResistancePlotter(ExperimentPlotter):
    name = "simple"

    def prepare(self):
        _, self._ax = plt.subplots(1, 1, num=self.fig.number)

    def update(self, df):

  • name プロパティは、可視化のロジックの区別に用います。

  • prepareコマンドは初回のみ実行されます。

  • updateコマンドは、定期的に実行されます。

ExperientProtocol クラス

実験のロジックは、以下のように定義されます。


        self._ax.cla()
        self._ax.plot(df["t"], df["R"])
        self._ax.set_xlabel("Time / s")
        self._ax.set_ylabel("Resistance / Ohm")
        self._ax.grid()


class ContinuousResistanceMesurement(ExperimentProtocol):
    name = "resistance"
    columns = ["R"]

  • クラスの columns プロパティを用いて、記録用のcsvファイルの列を指定します。

  • スクリプト実行時のカレントディレクトリ配下に data ディレクトリが作成され、csvファイルが保存されます。 クラスの name プロパティで指定した名前に、自動で日時が記録され、ファイル名となります。

  • クラスの plotter_classes プロパティで、定義したExperimentPlotterの クラス名 を指定してください。

  • steps関数に、実際の実験の処理を定義します。

    • 適切に実験を制御できるようにするため、実験中のループ内で ctx.loop() を毎回呼び出してください。

    • ctx.send_row() メソッドを用いて、測定結果のデータを記録することができます。

      • メソッド実行1回あたり、csvファイル1行になります。省略された項目の列は空欄となります。

      • ファイルへの保存やタイムスタンプの挿入などは、自動で行なわれます。

実験の実行

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


    def steps(self, ctx: ExperimentContext):
        multimeter = K34411A()
        while True:
            r = multimeter.measure_resistance()
            ctx.send_row({"R": r})
            ctx.loop()


if __name__ == "__main__":
    launch_experiment([ContinuousResistanceMesurement])

注釈

複数の実験を指定することができます。