🧪 3. ノードをテストする¶

mujou.run で始めた実行では dt が実時間の経過で決まり、ノードの更新順序も毎回違うので、結果は毎回わずかに変わる。このページでは 2. ノードを書くのノードを、毎回同じ結果になるように走らせる mujou.testing でテストする。

道具¶

import mujou.testing で読み込む。import mujou だけでは読み込まれない。

名前	役割
`mujou.testing.run(root, ...)`	`mujou.run` の `timer`・`inputs`・`state` を受け取って（`workers` は無い）木を走らせ、`ManualRun` を返す。ノードは `step` が更新するまで更新しない。既定では新しい `mujou.time.VirtualTimer` の上で走る
`ManualRun.step(dt, *nodes)`	タイマーを `dt` 秒進めてから、渡したノードを渡した順に 1 回ずつ更新する
`ManualRun.timer`	実行の `VirtualTimer`

ManualRun は Run なので、write・read・state_dict なども mujou.run と同じように使える。

テスト¶

class TestLeaky:
    def test_follows_the_exact_solution(self) -> None:
        tau, h, steps = 0.1, 0.001, 100
        node = Leaky(tau)
        one = torch.tensor(1.0, dtype=torch.float64)
        expected = 0.0
        with mujou.testing.run(node, inputs={node.prev: one}) as run:
            for _ in range(steps):
                run.step(h)  # 1 ms later, then every node once (period is ignored)
                expected += (1 - math.exp(-h / tau)) * (1.0 - expected)
            y, t = run.read_with_time(node.next)
        assert t is not None and t.ns == steps * 1_000_000  # sent at the last step
        torch.testing.assert_close(y, torch.tensor(expected, dtype=torch.float64))

uv run pytest examples/test_leaky.py

mujou.testing.run(node, inputs=...) はノードを時刻 0 の新しい VirtualTimer の上で走らせる。inputs は開始時からの入力の Port の値
各 run.step(h) は時間を h 秒進めてから全ノードを 1 回ずつ更新する。ノードの dt はちょうど h になる
参照は同じ係数で計算した厳密解。torch.testing.assert_close で比べる。参照が同じ演算を同じ順に行うなら torch.equal で比べられる（次の節）
ノードが出力を送った時刻はちょうど steps ミリ秒。手動の実行には実時間が入らない

更新の順序¶

class TestLeakyChain:
    def test_follows_the_discrete_algorithm(self) -> None:
        tau, h, steps = 0.01, 0.001, 50
        u = torch.tensor(1.0, dtype=torch.float64)
        # The reference: the discrete algorithm, a then b at every step.
        a, b = SCALAR.zeros(), SCALAR.zeros()
        for _ in range(steps):
            a += (1 - math.exp(-h / tau)) * (u - a)
            b += (1 - math.exp(-h / tau)) * (a - b)  # b reads the new a

        net = LeakyChain(tau)
        with mujou.testing.run(net, inputs={net.prev: u}) as run:
            for _ in range(steps):
                run.step(h, net.a, net.b)  # a, then b, which reads a's new value
            y = run.read(net.next)
        assert torch.equal(y, b)  # the same operations: equal to the last bit

ノードを渡さずに step(dt) を呼ぶと、木の順（属性を代入した順）に全ノードを 1 回ずつ更新する
順序で結果が変わる。step(h, net.a, net.b) では b は a がこのステップで送った値を読み、step(h, net.b, net.a) なら前のステップの値を読む
ノードの dt は前回の更新からの時間で、最初は実行の開始から。同じ step に 2 回渡したノードは 1 回だけ更新する。2 回目は時刻が進んでいないので飛ばす
ノードの周期は手動の実行では使われない
ノードが例外を投げると、step がその場で NetworkError を送出する

保存と復元¶

途中で状態を取得し、新しいインスタンスで続けた結果は、止めずに続けた結果とビット単位で一致する。

def run_steps(n: int, state: dict[str, Any] | None = None) -> dict[str, Any]:
    """Run a new ``LeakyChain`` for ``n`` steps of 1 ms and return its state."""
    net = LeakyChain(0.01)
    u = torch.tensor(1.0, dtype=torch.float64)
    with mujou.testing.run(net, inputs={net.prev: u}, state=state) as run:
        for _ in range(n):
            run.step(0.001)
    return run.state_dict()


class TestRestore:
    def test_a_restored_run_goes_on_as_one_that_did_not_stop(self) -> None:
        straight = run_steps(20)
        # Ten steps, then ten more in a new instance restored from the first half,
        # on a VirtualTimer of its own that starts at the time of the state.
        resumed = run_steps(10, state=run_steps(10))
        x = resumed["nodes"]["b"]["state"]["x"]
        assert torch.equal(x, straight["nodes"]["b"]["state"]["x"])

状態から復元した実行は、その状態の時刻から始まる自分の VirtualTimer で走る。

例外¶

ノードが例外を投げると、step は NetworkError を送出し、その __cause__ がノードの例外になる。以後は with ブロックの終わりの shutdown を含め、実行のどの呼び出しも同じ例外を送出する。そのため pytest.raises はブロック全体を囲むように同じ with 文に並べる。

with pytest.raises(NetworkError) as info, mujou.testing.run(net) as run:
    run.step(0.001)  # ここでノードが ValueError を投げる
assert isinstance(info.value.__cause__, ValueError)

決定的にするためのほかの条件¶

乱数を使うノードには seed を渡し、乱数は self.generator から引く（スレッドとデバイス）
時間は step でだけ進める。step を呼ばなければ何も送られないので、step を呼ぶスレッドで timeout なしに wait_for や subscribe で待つと、いつまでも終わらない
複数の実行を 1 本の時間軸に載せるには、mujou.time.VirtualTimer() を作って同じものを各実行に渡す。1 つの実行の step でほかも進む

次へ¶

学習するノード: 4. ノードの中で学習する