Testing

mujou.testing

Deterministic tests of nodes and networks.

Run the networks with run, which puts them on a new VirtualTimer, and step the nodes by hand by a fixed dt in a fixed order: every update then sees exactly that dt, and the result can be compared with the discrete algorithm. Mujou tests its own nodes with the same API.

Example::

from typing import override

import torch

import mujou
import mujou.testing
from mujou import Node, Port, TensorSpec


class Double(Node):
    def __init__(self) -> None:
        super().__init__()
        self.prev = Port(recv=TensorSpec((2,)))
        self.next = Port(send=TensorSpec((2,)))

    @override
    def on_update(self, dt: float) -> None:
        super().on_update(dt)
        self.next.send(2 * self.prev.recv())


class Chain(mujou.Network):
    def __init__(self) -> None:
        super().__init__()
        self.a, self.b = Double(), Double()
        self.a.next.connect(self.b.prev)
        self.prev, self.next = self.a.prev, self.b.next


net = Chain()
x = torch.ones(2)
with mujou.testing.run(net, inputs={net.prev: x}) as run:
    run.step(0.001, net.a, net.b)  # 1 ms later, a, then b reads a's new value
    y, t = run.read_with_time(net.next)
assert torch.equal(y, 4 * x) and t is not None and t.ns == 1_000_000

ManualRun

ManualRun(wiring: Wiring, runtime: Runtime, timer: VirtualTimer, clock: ReadOnlyTimer)

Bases: Run

A run whose nodes update only when step lets them, for tests.

mujou.testing.run returns it. The nodes start and stop as in any run, but between updates each waits for step, which advances the run's VirtualTimer first: every dt is exactly the time the test stepped, and the result follows the discrete algorithm of the chosen order.

timer property

timer: VirtualTimer

The run's timer: the VirtualTimer that mujou.testing.run made, or the one given to it.

step

step(dt: float, *nodes: Node) -> None

Advance the timer by dt seconds, then update each of nodes once, one after another in the given order, and return when done.

Without nodes, every node updates in the order of the tree (each sees what the earlier ones sent in the same step, as Gauss-Seidel does; give them from the output side back for the order of Jacobi). The period of the nodes does not apply. Each dt a node sees is the time since its last update, the first one from the start of the run (or the last update restored). A node given twice in one step updates once: the second time, its time has not moved. A timer shared with other runs moves for them too.

RAISES	DESCRIPTION
ValueError	A node is not in this run, or dt is negative or not finite.
RuntimeError	The run has stopped normally, or the timer is paused.
NetworkError	A node raised; the run has stopped.

run

run(
    root: Node | Network,
    *,
    timer: VirtualTimer | None = None,
    inputs: Mapping[P, Message] | None = None,
    state: Mapping[str, Any] | None = None,
) -> ManualRun

Run root as mujou.run does, with the updates left to ManualRun.step.

The default timer is a new mujou.time.VirtualTimer at time 0 (with state, at the time of the state), which ManualRun.timer returns and ManualRun.step advances. Pass a VirtualTimer of your own to share it between runs: a step of one then moves the time of the others too.

inputs and state are those of mujou.run, and so are the errors. There is no workers: the nodes update one at a time, in the order that ManualRun.step gives.