Targets and runtimes

A Target* lowers a Module to a backend. TargetNumpy returns the matching native-Python runtime view; TargetCpp emits an Eigen C++ library; TargetWasm emits a browser bundle (WASM + JS); TargetJax emits a jitted rollout.

Targets

manta.TargetNumpy

TargetNumpy(x: Sim, *, compile: bool = False, optimization: str | None = None, compile_timeout_s: float | None = DEFAULT_COMPILATION_TIMEOUT_S, max_instructions: int | None = DEFAULT_MAX_INSTRUCTIONS) -> NumpySim

TargetNumpy(x: Any, *, compile: bool = False, optimization: str | None = None, compile_timeout_s: float | None = DEFAULT_COMPILATION_TIMEOUT_S, max_instructions: int | None = DEFAULT_MAX_INSTRUCTIONS) -> NumpyRuntime

TargetNumpy(x, *, compile=False, optimization=None, compile_timeout_s=DEFAULT_COMPILATION_TIMEOUT_S, max_instructions=DEFAULT_MAX_INSTRUCTIONS)

Lower a typed Module — or any transform exposing .module() (Sim, EKF, LQR, a recurrence block) — to the matching native-Python view (sim / filter / recurrence / regulator), or the bare kernel engine when no view matches.

compile=True builds the kernel's CasADi functions with optimized native code (O1 by default for full-simulation graphs, -O3 -march=native for other runtime models). Each artifact caller may explicitly select a startup/balanced/runtime profile or O0/O1/O2 according to its own compile/runtime tradeoff and cold-build ceiling. It calls them as externals instead of interpreting the MX graph. Results are cached on disk; the default cold-build deadline is five minutes, and the artifact caller may replace or disable that deadline. It raises CompilationError if an external cannot be produced; explicit native execution never silently becomes interpretation. Pair with NumpySim's step_n to fold substeps for a further amortization.

max_instructions is the cost-benefit size gate, counted in CasADi instructions over every kernel (default DEFAULT_MAX_INSTRUCTIONS). Above it compilation is refused with an error naming this parameter. A deliberately large full-truth simulation whose owner has declared a finite cold-build ceiling raises it, or passes None to disable the gate entirely.

Source code in manta/codegen/numpy/__init__.py

def TargetNumpy(
    x: Any,
    *,
    compile: bool = False,
    optimization: str | None = None,
    compile_timeout_s: float | None = DEFAULT_COMPILATION_TIMEOUT_S,
    max_instructions: int | None = DEFAULT_MAX_INSTRUCTIONS,
) -> NumpyRuntime:
    """Lower a typed `Module` — or any transform exposing `.module()`
    (`Sim`, `EKF`, `LQR`, a recurrence block) — to the matching
    native-Python view (sim / filter / recurrence / regulator), or the
    bare kernel engine when no view matches.

    `compile=True` builds the kernel's CasADi functions with optimized native
    code (O1 by default for full-simulation graphs, `-O3 -march=native` for
    other runtime models). Each artifact caller may explicitly select a
    startup/balanced/runtime profile or O0/O1/O2 according to its own
    compile/runtime tradeoff and cold-build ceiling. It
    calls them as externals instead of interpreting the MX graph. Results are
    cached on disk; the default cold-build deadline is five minutes, and the
    artifact caller may replace or disable that deadline. It raises
    `CompilationError` if an external cannot be produced; explicit native
    execution never silently becomes interpretation. Pair with `NumpySim`'s
    `step_n` to fold substeps for a further amortization.

    `max_instructions` is the cost-benefit size gate, counted in CasADi
    instructions over every kernel (default `DEFAULT_MAX_INSTRUCTIONS`).
    Above it compilation is refused with an error naming this parameter. A
    deliberately large full-truth simulation whose owner has declared a
    finite cold-build ceiling raises it, or passes `None` to disable the
    gate entirely."""
    from ..target import as_module
    m = as_module(x, "TargetNumpy")
    if (
        optimization is not None
        or compile_timeout_s != DEFAULT_COMPILATION_TIMEOUT_S
    ) and not compile:
        raise ValueError(
            "optimization and compile_timeout_s require compile=True"
        )
    if max_instructions != DEFAULT_MAX_INSTRUCTIONS and not compile:
        raise ValueError("max_instructions requires compile=True")
    validate_max_instructions(max_instructions)
    if optimization is not None and optimization not in {
        "startup", "balanced", "runtime", "O0", "O1", "O2"
    }:
        raise ValueError(
            "optimization must be startup/balanced/runtime or O0/O1/O2"
        )
    runtime = _select_view(m)(m)
    return (
        runtime._enable_compile(
            optimization=optimization,
            timeout_s=compile_timeout_s,
            max_instructions=max_instructions,
        )
        if compile else runtime
    )

manta.TargetFilterReplay

TargetFilterReplay(value, *, max_operations, max_checkpoints, max_execution_bytes=_DEFAULT_EXECUTION_BYTE_CAP, optimization='runtime')

Generate a bounded exact-sequential EKF/UKF replay target.

Source code in manta/codegen/numpy/_filter_replay.py

def TargetFilterReplay(
    value,
    *,
    max_operations: int,
    max_checkpoints: int,
    max_execution_bytes: int = _DEFAULT_EXECUTION_BYTE_CAP,
    optimization: str = "runtime",
) -> NativeFilterReplay:
    """Generate a bounded exact-sequential EKF/UKF replay target."""
    return NativeFilterReplay(
        value,
        max_operations=max_operations,
        max_checkpoints=max_checkpoints,
        max_execution_bytes=max_execution_bytes,
        optimization=optimization,
    )

manta.TargetCpp

TargetCpp(x, out_dir, *, class_name, basename=None, namespace='manta_gen')

C++ codegen target.

Args: x — a Module, or a transform with .module(). out_dir — destination directory (created if missing). class_name — C++ class name. Conventionally PascalCase. basename — filename stem; defaults to class_name.lower(). namespace — C++ namespace enclosing the emitted class.

Returns: EmitResult with paths to every emitted file plus a small funcs summary (world_name / dims).

Source code in manta/codegen/cpp/__init__.py

def TargetCpp(x,
              out_dir: str | Path,
              *,
              class_name: str,
              basename: str | None = None,
              namespace: str = "manta_gen") -> EmitResult:
    """C++ codegen target.

    Args:
        x           — a `Module`, or a transform with `.module()`.
        out_dir     — destination directory (created if missing).
        class_name  — C++ class name. Conventionally PascalCase.
        basename    — filename stem; defaults to `class_name.lower()`.
        namespace   — C++ namespace enclosing the emitted class.

    Returns:
        `EmitResult` with paths to every emitted file plus a small `funcs`
        summary (world_name / dims).
    """
    return emit_module(x, out_dir, class_name=class_name,
                       basename=basename, namespace=namespace)

manta.TargetWasm

TargetWasm(x, out_dir, *, class_name, basename=None)

WASM codegen target.

Args: x — a Module, or a transform with .module(). out_dir — destination directory (created if missing). class_name — public name for the bundle. Conventionally PascalCase. basename — filename stem; defaults to class_name.lower().

Returns: WasmEmitResult with paths to every emitted file plus the descriptor.

Source code in manta/codegen/wasm/__init__.py

def TargetWasm(x,
               out_dir: str | Path,
               *,
               class_name: str,
               basename: str | None = None) -> WasmEmitResult:
    """WASM codegen target.

    Args:
        x           — a `Module`, or a transform with `.module()`.
        out_dir     — destination directory (created if missing).
        class_name  — public name for the bundle. Conventionally PascalCase.
        basename    — filename stem; defaults to `class_name.lower()`.

    Returns:
        `WasmEmitResult` with paths to every emitted file plus the descriptor.
    """
    return emit_module_wasm(x, out_dir, class_name=class_name,
                            basename=basename)

manta.TargetJax

TargetJax(x)

Lower a Module (or any transform exposing .module()) to jitted JAX kernels by name + a lax.scan rollout builder — the functional artifacts a training loop wants (see manta.codegen.jax.JaxModule). jax is imported only when this is called, so manta itself never requires it.

Source code in manta/codegen/__init__.py

def TargetJax(x):
    """Lower a Module (or any transform exposing `.module()`) to jitted
    JAX kernels by name + a `lax.scan` rollout builder — the functional
    artifacts a training loop wants (see `manta.codegen.jax.JaxModule`).
    `jax` is imported only when this is called, so manta itself never
    requires it."""
    from .jax import TargetJax as _TargetJax
    return _TargetJax(x)

manta.NoiseDriver

NoiseDriver(seed)

Draws the per-step samples that make an oracle Module's noise live.

Binds to the NOISE port's fields (name, dim, σ); each sample() is an independent N(0, σ²) draw per active channel. Deliberately thin and swappable — kept out of the pure kernels — and simply omitted on a deploy target.

Source code in manta/codegen/numpy/_noise.py

def __init__(self, seed: int) -> None:
    if isinstance(seed, bool) or not isinstance(seed, int):
        raise TypeError("NoiseDriver seed must be an explicit integer")
    self._seed = seed
    self._rng = np.random.default_rng(seed)
    self._channels: list[tuple[str, int, float]] = []

sample

sample(names=None)

Draw only active channels, or every channel when unspecified.

Dependency-scheduled simulation uses this boundary so measurement white noise advances on acquisition, not on unrelated plant ticks. Channel order remains the bound NOISE-port order for deterministic replay.

Source code in manta/codegen/numpy/_noise.py

def sample(self, names: set[str] | frozenset[str] | None = None
           ) -> dict[str, np.ndarray]:
    """Draw only active channels, or every channel when unspecified.

    Dependency-scheduled simulation uses this boundary so measurement
    white noise advances on acquisition, not on unrelated plant ticks.
    Channel order remains the bound NOISE-port order for deterministic
    replay.
    """
    if names is not None:
        known = {name for name, _dim, _sigma in self._channels}
        unknown = set(names) - known
        if unknown:
            raise KeyError(f"NoiseDriver.sample: unknown channel(s) {sorted(unknown)}")
    return {name: self._rng.normal(0.0, sigma, dim)
            for name, dim, sigma in self._channels
            if sigma > 0.0 and (names is None or name in names)}

Numpy runtime views

The view TargetNumpy(x) returns is determined by the Module's shape.

manta.codegen.NumpyRuntime

NumpyRuntime(module)

The generic engine over a typed Module: state storage + the typed-arg gather → kernel call → scatter. Views subclass it.

Source code in manta/codegen/numpy/_runtime.py

def __init__(self, module: Module) -> None:
    self.module = module
    self._functions = module.functions      # swapped by _enable_compile()
    self._spec = module.spec
    self._state: dict[str, np.ndarray] = {}
    for f in module.state.fields:
        a = finite_array(f.init, who=f"{module.name} state {f.name!r}",
                         size=int(np.prod(f.shape)))
        # Always copy: the Module is shared, immutable IR — a view
        # here would alias every runtime built from the same
        # transform onto one array (and let a runtime mutate the
        # Module's init in place).
        self._state[f.name] = (a.reshape(f.shape).copy()
                               if f.kind == "matrix"
                               else a.reshape(-1).copy())

    self._u_port = module.sole_port(Role.CONTROL)
    self._noise_port = module.sole_port(Role.NOISE)
    self._meas_ports_ir = module.ports_by_role(Role.MEASUREMENT)
    self._y_port = module.sole_port(Role.OUTPUT)
    self._x_port = module.sole_port(Role.STATE)
    self._param_port = module.sole_port(Role.PARAMETER)
    self._param_overrides: dict[str, np.ndarray] = {}
    self._param_vector_cache: np.ndarray | None = None
    # Keyed by the actual ca.Function identity because selected functions
    # may be replaced by compiled externals after construction.
    self._evaluation_buffers: dict[int, _DenseEvaluationBuffer | None] = {}
    # Per entry point: argument sizes and output layouts resolved once.
    # The Module is immutable, so this never goes stale; resolving it
    # per call was most of a simulator tick for wide truth plants.
    self._entry_plans: dict[str, _EntryPlan] = {}
    self._input_name_list = [f.name for f in self._u_fields()]
    # Most estimator calls carry no vehicle control. Packing immutable
    # Module defaults used to resolve suffixes, allocate arrays, and
    # revalidate the same declaration on every predict/update. Keep one
    # pristine vector for internal read-only kernel calls; the public
    # ``build_u`` API still returns owned storage.
    self._default_u_vector = pack_fields(
        self._u_fields(), {}, default=lambda f: f.default, who="build_u"
    )
    self._state_revision = 0

    self._t = 0.0

state_revision property

state_revision

Monotonic revision for consumers caching state projections.

input_names property

input_names

The Module's declared control-input names, in order.

compile_functions

compile_functions(function_names, *, optimization='balanced', timeout_s=DEFAULT_COMPILATION_TIMEOUT_S, max_instructions=DEFAULT_MAX_INSTRUCTIONS)

Compile a selected hot subset of this runtime's kernels.

Large transforms need not make native execution all-or-nothing. A rate loop can compile its dominant kernel while retaining interpreted entry points that execute rarely. Selection is by stable Module function identity, and a failure leaves the runtime unchanged. max_instructions raises or disables (None) the size gate.

Source code in manta/codegen/numpy/_runtime.py

def compile_functions(
    self,
    function_names: Iterable[str],
    *,
    optimization: Optimization = "balanced",
    timeout_s: float = DEFAULT_COMPILATION_TIMEOUT_S,
    max_instructions: int | None = DEFAULT_MAX_INSTRUCTIONS,
) -> NumpyRuntime:
    """Compile a selected hot subset of this runtime's kernels.

    Large transforms need not make native execution all-or-nothing. A
    rate loop can compile its dominant kernel while retaining interpreted
    entry points that execute rarely. Selection is by stable Module
    function identity, and a failure leaves the runtime unchanged.
    ``max_instructions`` raises or disables (``None``) the size gate.
    """
    names = tuple(dict.fromkeys(function_names))
    if not names:
        raise ValueError("compile_functions requires at least one function name")
    unknown = sorted(set(names) - set(self.module.functions))
    if unknown:
        raise KeyError(f"unknown Module function(s) {unknown}")
    selected = {name: self.module.functions[name] for name in names}
    compiled = compile_functions(
        selected, max_instructions=max_instructions,
        optimization=optimization, timeout_s=timeout_s,
    )
    self._functions = {**self._functions, **compiled}
    return self

call

call(method, values=None, **kw)

Run one entry point. values/kwargs are keyed by port or state name (use the dict for dotted names); TIME ports default to 0.

The Hosting contract: a THREADED module's state is the caller's — supply state fields by name in values (unsupplied ones fall back to the engine's last-written copy) and read the fresh writes from the returned dict alongside the entry's returns. A HELD module's state lives in the runtime and is read/written in place; only the entry's returns come back.

Source code in manta/codegen/numpy/_runtime.py

def call(self, method: str, values: dict[str, Any] | None = None,
         **kw) -> dict[str, np.ndarray]:
    """Run one entry point. `values`/kwargs are keyed by port or state
    name (use the dict for dotted names); TIME ports default to 0.

    The Hosting contract: a THREADED module's state is the caller's —
    supply state fields by name in `values` (unsupplied ones fall back
    to the engine's last-written copy) and read the fresh writes from
    the returned dict alongside the entry's returns. A HELD module's
    state lives in the runtime and is read/written in place; only the
    entry's returns come back."""
    vals = dict(values or {})
    vals.update(kw)
    return self._run(self.module.entry(method), vals)

build_u

build_u(u)

Resolve a {name: value} dict (full or suffix names) to the flat control vector over the Module's declared defaults.

Source code in manta/codegen/numpy/_runtime.py

def build_u(self, u: dict[str, Any] | None) -> np.ndarray:
    """Resolve a `{name: value}` dict (full or suffix names) to the
    flat control vector over the Module's declared defaults."""
    if not u:
        return self._default_u_vector.copy()
    names = self._input_names()
    source = {resolve_suffix(k, names, label="input",
                             who=type(self).__name__): v
              for k, v in (u or {}).items()}
    return pack_fields(self._u_fields(), source,
                       default=lambda f: f.default, who="build_u")

set_parameters

set_parameters(values)

Override promoted-parameter values (full or suffix names); every subsequent kernel call uses them. Values not overridden stay at the Module's declared defaults.

Source code in manta/codegen/numpy/_runtime.py

def set_parameters(self, values: dict[str, Any]) -> None:
    """Override promoted-parameter values (full or suffix names);
    every subsequent kernel call uses them. Values not overridden
    stay at the Module's declared defaults."""
    if self._param_port is None:
        raise ValueError(
            f"{self.module.name}: module declares no parameter port — "
            f"build the transform with parameters=[...] to promote "
            f"tunable Parameters.")
    names = [f.name for f in self._param_port.fields]
    dims = {f.name: f.dim for f in self._param_port.fields}
    staged = dict(self._param_overrides)
    for k, v in values.items():
        full = resolve_suffix(k, names, label="parameter",
                              who=type(self).__name__)
        arr = finite_array(v, who=f"set_parameters: {full!r}",
                           size=dims[full]).ravel()
        staged[full] = arr
    self._param_overrides = staged
    self._param_vector_cache = None

param_vector

param_vector()

The flat promoted-parameter vector: declared defaults merged with set_parameters overrides, in port-field order.

Source code in manta/codegen/numpy/_runtime.py

def param_vector(self) -> np.ndarray:
    """The flat promoted-parameter vector: declared defaults merged
    with `set_parameters` overrides, in port-field order."""
    return self._kernel_param_vector().copy()

manta.codegen.NumpySim

NumpySim(module)

Bases: NumpyRuntime

The simulation oracle. The runtime holds the nested state dict (sim.state); step(dt, u={...}) applies the commands and advances it, realizing that step's sensor readings. Rate-limited measurement kernels are called only when due and their last values are held between samples. Read sensors with outputs() (raw nested) or reading(name) (one, by name).

Source code in manta/codegen/numpy/_sim.py

def __init__(self, module) -> None:
    super().__init__(module)
    self._driver: NoiseDriver | None = None
    self._outputs: dict[str, dict[str, Any]] = {}
    self._sim_state: PackedSimState | None = None
    self._stepn_cache: OrderedDict[int, Any] = OrderedDict()
    self._coupled_models: list[Any] = []
    profile = module.metadata.get("transform_profile", {})
    raw_noise_dependencies = profile.get("noise_dependencies")
    self._noise_dependencies = (
        None
        if raw_noise_dependencies is None
        else {
            method: frozenset(names)
            for method, names in raw_noise_dependencies
        }
    )
    contract = tuple(profile.get("scheduled_measurement_groups", ()))
    self._scheduled: dict[str, tuple[tuple[str, ...], float]] = {}
    for item in contract:
        if not isinstance(item, (tuple, list)) or len(item) != 3:
            raise ValueError(
                f"{module.name}: invalid scheduled measurement metadata")
        method, rate, fulls = item
        fulls = tuple(fulls)
        if not fulls:
            raise ValueError(
                f"{module.name}: scheduled group {method!r} is empty")
        rate = float(require_positive(
            rate, name=f"{module.name} measurement group rate {method!r}"))
        for full in fulls:
            port = module.port(full)
            if port.role is not Role.MEASUREMENT or port.rate != rate:
                raise ValueError(
                    f"{module.name}: schedule for {full!r} disagrees "
                    "with its measurement port")
        ep = module.entry(method)
        if ep.writes or ep.returns != fulls:
            raise ValueError(
                f"{module.name}: scheduled entry {method!r} returns "
                "a different measurement group")
        self._scheduled[method] = (fulls, 1.0 / rate)
    self._next_sample = {method: 0.0 for method in self._scheduled}

state property writable

state

The held nested state (lazy-seeded; mutate in place to set commands or override slots).

Aliasing rule: the OWNER DICTS stay live across steps (holding st = sim.state['craft'] keeps working), but the slot VALUES are replaced each step — a reference to sim.state['c']['position'] goes stale after step(); read through the dict, don't cache the array. Unknown keys are rejected at the next step() (a typo'd slot would otherwise be a silent no-op).

time property

time

Current logical time shared by the spatial and coupled models.

initial_state

initial_state()

Fresh nested initial state: the manifold slots' defaults plus input/noise placeholder entries (commands you may set; noise seeds that stay at zero — a NoiseDriver draw never enters the dict).

Source code in manta/codegen/numpy/_sim.py

def initial_state(self) -> dict[str, dict[str, Any]]:
    """Fresh nested initial state: the manifold slots' defaults plus
    input/noise placeholder entries (commands you may set; noise seeds
    that stay at zero — a `NoiseDriver` draw never enters the dict)."""
    nested = self._spec.to_nested(self.module.state.field("x").init)
    for f in self._u_fields():
        owner, rest = _split(f.name)
        nested.setdefault(owner, {}).setdefault(rest, f.default)
    if self._noise_port is not None:
        for f in self._noise_port.fields:
            owner, rest = _split(f.name)
            nested.setdefault(owner, {}).setdefault(
                rest, np.zeros(f.dim) if f.dim > 1 else 0.0)
    return nested

model_state

model_state()

Only manifold state slots, excluding commands/noise placeholders.

Source code in manta/codegen/numpy/_sim.py

def model_state(self) -> dict[str, dict[str, Any]]:
    """Only manifold state slots, excluding commands/noise placeholders."""
    flat = flatten_nested(self.state)
    return self._spec.to_nested(self._spec.pack_projected(flat))

step

step(dt, *, t=None, u=None)

Advance the held state by dt. u is {input: value} (full or suffix names) applied this step over the held sim.state inputs. Runs the oracle kernel (one noise draw) and returns the new state dict. Downstream code owns any actuator intake hold policy.

Source code in manta/codegen/numpy/_sim.py

def step(self, dt: float, *, t: float | None = None,
         u: dict[str, Any] | None = None
         ) -> dict[str, dict[str, Any]]:
    """Advance the held state by `dt`. `u` is `{input: value}` (full or
    suffix names) applied this step over the held `sim.state` inputs.
    Runs the oracle kernel (one noise draw) and returns the new state
    dict. Downstream code owns any actuator intake hold policy."""
    if isinstance(dt, dict):
        raise TypeError(
            "NumpySim.step: the functional step(state, dt) form was "
            "removed — pass commands as step(dt, u={...}).")
    dt = require_positive(dt, name="NumpySim.step dt")
    t0 = self._t if t is None else float(require_finite(t, name="NumpySim.step t"))
    schedule_resync = (
        t is not None
        and abs(t0 - self._t) > 1e-12 * max(1.0, abs(t0), abs(self._t))
    )
    next_t = float(require_finite(
        t0 + dt, name="NumpySim.step resulting time"))
    if not self._coupled_models:
        noise_before = self._driver.checkpoint() if self._driver else None
        x_storage_before = self._state["x"]
        x_before = x_storage_before.copy()
        try:
            self._sim_state = self._advance(
                self.state, dt, t0, u,
                reset_schedule=schedule_resync)
        except Exception:
            x_storage_before[:] = x_before
            self._state["x"] = x_storage_before
            if self._driver is not None and noise_before is not None:
                self._driver.restore(noise_before)
            raise
        self._t = next_t
        return self._sim_state

    before = self.checkpoint()
    try:
        merged_u = self._coupled_inputs(u, t0, dt)
        self._sim_state = self._advance(
            self.state, dt, t0, merged_u,
            reset_schedule=schedule_resync)
        for model in self._coupled_models:
            model.post_step(self, next_t, dt)
    except Exception:
        self.restore(before)
        raise
    self._t = next_t
    return self._sim_state

attach_model

attach_model(model)

Attach physical non-spatial state to this simulation clock.

A coupled model contributes pre-step Manta inputs and advances once after the corresponding physics tick. Checkpoint/restore and failures are atomic across the spatial model, noise driver, and every attached model.

Source code in manta/codegen/numpy/_sim.py

def attach_model(self, model):
    """Attach physical non-spatial state to this simulation clock.

    A coupled model contributes pre-step Manta inputs and advances once
    after the corresponding physics tick. Checkpoint/restore and failures
    are atomic across the spatial model, noise driver, and every attached
    model.
    """
    required = ("name", "bind", "inputs", "post_step", "checkpoint",
                "validate_checkpoint", "restore")
    missing = [name for name in required if not hasattr(model, name)]
    if missing:
        raise TypeError(f"coupled model is missing {missing}")
    if (not isinstance(model.name, str) or not model.name.isidentifier()
            or any(existing.name == model.name
                   for existing in self._coupled_models)):
        raise ValueError("coupled model name must be a unique identifier")
    model.bind(self)
    self._coupled_models.append(model)
    return model

step_n

step_n(dt, n, *, t=None, u=None)

Advance n substeps of dt in ONE folded call — u commands held (ZOH) for the block, state chained through a mapaccum of the step kernel. Output readings + state are bit-identical to n sequential step(dt, u=u) calls; compiled (_enable_compile) it runs the whole inner loop in C.

Falls back to sequential stepping when a NoiseDriver is attached (a fresh stochastic draw per substep cannot be folded).

Source code in manta/codegen/numpy/_sim.py

def step_n(self, dt: float, n: int, *, t: float | None = None,
           u: dict[str, Any] | None = None
           ) -> dict[str, dict[str, Any]]:
    """Advance `n` substeps of `dt` in ONE folded call — `u` commands held
    (ZOH) for the block, state chained through a `mapaccum` of the step
    kernel. Output readings + state are bit-identical to `n` sequential
    `step(dt, u=u)` calls; compiled (`_enable_compile`) it runs the whole
    inner loop in C.

    Falls back to sequential stepping when a `NoiseDriver` is attached (a
    fresh stochastic draw per substep cannot be folded)."""
    dt = require_positive(dt, name="NumpySim.step_n dt")
    if isinstance(n, bool) or int(n) != n or n < 0:
        raise ValueError(f"NumpySim.step_n n must be a non-negative integer, got {n!r}")
    n = int(n)
    if t is not None:
        t = float(require_finite(t, name="NumpySim.step_n t"))
    if (n <= 1 or self._driver is not None or self._coupled_models
            or self._scheduled):
        before = self.checkpoint()
        try:
            for k in range(n):
                self.step(dt, t=None if t is None else t + k * dt, u=u)
        except Exception:
            self.restore(before)
            raise
        return self.state          # property: seeds when n == 0
    t0 = self._t if t is None else t
    next_t = float(require_finite(
        t0 + n * dt, name="NumpySim.step_n resulting time"))
    before = self.checkpoint()
    try:
        self._sim_state = self._advance_n(self.state, dt, n, t0, u)
    except Exception:
        self.restore(before)
        raise
    self._t = next_t
    return self._sim_state

outputs

outputs()

Sensor readings from the most recent step (nested, realized with that step's noise draw).

Source code in manta/codegen/numpy/_sim.py

def outputs(self) -> dict[str, dict[str, Any]]:
    """Sensor readings from the most recent step (nested, realized
    with that step's noise draw)."""
    return self._outputs

attach_driver

attach_driver(driver)

Attach a stochastic NoiseDriver: every active (σ>0) channel of the Module's NOISE port is sampled each step, so truth is noisy with the very σ the EKF reads for R/Q. Without one the sim is a noiseless oracle.

Source code in manta/codegen/numpy/_sim.py

def attach_driver(self, driver: NoiseDriver) -> NoiseDriver:
    """Attach a stochastic `NoiseDriver`: every active (σ>0) channel of
    the Module's NOISE port is sampled each step, so truth is noisy
    with the very σ the EKF reads for R/Q. Without one the sim is a
    noiseless oracle."""
    if self._noise_port is None:
        raise ValueError(
            f"{self.module.name}: module declares no noise port.")
    driver.bind(self._noise_port.fields)
    self._driver = driver
    return driver

reading

reading(name)

The latest raw reading for a sensor (full or suffix name) from the most recent acquisition. A rate-limited reading is held between acquisitions; an unrated reading is realized every plant step.

Source code in manta/codegen/numpy/_sim.py

def reading(self, name: str) -> Any:
    """The latest raw reading for a sensor (full or suffix name) from the
    most recent acquisition. A rate-limited reading is held between
    acquisitions; an unrated reading is realized every plant step."""
    full = resolve_suffix(name, [p.name for p in self._meas_ports_ir],
                          label="output", who=type(self).__name__)
    owner, slot = _split(full)
    return self._outputs.get(owner, {}).get(slot)

manta.codegen.NumpyFilter

NumpyFilter(module)

Bases: NumpyRuntime

A predict/update filter over a held x/P, with baked per-sensor update kernels — the same surface every backend emits.

You own the loop, identically in numpy and C++: fold each fresh measurement at the pre-predict state, then predict.

for nm in sensors:
    if gate[nm].due(t):
        ekf.update(nm, sim.reading(nm), u=u)   # update-then-...
ekf.predict(dt, u=u)                           # ...-predict

The update-then-predict order is yours to keep: a reading sampled at the interval start belongs against the current (pre-predict) state.

Clock: same convention as NumpySim — the runtime tracks t, predict(dt) advances it, and an explicit t= overrides it for that call. (The kernels stay pure; this is caller-side bookkeeping the two runtimes must agree on: a filter that silently pinned t=0 while the sim advanced left every time-dependent world linearized at t=0.)

Source code in manta/codegen/numpy/_filter.py

def __init__(self, module) -> None:
    super().__init__(module)
    self._Q: np.ndarray | None = None        # default process noise
    self._custom_h_cache: OrderedDict = OrderedDict()
    rho_warning = module.metadata.get("rho_warning")
    for sensor, rho in module.metadata.get("rho_by_sensor", {}).items():
        if rho_warning is not None and rho > rho_warning:
            _LOG.warning(
                "%s disturbance-observer noise ratio rho[%s]=%.6g exceeds "
                "the warning level %.3g (ceiling %.3g)",
                module.name, sensor, rho, rho_warning,
                module.metadata["rho_ceiling"])
            continue
        # rho is a useful dimensionless diagnostic, not a universal
        # estimator-selection threshold. Its acceptable range depends on
        # the identified model error, spectra, operating envelope, and
        # application policy, none of which this generic runtime owns.
        _LOG.info("%s disturbance-observer noise ratio rho[%s]=%.6g",
                  module.name, sensor, rho)

P_consider property

P_consider

Navigation-to-Schmidt-nuisance cross-covariance, when present.

time property

time

Current logical filter time (advanced only by predict).

rho_by_sensor property

rho_by_sensor

INS IMU/model noise ratio diagnostics; empty for EKF/UKF.

Q property writable

Q

Default process noise for predict (overridden per-call by predict(dt, Q=...); None uses the model's baked L Σ Lᵀ).

state_dict

state_dict()

Current estimate nested by owner.

Source code in manta/codegen/numpy/_filter.py

def state_dict(self) -> dict[str, dict[str, Any]]:
    """Current estimate nested by owner."""
    return self._spec.to_nested(self._state["x"])

reset

reset(state=None, *, P=None)

Reset state, covariance, and clock from the Module defaults.

state is merged over the declared initial state and P may replace the declared initial covariance. To move only the nominal state while deliberately preserving covariance, use :meth:set_state_keep_covariance.

Source code in manta/codegen/numpy/_filter.py

def reset(self, state: dict | None = None, *,
          P: np.ndarray | None = None) -> None:
    """Reset state, covariance, and clock from the Module defaults.

    ``state`` is merged over the declared initial state and ``P`` may
    replace the declared initial covariance. To move only the nominal
    state while deliberately preserving covariance, use
    :meth:`set_state_keep_covariance`.
    """
    physical_prior = "initialize_prior" in self.module.functions
    x_field = (self.module.port("prior_x") if physical_prior
               else self.module.state.field("x"))
    packing_spec = x_field.spec if physical_prior else self._spec
    next_x = (packing_spec.pack_any(state, base=x_field.init)
              if state is not None else
              np.asarray(x_field.init, dtype=float).reshape(-1).copy())
    pf = self.module.port("prior_P") if physical_prior else self.module.state.field("P")
    next_P = np.asarray(pf.init, dtype=float).reshape(pf.shape).copy()
    if P is not None:
        next_P = self._validate_covariance(
            P, who="reset P", positive_definite=False, dim=pf.shape[0])
    if physical_prior:
        mapped = self._functions["initialize_prior"](next_x, next_P)
        next_x, next_P = np.asarray(mapped[0]).ravel(), np.asarray(mapped[1])
    staged = {"x": next_x, "P": next_P}
    if "P_consider" in self._state:
        field = self.module.state.field("P_consider")
        staged["P_consider"] = np.asarray(
            field.init, dtype=float
        ).reshape(field.shape).copy()
    self._validate_staged_state(staged)
    self._state.update(staged)
    self._state_revision += 1
    self._t = 0.0

reset_from_model_record

reset_from_model_record(record, *, P=None)

Reset from a broader authoring record containing inputs/noise.

This explicit projection is for Craft.initial_state()-style records. Ordinary :meth:reset remains strict so typoed state keys cannot disappear among unrelated model fields.

Source code in manta/codegen/numpy/_filter.py

def reset_from_model_record(self, record: dict, *,
                            P: np.ndarray | None = None) -> None:
    """Reset from a broader authoring record containing inputs/noise.

    This explicit projection is for ``Craft.initial_state()``-style
    records.  Ordinary :meth:`reset` remains strict so typoed state keys
    cannot disappear among unrelated model fields.
    """
    projected = self._spec.pack_projected(record)
    self.reset(self._spec.to_nested(projected), P=P)

checkpoint

checkpoint()

Capture nominal state, covariance, and logical time atomically.

Source code in manta/codegen/numpy/_filter.py

def checkpoint(self) -> FilterCheckpoint:
    """Capture nominal state, covariance, and logical time atomically."""
    return FilterCheckpoint(self._state["x"], self._state["P"],
                            float(self._t), self.module.artifact_id,
                            self._state.get("P_consider"))

restore

restore(checkpoint)

Restore a checkpoint after strict shape/finite validation.

Restore never partially mutates the live filter: all values are validated and copied before any runtime field changes.

Source code in manta/codegen/numpy/_filter.py

def restore(self, checkpoint: FilterCheckpoint) -> None:
    """Restore a checkpoint after strict shape/finite validation.

    Restore never partially mutates the live filter: all values are
    validated and copied before any runtime field changes.
    """
    if not isinstance(checkpoint, FilterCheckpoint):
        raise TypeError("restore: expected FilterCheckpoint")
    if checkpoint.artifact_id != self.module.artifact_id:
        raise ValueError(
            "restore: checkpoint belongs to a different Module artifact")
    x = np.asarray(checkpoint.x, dtype=float)
    P = np.asarray(checkpoint.P, dtype=float)
    expected_x = (self._spec.ambient_dim,)
    expected_P = (self._spec.tangent_dim, self._spec.tangent_dim)
    if x.shape != expected_x:
        raise ValueError(
            f"restore: x shape {x.shape} doesn't match {expected_x}")
    if P.shape != expected_P:
        raise ValueError(
            f"restore: P shape {P.shape} doesn't match {expected_P}")
    t = float(checkpoint.time)
    if not np.all(np.isfinite(x)) or not np.all(np.isfinite(P)) \
            or not np.isfinite(t):
        raise ValueError("restore: checkpoint contains non-finite values")
    if not np.allclose(P, P.T, rtol=1e-10, atol=1e-12):
        raise ValueError("restore: P must be symmetric")
    if np.linalg.eigvalsh(P).min() < -_psd_roundoff_tolerance(P):
        raise ValueError("restore: P must be positive semidefinite")
    has_consider = "P_consider" in self._state
    if has_consider != (checkpoint.P_consider is not None):
        raise ValueError(
            "restore: checkpoint Schmidt consider-state layout differs"
        )
    staged = {"x": x.copy(), "P": P.copy()}
    if has_consider:
        field = self.module.state.field("P_consider")
        cross = np.asarray(checkpoint.P_consider, dtype=float)
        if cross.shape != field.shape:
            raise ValueError(
                f"restore: P_consider shape {cross.shape} doesn't match "
                f"{field.shape}"
            )
        staged["P_consider"] = cross.copy()
    self._validate_staged_state(staged)
    self._state.update(staged)
    self._state_revision += 1
    self._t = t

set_state_keep_covariance

set_state_keep_covariance(state)

Replace the nominal state while preserving covariance and clock.

This is intentionally separate from :meth:reset: retaining a covariance after moving its linearization point is an advanced, explicit operation.

Source code in manta/codegen/numpy/_filter.py

def set_state_keep_covariance(self, state: dict) -> None:
    """Replace the nominal state while preserving covariance and clock.

    This is intentionally separate from :meth:`reset`: retaining a
    covariance after moving its linearization point is an advanced,
    explicit operation.
    """
    self._state["x"] = self._spec.pack_any(
        state, base=self.module.state.field("x").init)
    self._state_revision += 1

preintegrated_inputs

preintegrated_inputs(packet, *, u=None)

Merge an IMUPreintegrator readout into INS inputs.

This is a naming/validation convenience for the NumPy runtime. The generated C++ filter exposes the same fields directly on Inputs. packet is the dict returned by the recurrence's step or readouts method.

Source code in manta/codegen/numpy/_filter.py

def preintegrated_inputs(
        self, packet: Mapping[str, Any], *,
        u: dict[str, Any] | None = None) -> dict[str, Any]:
    """Merge an ``IMUPreintegrator`` readout into INS inputs.

    This is a naming/validation convenience for the NumPy runtime. The
    generated C++ filter exposes the same fields directly on ``Inputs``.
    ``packet`` is the dict returned by the recurrence's ``step`` or
    ``readouts`` method.
    """
    mapping = dict(self.module.metadata.get(
        "preintegration_input_map", {}))
    if not mapping:
        raise TypeError(
            "preintegrated_inputs requires an INS constructed with "
            "propagation='preintegrated'")
    if not isinstance(packet, Mapping):
        raise TypeError("preintegrated_inputs packet must be a mapping")
    missing = sorted(set(mapping) - set(packet))
    if missing:
        raise KeyError(
            f"preintegrated_inputs packet is missing {missing}")
    merged = dict(u or {})
    input_names = self._input_names()
    occupied = {
        resolve_suffix(key, input_names, label="input",
                       who="preintegrated_inputs")
        for key in merged
    }
    collisions = sorted(occupied & set(mapping.values()))
    if collisions:
        raise ValueError(
            "preintegrated_inputs: u also supplies packet-owned input(s) "
            f"{collisions}")
    merged.update({full: packet[short]
                   for short, full in mapping.items()})
    return merged

predict_preintegrated

predict_preintegrated(packet, *, t=None, u=None, Q=None)

Advance a preintegrated INS by the packet's accumulated duration.

Source code in manta/codegen/numpy/_filter.py

def predict_preintegrated(
        self, packet: Mapping[str, Any], *, t: float | None = None,
        u: dict[str, Any] | None = None,
        Q: np.ndarray | None = None) -> None:
    """Advance a preintegrated INS by the packet's accumulated duration."""
    if "duration" not in packet:
        raise KeyError("predict_preintegrated packet is missing 'duration'")
    dt = require_positive(
        packet["duration"], name="predict_preintegrated packet duration")
    self.predict(dt, t=t, u=self.preintegrated_inputs(packet, u=u), Q=Q)

predict

predict(dt, *, t=None, u=None, Q=None)

Advance the estimate by dt. Process noise: an explicit Q, else self.Q, else the model's baked L Σ Lᵀ. u is {input: value} (unset inputs fall to the Module's declared defaults); pass the same held u truth ran on. t=None uses (and advances) the runtime's clock, matching NumpySim.step; an explicit t overrides and resynchronizes it.

Source code in manta/codegen/numpy/_filter.py

def predict(self, dt: float, *, t: float | None = None,
            u: dict[str, Any] | None = None,
            Q: np.ndarray | None = None) -> None:
    """Advance the estimate by `dt`. Process noise: an explicit `Q`, else
    `self.Q`, else the model's baked `L Σ Lᵀ`. `u` is `{input: value}`
    (unset inputs fall to the Module's declared defaults); pass the same
    held `u` truth ran on. `t=None` uses (and advances) the runtime's
    clock, matching `NumpySim.step`; an explicit `t` overrides and
    resynchronizes it."""
    dt = require_positive(dt, name=f"{type(self).__name__}.predict dt")
    t0 = self._t if t is None else float(require_finite(
        t, name=f"{type(self).__name__}.predict t"))
    next_t = float(require_finite(
        t0 + dt, name=f"{type(self).__name__}.predict resulting time"))
    process_Q = Q if Q is not None else self._Q
    if process_Q is not None:
        process_Q = self._validate_covariance(
            process_Q, who="predict Q", positive_definite=False)
    u_vec = self._kernel_u(u)
    self._check_packet_duration(dt, u_vec)
    self._predict_kernel(dt, t0, u_vec, process_Q)
    self._t = next_t

update

update(target, z=None, R=None, *, t=None, u=None)

Fold one measurement at the current state.

• update("gps.position", z) — by sensor name (full or suffix), through the baked covariance and gate.
• update("gps.position", z, R=sample_R) — typed per-sample device covariance through the deployable Module entry point; non-overrideable white model covariance remains additive, while static calibration uncertainty remains in the Schmidt recursion.
• update(h_sym, z, R=R) — a caller-supplied h(x) callable + measurement covariance (custom measurements; numpy-only).

t=None reads the runtime's clock (which predict advances); a measurement is dt-independent, so update never advances it.

Source code in manta/codegen/numpy/_filter.py

def update(self, target, z=None, R=None, *, t: float | None = None,
           u: dict[str, Any] | None = None) -> UpdateResult:
    """Fold one measurement at the current state.

    * `update("gps.position", z)` — by sensor name (full or suffix),
      through the baked covariance and gate.
    * `update("gps.position", z, R=sample_R)` — typed per-sample
      device covariance through the deployable Module entry point;
      non-overrideable white model covariance remains additive, while
      static calibration uncertainty remains in the Schmidt recursion.
    * `update(h_sym, z, R=R)` — a caller-supplied `h(x)` callable +
      measurement covariance (custom measurements; numpy-only).

    `t=None` reads the runtime's clock (which `predict` advances); a
    measurement is dt-independent, so `update` never advances it.
    """
    if callable(target):
        if z is None or R is None:
            raise TypeError("update(h_sym, z, R=...): z and R required")
        return self._update_custom(target, z, R)
    return self._fold_sensor(
        self._resolve_sensor(target), z, self._kernel_u(u), R=R,
        t=self._t if t is None else float(require_finite(
            t, name=f"{type(self).__name__}.update t")))

compile_sensor_updates

compile_sensor_updates(sensor_names, *, covariance='model', optimization='balanced', timeout_s=DEFAULT_COMPILATION_TIMEOUT_S, max_instructions=DEFAULT_MAX_INSTRUCTIONS)

Compile selected sensor-fold kernels through the public filter API.

covariance="model" selects the ordinary diagnostic update whose covariance is baked into the estimator. "per_sample" selects the update entry accepting an R= override. Sensor names use the same full-name/unambiguous-suffix rules as :meth:update; callers never need to construct or depend on generated Module entry names.

Source code in manta/codegen/numpy/_filter.py

def compile_sensor_updates(
    self,
    sensor_names: Iterable[str],
    *,
    covariance: Literal["model", "per_sample"] = "model",
    optimization: Optimization = "balanced",
    timeout_s: float = DEFAULT_COMPILATION_TIMEOUT_S,
    max_instructions: int | None = DEFAULT_MAX_INSTRUCTIONS,
) -> NumpyFilter:
    """Compile selected sensor-fold kernels through the public filter API.

    ``covariance="model"`` selects the ordinary diagnostic update whose
    covariance is baked into the estimator. ``"per_sample"`` selects the
    update entry accepting an ``R=`` override. Sensor names use the same
    full-name/unambiguous-suffix rules as :meth:`update`; callers never
    need to construct or depend on generated Module entry names.
    """
    if covariance not in {"model", "per_sample"}:
        raise ValueError("covariance must be 'model' or 'per_sample'")
    prefix = (
        "update_diagnostic_" if covariance == "model"
        else "update_with_R_"
    )
    requested = (sensor_names,) if isinstance(sensor_names, str) \
        else sensor_names
    full_names = tuple(dict.fromkeys(
        self._resolve_sensor(name) for name in requested
    ))
    function_names = tuple(
        self.module.entry(prefix + entry_ident(full)).fn
        for full in full_names
    )
    self.compile_functions(
        function_names,
        optimization=optimization,
        timeout_s=timeout_s,
        max_instructions=max_instructions,
    )
    return self

manta.FilterCheckpoint dataclass

FilterCheckpoint(x, P, time, artifact_id, P_consider=None)

Complete restart point for a filter runtime.

Arrays are owned snapshots rather than views into the live runtime. time is the filter's logical model time, not a wall clock.

manta.UpdateResult dataclass

UpdateResult(sensor, innovation, innovation_covariance, nis, accepted, gate, covariance_overridden)

Diagnostics and disposition of one measurement fold.

manta.FilterReplayProgram dataclass

FilterReplayProgram(kernel_identity, initial, operation_count, checkpoint_count, _kinds, _sensors, _times, _dts, _controls, _measurements, _measurement_covariances, _process_covariances, _use_measurement_covariance, _use_process_covariance, _checkpoint_flags)

Validated, bounded native input owned by one kernel identity.

packed_bytes property

packed_bytes

Owned numeric storage, useful for capacity accounting.

manta.FilterReplayResult dataclass

FilterReplayResult(final, updates, checkpoints)

Final state plus ordered per-update diagnostics and checkpoints.

manta.codegen.NumpyRegulator

NumpyRegulator(module)

Bases: NumpyRuntime

A stateless control law: map a state estimate to commands via control(estimate) -> {input: value}.

Holds the live operating point the law flies about — the reference x_ref plus every MATRIX-role coefficient the Module declares (an LQR's gain K and feed-forward u_ff), each seeded from its declared default. retarget() moves the reference alone; reprogram() installs a whole re-solved operating point.

Source code in manta/codegen/numpy/_regulator.py

def __init__(self, module) -> None:
    super().__init__(module)
    st = module.ports_by_role(Role.STATE)
    self._ref_port = st[1] if len(st) > 1 else None
    self._x_ref = (np.asarray(self._ref_port.init, dtype=float)
                   .reshape(-1).copy() if self._ref_port is not None
                   else np.asarray(self._x_port.init,
                                   dtype=float).reshape(-1).copy())
    # Law coefficients supplied as data (defaults = the built solve).
    self._coeffs = {p.name: np.asarray(p.init, dtype=float).copy()
                    for p in module.ports_by_role(Role.MATRIX)
                    if p.init is not None}

x_ref property

x_ref

The live reference point (flat ambient vector).

gain property

gain

The live feedback gain K (n_u × regulated tangent).

u_ff property

u_ff

The live feed-forward command (n_u).

retarget

retarget(state)

Move the reference point the law regulates to (nested or flat dict, merged over the CURRENT reference). The gain is NOT re-solved: exact wherever the dynamics are invariant along the moved direction (e.g. translating a hover setpoint), but NOT for a heading change — there the world-frame feedback rotates with the reference. Use reprogram(lqr.resolve_at(...)) for those.

Source code in manta/codegen/numpy/_regulator.py

def retarget(self, state: dict) -> None:
    """Move the reference point the law regulates to (nested or flat
    dict, merged over the CURRENT reference). The gain is NOT
    re-solved: exact wherever the dynamics are invariant along the
    moved direction (e.g. translating a hover setpoint), but NOT for
    a heading change — there the world-frame feedback rotates with
    the reference. Use `reprogram(lqr.resolve_at(...))` for those.
    """
    if self._ref_port is None:
        raise AttributeError(
            f"{type(self).__name__}: module {self.module.name!r} has "
            "no reference port — its control law is not retargetable.")
    self._x_ref = self._ref_port.spec.pack_any(state,
                                                   base=self._x_ref)

reprogram

reprogram(solution)

Install a re-solved operating point — gain, feed-forward, and reference at once (a gain is only valid about the point it was solved at, so they move together).

solution is anything carrying K, u_ff and x_ref — an LQRSolution from LQR.resolve_at, or a plain object/namespace rebuilt from JSON on the far side of a retarget service.

Source code in manta/codegen/numpy/_regulator.py

def reprogram(self, solution) -> None:
    """Install a re-solved operating point — gain, feed-forward, and
    reference at once (a gain is only valid about the point it was
    solved at, so they move together).

    `solution` is anything carrying `K`, `u_ff` and `x_ref` — an
    `LQRSolution` from `LQR.resolve_at`, or a plain object/namespace
    rebuilt from JSON on the far side of a retarget service.
    """
    from ...control.lqr import LQRSolution
    if not isinstance(solution, LQRSolution):
        raise TypeError(
            f"{type(self).__name__}.reprogram: expected LQRSolution")
    expected_controller = self.module.metadata.get("controller_id")
    if solution.controller_id != expected_controller:
        raise ValueError(
            f"{type(self).__name__}.reprogram: solution belongs to a "
            "different controller artifact")
    staged: dict[str, np.ndarray] = {}
    for name in ("K", "u_ff"):
        if name not in self._coeffs:
            raise AttributeError(
                f"{type(self).__name__}: module {self.module.name!r} "
                f"declares no {name!r} port — its control law is not "
                f"reprogrammable.")
        want = self._coeffs[name].shape
        raw = np.asarray(getattr(solution, name))
        if raw.dtype.kind not in "iuf" or raw.size != int(np.prod(want)):
            raise ValueError(
                f"{type(self).__name__}.reprogram: {name} must contain "
                f"{int(np.prod(want))} real values")
        arr = np.asarray(raw, dtype=float).reshape(want)
        if not np.all(np.isfinite(arr)):
            raise ValueError(
                f"{type(self).__name__}.reprogram: {name} must be finite")
        staged[name] = arr.copy()
    next_ref = self._ref_port.spec.pack_any(
        np.asarray(solution.x_ref, dtype=float).reshape(-1))
    self._coeffs.update(staged)
    self._x_ref = next_ref

u

u(x_flat)

Control vector for a flat ambient state (full-spec layout).

Source code in manta/codegen/numpy/_regulator.py

def u(self, x_flat) -> np.ndarray:
    """Control vector for a flat ambient state (full-spec layout)."""
    vals = {"x": np.asarray(x_flat, dtype=float)}
    if self._ref_port is not None:
        vals[self._ref_port.name] = self._x_ref
    vals.update(self._coeffs)
    return self.call("control", vals)["u"]

control

control(state)

Map a state estimate (nested or flat dict) → {input: value}, merged over the live reference point (unsupplied slots sit at the reference, i.e. zero error).

Source code in manta/codegen/numpy/_regulator.py

def control(self, state: dict) -> dict[str, Any]:
    """Map a state estimate (nested or flat dict) → `{input: value}`,
    merged over the live reference point (unsupplied slots sit at
    the reference, i.e. zero error)."""
    x = self._x_port.spec.pack_any(state, base=self._x_ref)
    return unpack_fields(self._u_fields(), self.u(x))

manta.codegen.NumpyRecurrence

NumpyRecurrence(module)

Bases: NumpyRuntime

A stateful dataflow block (PID, Madgwick, …): step(dt, **inputs) advances the held state and computes the readouts.

Source code in manta/codegen/numpy/_recurrence.py

def __init__(self, module) -> None:
    super().__init__(module)
    self._y = np.zeros(self._y_port.size)

state property

state

Held state, {slot: value} by name.

reset

reset()

Reset the held state to the Module's declared initial values.

Source code in manta/codegen/numpy/_recurrence.py

def reset(self) -> None:
    """Reset the held state to the Module's declared initial values."""
    x_field = self.module.state.field("x")
    self._state["x"] = np.asarray(
        x_field.init, dtype=float).reshape(-1).copy()
    self._y = np.zeros(self._y_port.size)
    self._t = 0.0

readouts

readouts()

Last-computed readouts by output-field name (scalars unwrapped).

Source code in manta/codegen/numpy/_recurrence.py

def readouts(self) -> dict[str, Any]:
    """Last-computed readouts by output-field name (scalars unwrapped)."""
    return unpack_fields(self._y_port.fields, self._y)