System identification

Fit a model's physical parameters (Fit) or its noise model (NoiseFit) to recorded data. See Fit parameters from a log for a worked recipe.

Parameter fit

manta.Fit

Fit(world, parameters, *, priors=())

MAP parameter fit over recorded windows (see module docstring).

Args: world — the model. Finalized by the internal Sim; the fit never mutates it (until result.apply()). parameters — {name: Prior | Tied | Free | None}. Prior/None keys resolve against the model's promotable Parameters (<craft>.<part>.<param>); None = flat prior (only safe for parameters the data fully observes). A Tied key is promoted but derives from another entry's decision variable; a Free key is a fresh auxiliary name (not a model parameter) that exists to source ties.

Source code in manta/fit/_map.py

def __init__(self, world, parameters: dict, *,
             priors: tuple[GaussianTangentPrior, ...] | list[
                 GaussianTangentPrior] = ()) -> None:
    for k, spec in parameters.items():
        if spec is not None and not isinstance(spec, (Prior, Tied, Free)):
            raise TypeError(
                f"Fit: parameters[{k!r}] must be Prior, Tied, Free, or "
                f"None, got {type(spec).__name__}.")
    free_specs = {k: v for k, v in parameters.items()
                  if isinstance(v, Free)}
    promoted_keys = [k for k in parameters if k not in free_specs]
    if not promoted_keys:
        raise ValueError(
            "Fit: no promotable parameters named — Free variables "
            "alone fit nothing.")

    # A ModelArtifact is an immutable executable revision, not an
    # authoring surface.  Fit against an editable copy so derive/apply
    # semantics remain coherent without mutating the source artifact.
    source = world
    self.world = (world.world_copy()
                  if isinstance(world, ModelArtifact) else world)
    self.sim = Sim(source, parameters=promoted_keys)
    self.model_world = self.sim.world
    # Fitting owns observation availability through Window.z_mask. It
    # needs every declared measurement expression in the batched step;
    # the runtime acquisition scheduler is deliberately irrelevant here.
    self.module = self.sim.inline_module()
    self._spec = self.module.spec
    port = self.module.port("params")
    fulls = [f.name for f in port.fields]
    by_full = {resolve_suffix(k, fulls, label="parameter", who="Fit"): v
               for k, v in parameters.items() if k not in free_specs}
    self._fields = [(f.name, f.dim) for f in port.fields]
    manifolds = {
        parameter.full: parameter.owner
        .promotable_parameter_declarations()[parameter.name]
        .manifold
        for parameter in self.sim.sys.param_specs
    }

    # Decision blocks: one per non-tied promoted parameter (in port
    # order == kernel `p` layout), then one per Free variable.
    self._blocks: list[_Block] = []
    self._block_by_name: dict[str, _Block] = {}
    off = 0
    for f in port.fields:
        if isinstance(by_full.get(f.name), Tied):
            continue
        blk = _Block(f.name, f.dim, off,
                     np.asarray(f.default, dtype=float).ravel(),
                     by_full.get(f.name), manifold=manifolds[f.name])
        self._blocks.append(blk)
        self._block_by_name[f.name] = blk
        off += blk.dim
    for name, fr in free_specs.items():
        if name in {f.name for f in port.fields}:
            raise ValueError(
                f"Fit: Free name {name!r} collides with a promoted "
                f"parameter of the same name.")
        init = np.atleast_1d(np.asarray(fr.init, dtype=float)).ravel()
        blk = _Block(name, init.size, off, init, fr.prior)
        self._blocks.append(blk)
        self._block_by_name[name] = blk
        off += init.size
    self.n_v = off

    self._tangent_priors = self._resolve_tangent_priors(priors)

    # Resolve ties: target field → (source block, A, b), ambient.
    sources = list(self._block_by_name)
    self._ties: dict[str, tuple] = {}
    for f in port.fields:
        spec = by_full.get(f.name)
        if not isinstance(spec, Tied):
            continue
        try:
            src_full = resolve_suffix(spec.source, sources,
                                      label="tie source", who="Fit")
        except KeyError:
            tied_names = [n for n, s in by_full.items()
                          if isinstance(s, Tied)]
            try:
                resolve_suffix(spec.source, tied_names,
                               label="tie source", who="Fit")
            except KeyError:
                raise KeyError(
                    f"Fit: Tied {f.name!r}: unknown source "
                    f"{spec.source!r}. Available: {sorted(sources)}")
            raise ValueError(
                f"Fit: Tied {f.name!r}: source {spec.source!r} is "
                f"itself tied — chains are not supported; tie every "
                f"copy to the same free source.")
        src = self._block_by_name[src_full]
        A, b = _tie_map(
            spec, f.dim, src.ambient_dim, target=f.name
        )
        self._ties[f.name] = (src, A, b)
    self._stepk_cache: dict[int, ca.Function] = {}

solve

solve(windows, *, weights=None, state_weights=None, window_weights=None, state_robust_delta=None, initial_values=None, compute_posterior=True, solver='ipopt', least_squares_options=None, verbose=False, progress=None, posterior_progress=None, ipopt_options=None)

Build the windowed prediction-error + prior NLP and solve it.

Args: windows — the recorded data (≥ 1 Window). weights — optional per-sensor scalar weights on the squared residuals ({sensor name/suffix: w}); use 1/σ_meas² to whiten mixed-unit sensors. Default 1. state_weights — optional per-state-slot weights on tangent-space trajectory residuals. Default 1 for every recorded slot. window_weights — optional positive scalar per window. This is the dataset-composition boundary for independently normalized real and synthetic groups; default 1 for every window. state_robust_delta — optional positive pseudo-Huber transition in normalized trajectory-RMS units. It applies to state slots carrying Window.x_scale and limits the influence of one structurally unrepresentable rollout without introducing a non-differentiable clipping point. initial_values — optional warm start for decision parameters in ambient units. Keys use the same exact-or-unique-suffix resolution as parameter declarations. Priors and bounds are unchanged; only IPOPT's starting iterate moves. compute_posterior — build the full residual Jacobian used only for identifiability diagnostics. Disable on large production fits to save peak memory; fitted values are unchanged. solver — "ipopt" or "gauss-newton". The latter exploits the residual structure without constructing symbolic second derivatives and is preferred for large unrolled plant fits. least_squares_options — damped Gauss-Newton tolerances, iteration limit, and damping options. Used only by that solver. verbose — IPOPT iteration output. progress — optional callback after every accepted IPOPT iteration. It receives the current and retained-best objectives plus checkpoint-safe ambient parameter values. Return False to stop cleanly at that boundary. This is independent of verbose and cannot be combined with a raw CasADi iteration_callback option. posterior_progress — optional callback after each residual block contributes to the posterior normal matrix. ipopt_options — extra nlpsol options, merged last.

Source code in manta/fit/_map.py

def solve(self, windows: list[Window], *, weights: dict | None = None,
          state_weights: dict | None = None,
          window_weights: list[float] | tuple[float, ...] | None = None,
          state_robust_delta: float | None = None,
          initial_values: dict | None = None,
          compute_posterior: bool = True,
          solver: str = "ipopt",
          least_squares_options: dict | None = None,
          verbose: bool = False,
          progress: Callable[[FitProgress], bool | None] | None = None,
          posterior_progress: Callable[[FitPosteriorProgress], None]
          | None = None,
          ipopt_options: dict | None = None) -> FitResult:
    """Build the windowed prediction-error + prior NLP and solve it.

    Args:
        windows — the recorded data (≥ 1 `Window`).
        weights — optional per-sensor scalar weights on the squared
                  residuals (`{sensor name/suffix: w}`); use
                  `1/σ_meas²` to whiten mixed-unit sensors. Default 1.
        state_weights — optional per-state-slot weights on tangent-space
                  trajectory residuals. Default 1 for every recorded slot.
        window_weights — optional positive scalar per window. This is the
                  dataset-composition boundary for independently normalized
                  real and synthetic groups; default 1 for every window.
        state_robust_delta — optional positive pseudo-Huber transition in
                  normalized trajectory-RMS units. It applies to state
                  slots carrying ``Window.x_scale`` and limits the
                  influence of one structurally unrepresentable rollout
                  without introducing a non-differentiable clipping point.
        initial_values — optional warm start for decision parameters in
                  ambient units. Keys use the same exact-or-unique-suffix
                  resolution as parameter declarations. Priors and bounds
                  are unchanged; only IPOPT's starting iterate moves.
        compute_posterior — build the full residual Jacobian used only for
                  identifiability diagnostics. Disable on large production
                  fits to save peak memory; fitted values are unchanged.
        solver — ``"ipopt"`` or ``"gauss-newton"``. The latter exploits
                  the residual structure without constructing symbolic
                  second derivatives and is preferred for large unrolled
                  plant fits.
        least_squares_options — damped Gauss-Newton tolerances, iteration
                  limit, and damping options. Used only by that solver.
        verbose — IPOPT iteration output.
        progress — optional callback after every accepted IPOPT iteration.
                  It receives the current and retained-best objectives plus
                  checkpoint-safe ambient parameter values. Return False
                  to stop cleanly at that boundary. This is independent of
                  ``verbose`` and cannot be combined with a raw CasADi
                  ``iteration_callback`` option.
        posterior_progress — optional callback after each residual block
                  contributes to the posterior normal matrix.
        ipopt_options — extra `nlpsol` options, merged last.
    """
    if not windows:
        raise ValueError("Fit.solve: needs at least one Window.")
    if solver not in ("ipopt", "gauss-newton"):
        raise ValueError("Fit.solve: solver must be 'ipopt' or 'gauss-newton'")
    if solver == "ipopt" and least_squares_options:
        raise ValueError("least_squares_options require solver='gauss-newton'")
    if solver == "gauss-newton" and ipopt_options:
        raise ValueError("ipopt_options require solver='ipopt'")
    if (state_robust_delta is not None
            and (not np.isfinite(state_robust_delta)
                 or state_robust_delta <= 0.0)):
        raise ValueError("state_robust_delta must be positive and finite")
    if window_weights is None:
        resolved_window_weights = np.ones(len(windows), dtype=float)
    else:
        resolved_window_weights = np.asarray(
            window_weights, dtype=float
        ).ravel()
        if (
            resolved_window_weights.shape != (len(windows),)
            or not np.all(np.isfinite(resolved_window_weights))
            or np.any(resolved_window_weights <= 0.0)
        ):
            raise ValueError(
                "window_weights must contain one positive finite value "
                "per window"
            )

    latent_blocks, latent_by_window = self._initial_state_blocks(windows)
    decision_blocks = [*self._blocks, *latent_blocks]
    n_decision = sum(block.dim for block in decision_blocks)
    v = ca.MX.sym("v", n_decision, 1)
    p = self._p_of_v(v)

    meas_names = [pt.name for pt in
                  self.module.ports_by_role(Role.MEASUREMENT)]
    w_by_full: dict[str, float] = {}
    for k, val in (weights or {}).items():
        full = resolve_suffix(k, meas_names, label="sensor", who="Fit")
        w_by_full[full] = float(val)
    state_names = [slot.name for slot in self._spec.slots]
    wx_by_full: dict[str, float] = {}
    for k, val in (state_weights or {}).items():
        full = resolve_suffix(k, state_names, label="state slot", who="Fit")
        wx_by_full[full] = float(val)

    loss = ca.MX(0.0)
    residuals: list[ca.MX] = []
    for w, selected, window_weight in zip(
        windows, latent_by_window, resolved_window_weights, strict=True
    ):
        loss, residuals = self._add_window(
            w, p, loss, residuals, w_by_full, wx_by_full, meas_names,
            state_names, state_robust_delta,
            x0=self._window_initial_state(w, selected, v),
            window_weight=float(window_weight))

    # MAP prior term (skipped for flat-prior components).
    loss = loss + prior_penalty(v, decision_blocks)
    loss = loss + self._tangent_prior_penalty(v)
    initial_v = np.concatenate([block.init for block in decision_blocks])
    block_names = [block.full for block in self._blocks]
    for key, value in (initial_values or {}).items():
        full = resolve_suffix(
            key, block_names, label="warm-start parameter", who="Fit")
        block = self._block_by_name[full]
        ambient = np.atleast_1d(np.asarray(value, dtype=float)).ravel()
        decision = block.v_from_theta(ambient)
        if (np.any(decision < block.lower)
                or np.any(decision > block.upper)):
            raise ValueError(
                f"Fit: warm start for {full!r} violates its bounds.")
        initial_v[block.offset:block.offset + block.dim] = decision
    initial_objective = float(ca.DM(
        ca.Function("fit_initial_loss", [v], [loss])(initial_v)))

    p_fn = ca.Function("p", [v], [p])
    promoted = {full for full, _dim in self._fields}

    def parameter_values(decision: np.ndarray) -> dict[str, object]:
        packed = np.asarray(ca.DM(p_fn(decision))).ravel()
        result: dict[str, object] = {}
        offset = 0
        for full, dim in self._fields:
            value = packed[offset:offset + dim].copy()
            result[full] = float(value[0]) if dim == 1 else value
            offset += dim
        for block in self._blocks:
            if block.full in promoted:
                continue
            value = block.theta_of_v(
                decision[block.offset:block.offset + block.dim]
            ).copy()
            result[block.full] = (
                float(value[0]) if block.dim == 1 else value
            )
        return result

    def emit_progress(
        iteration: int,
        _current_v: np.ndarray,
        objective: float,
        best_v: np.ndarray,
        best_objective: float,
        callback_initial_objective: float,
    ) -> bool | None:
        if progress is None:
            return None
        return progress(FitProgress(
            iteration=iteration,
            objective=float(objective),
            best_objective=float(best_objective),
            initial_objective=float(callback_initial_objective),
            values=parameter_values(best_v),
        ))

    if solver == "gauss-newton":
        complete_residuals = [
            *residuals,
            *prior_residuals(v, decision_blocks),
            *self._tangent_prior_residuals(v),
        ]
        complete_residual = (
            ca.vertcat(*complete_residuals)
            if complete_residuals
            else ca.MX.zeros(0, 1)
        )
        v_opt, objective, stats, expanded = solve_blocks_least_squares(
            "fit",
            v,
            complete_residual,
            decision_blocks,
            initial=initial_v,
            progress=emit_progress if progress is not None else None,
            options=least_squares_options,
        )
    else:
        v_opt, objective, stats, expanded = solve_blocks_nlp(
            "fit", v, loss, decision_blocks,
            verbose=verbose, ipopt_options=ipopt_options,
            initial=initial_v, retain_best=True,
            progress=emit_progress if progress is not None else None)

    # Data-only Gauss-Newton information JᵀJ is optional. Evaluate one
    # residual block at a time: concatenating every window/channel into a
    # monolithic dense Jacobian caused posterior computation to dominate
    # memory even when only a six-dimensional mount block was requested.
    # The accumulated normal matrix is algebraically identical.
    if compute_posterior:
        JtJ = np.zeros((n_decision, n_decision))
        for index, residual in enumerate(residuals):
            J_fn = ca.Function(
                f"J_block_{index}", [v], [ca.jacobian(residual, v)]
            )
            J_fn = expand_or_none(J_fn) or J_fn
            J = np.asarray(ca.DM(J_fn(v_opt)))
            JtJ += J.T @ J
            if posterior_progress is not None:
                posterior_progress(FitPosteriorProgress(
                    completed_blocks=index + 1,
                    total_blocks=len(residuals),
                ))
    else:
        JtJ = np.zeros((n_decision, n_decision))

    p_opt = np.asarray(ca.DM(p_fn(v_opt))).ravel()
    tie_sources = {full: src.full
                   for full, (src, _A, _b) in self._ties.items()}
    res = FitResult(self._blocks, self._fields, tie_sources, v_opt,
                    p_opt, JtJ, objective, stats, self.world,
                    self.sim.model.model_id,
                    self.sim.model.artifact_id,
                    self.sim.model.derivation,
                    posterior_computed=compute_posterior,
                    initial_objective=initial_objective,
                    tangent_prior_information=(
                        self._tangent_prior_information(n_decision)
                    ),
                    latent_blocks=latent_blocks)
    res.expanded = expanded
    # Identity of the training set: `evidence()` refuses any of these
    # as a held-out window (the acceptance set must be untouched).
    res._training_digests = tuple(window_digest(w) for w in windows)
    u_fields = self.module.port("u").fields
    res._training_default_fills = tuple(sorted((
        fill
        for window, digest in zip(
            windows, res._training_digests, strict=True
        )
        for fill in default_fills_for_window(
            self.model_world,
            self._spec,
            window,
            dataset_role="training",
            window_digest=digest,
            input_names=[field.name for field in u_fields],
            input_defaults=[field.default for field in u_fields],
            input_fields=u_fields,
        )
    ), key=lambda fill: (
        fill.dataset_role, fill.window_digest, fill.source, fill.name
    )))
    return res

sensor_residuals

sensor_residuals(result, windows)

Replay fitted mean predictions and return raw sensor residuals.

This is the conditional residual boundary used by inexpensive sensor noise characterization after a mean fit. It reuses the same compiled step/mapaccum functions and window conventions as :meth:solve; callers do not need to reconstruct simulation ordering themselves.

Source code in manta/fit/_map.py

def sensor_residuals(
    self,
    result: FitResult,
    windows: list[Window] | tuple[Window, ...],
) -> dict[str, np.ndarray]:
    """Replay fitted mean predictions and return raw sensor residuals.

    This is the conditional residual boundary used by inexpensive sensor
    noise characterization after a mean fit. It reuses the same compiled
    step/mapaccum functions and window conventions as :meth:`solve`;
    callers do not need to reconstruct simulation ordering themselves.
    """
    if not isinstance(result, FitResult):
        raise TypeError("sensor_residuals requires a FitResult")
    if result._source_artifact_id != self.sim.model.artifact_id:
        raise ValueError("FitResult was produced by a different model artifact")
    if not result.converged:
        raise RuntimeError("sensor residuals require a converged mean fit")
    measurement_names = [
        port.name for port in self.module.ports_by_role(Role.MEASUREMENT)
    ]
    dimensions = {
        name: self.module.port(name).size for name in measurement_names
    }
    entry = self.module.entry("step")
    input_fields = self.module.port("u").fields
    noise_size = self.module.port("noise").size
    collected: dict[str, list[np.ndarray]] = {}

    for window_index, window in enumerate(windows):
        if not window.z:
            continue
        observed, steps = resolve_traces(
            window.z,
            measurement_names,
            dimensions,
            who="Fit.sensor_residuals",
        )
        masks = resolve_trace_masks(
            window.z_mask,
            observed,
            measurement_names,
            steps,
            who="Fit.sensor_residuals",
        )
        controls = pack_u_trace(
            window.u,
            [field.name for field in input_fields],
            [float(np.asarray(field.default).ravel()[0]) for field in input_fields],
            steps,
            who="Fit.sensor_residuals",
        )
        initial = np.asarray(pack_x0(self.model_world, self._spec, window))
        tangent = np.zeros(self._spec.tangent_dim, dtype=float)
        for (index, slot_name), delta in result.window_initial_state_deltas.items():
            if index != window_index:
                continue
            slot = self._spec.slot(slot_name)
            tangent[
                slot.tangent_offset:slot.tangent_offset + slot.tangent_dim
            ] = np.asarray(delta, dtype=float).ravel()
        if np.any(tangent):
            initial = np.asarray(ca.DM(
                self._spec.boxplus_sym(ca.DM(initial), ca.DM(tangent))
            )).reshape(-1, 1)
        else:
            initial = initial.reshape(-1, 1)
        arguments = {
            "x": ca.DM(initial),
            "u": ca.DM(controls) if controls.size else ca.DM(0, steps),
            "noise": (
                ca.DM.zeros(noise_size, steps)
                if noise_size
                else ca.DM(0, steps)
            ),
            "params": ca.repmat(ca.DM(result._p), 1, steps),
            "dt": ca.repmat(ca.DM(float(window.dt)), 1, steps),
            "t": ca.DM(np.array([[
                window.t0 + index * window.dt for index in range(steps)
            ]])),
        }
        ordered = [
            arguments[arg.name if isinstance(arg, PortRef) else "x"]
            for arg in entry.args
        ]
        raw_outputs = self._stepk(steps)(*ordered)
        outputs = (
            list(raw_outputs)
            if isinstance(raw_outputs, (list, tuple))
            else [raw_outputs]
        )
        for full, values in observed.items():
            selected = np.flatnonzero(masks[full])
            if not selected.size:
                continue
            output_index = 1 + entry.returns.index(full)
            predicted = np.asarray(outputs[output_index], dtype=float).T
            collected.setdefault(full, []).append(
                predicted[selected] - np.asarray(values, dtype=float)[selected]
            )

    return {
        name: np.concatenate(chunks, axis=0)
        for name, chunks in collected.items()
    }

manta.FitResult

FitResult(blocks, fields, tie_sources, v_opt, p_opt, JtJ, objective, stats, world, source_model_id, source_artifact_id, source_derivation, *, posterior_computed, initial_objective, tangent_prior_information=None, latent_blocks=())

Fitted values + Gauss-Newton posterior diagnostics.

Attributes: values — {name: fitted value} (float for scalars, ndarray for vectors) — every promoted parameter (tied ones derived through their affine map) plus every Free variable. labels — one entry per fitted scalar component of the DECISION vector (tied parameters don't appear; their source does). log_scale — per-component bool; True ⇒ the sigmas below are RELATIVE (log-space). prior_sigma — per-component prior σ (inf = no prior). posterior_sigma — per-component Gauss-Newton posterior σ from (JᵀJ + Σ₀⁻¹)⁻¹. ≈ prior σ ⇒ the data did not inform this component. JtJ — data-only Gauss-Newton information matrix in decision space; its small eigenvalues are the unidentifiable directions. objective — final loss value. stats — IPOPT return stats. converged — IPOPT's success flag; False ⇒ the values below are the failed solve's final iterate (a RuntimeWarning was emitted), not an optimum. expanded — True when the NLP ran SX-expanded; False means the loss graph kept a Linsol node and IPOPT evaluated the (order-of-magnitude slower) interpreted MX graph (a RuntimeWarning said so at solve time).

Source code in manta/fit/_map.py

def __init__(self, blocks, fields, tie_sources, v_opt, p_opt, JtJ,
             objective, stats, world, source_model_id, source_artifact_id,
             source_derivation, *, posterior_computed: bool,
             initial_objective: float,
             tangent_prior_information: np.ndarray | None = None,
             latent_blocks=()) -> None:
    self._blocks = blocks
    self._latent_blocks = tuple(latent_blocks)
    self._diagnostic_blocks = (*blocks, *self._latent_blocks)
    self._fields = fields              # [(full, dim)] in port order
    self._tie_sources = tie_sources    # {tied full: source name}
    self._world = world
    self._source_model_id = source_model_id
    self._source_artifact_id = source_artifact_id
    self._source_derivation = dict(source_derivation)
    self.v = np.asarray(v_opt, dtype=float).ravel()
    self._p = np.asarray(p_opt, dtype=float).ravel()
    self.JtJ = JtJ
    self.objective = float(objective)
    self.stats = stats
    self.posterior_computed = posterior_computed
    self.initial_objective = float(initial_objective)
    iteration_objectives = stats.get("iterations", {}).get("obj", ())
    self.objective_history = tuple(
        float(value) for value in iteration_objectives)
    if (not self.objective_history
            or not np.isclose(self.objective_history[0],
                              self.initial_objective)):
        self.objective_history = (self.initial_objective,
                                  *self.objective_history)
    if (not self.objective_history
            or not np.isclose(self.objective_history[-1], self.objective)):
        # A limited solve can terminate away from an earlier, better
        # accepted iterate. The fitter restores that retained incumbent.
        self.objective_history = (*self.objective_history, self.objective)
    self.converged = solver_converged(stats, who="Fit")

    # Every promoted parameter's ambient value (tied ones included),
    # sliced off the assembled parameter vector…
    self.values: dict[str, object] = {}
    off = 0
    for full, dim in fields:
        theta = self._p[off:off + dim]
        self.values[full] = float(theta[0]) if dim == 1 else theta
        off += dim
    # …plus the Free variables (decision-only, not in the port).
    promoted = {full for full, _ in fields}
    self.labels: list[str] = []
    self.log_scale: list[bool] = []
    prior_sig = []
    for b in blocks:
        theta = b.theta_of_v(self.v[b.offset:b.offset + b.dim])
        if b.full not in promoted:
            self.values[b.full] = float(theta[0]) if b.dim == 1 else theta
        self.labels += b.labels()
        self.log_scale += [b.log] * b.dim
        prior_sig.append(b.sigma)
    self.window_initial_state_deltas = {
        (block.window_index, block.slot_name): block.theta_of_v(
            self.v[block.offset:block.offset + block.dim]
        )
        for block in self._latent_blocks
    }
    for block in self._latent_blocks:
        self.labels += block.labels()
        self.log_scale += [False] * block.dim
        prior_sig.append(block.sigma)
    self.prior_sigma = np.concatenate(prior_sig)

    prior_prec = np.where(np.isinf(self.prior_sigma), 0.0,
                          1.0 / np.square(self.prior_sigma))
    self.prior_information = np.diag(prior_prec)
    if tangent_prior_information is not None:
        extra = np.asarray(tangent_prior_information, dtype=float)
        if extra.shape != self.prior_information.shape:
            raise ValueError("tangent prior information shape mismatch")
        self.prior_information += extra
    # eigh-based: flat-prior components the data never touched come
    # back inf, without poisoning the identified ones.
    self.posterior_sigma = (
        laplace_sigma(JtJ + self.prior_information)
        if posterior_computed else np.full_like(self.prior_sigma, np.nan))
    self.parameter_component_count = sum(block.dim for block in blocks)
    self.parameter_labels = tuple(
        self.labels[:self.parameter_component_count]
    )
    self.parameter_posterior_ratio = tuple(
        float(post / prior)
        if np.isfinite(prior) and prior > 0.0 else (
            0.0 if np.isfinite(post) else float("inf")
        )
        for prior, post in zip(
            self.prior_sigma[:self.parameter_component_count],
            self.posterior_sigma[:self.parameter_component_count],
            strict=True,
        )
    )
    active_bounds = []
    for block in blocks:
        decision = self.v[block.offset:block.offset + block.dim]
        for index, (value, lower, upper) in enumerate(zip(
            decision, block.lower, block.upper, strict=True
        )):
            scale = max(1.0, abs(float(value)))
            if (
                np.isfinite(lower) and value - lower <= 1e-6 * scale
                or np.isfinite(upper) and upper - value <= 1e-6 * scale
            ):
                active_bounds.append(block.labels()[index])
    self.active_parameter_bounds = tuple(active_bounds)
    self._profile_id = sha256(
        b"manta-parameter-fit-profile-v1\0" + canonical_derivation_bytes({
            "source_model_id": self._source_model_id,
            "source_artifact_id": self._source_artifact_id,
            "objective": self.objective,
            "values": self.values,
        })).hexdigest()

weak_directions

weak_directions(k=3)

The k least-informed directions of the DATA alone: list of (eigenvalue, {label: component}) for the smallest eigenvalues of JᵀJ. A near-zero eigenvalue is an unidentifiable parameter combination (e.g. the thrust/mass scale).

Source code in manta/fit/_map.py

def weak_directions(self, k: int = 3):
    """The `k` least-informed directions of the DATA alone: list of
    `(eigenvalue, {label: component})` for the smallest eigenvalues
    of JᵀJ. A near-zero eigenvalue is an unidentifiable parameter
    combination (e.g. the thrust/mass scale)."""
    vals, vecs = np.linalg.eigh(self.JtJ)
    out = []
    for i in range(min(k, len(vals))):
        comp = {lbl: float(vecs[j, i])
                for j, lbl in enumerate(self.labels)
                if abs(vecs[j, i]) > 1e-3}
        out.append((float(vals[i]), comp))
    return out

linear_contrast_posterior_sigma

linear_contrast_posterior_sigma(coefficients)

Posterior sigma of a local linear parameter combination.

coefficients names decision-space component labels from parameter_labels. The covariance is marginalized over window-local initial states, so this can distinguish a well-constrained relative quantity (for example one mount offset minus another) from two weak absolute parameters. A contrast touching an information-null direction reports infinity rather than false confidence.

This is a local tangent-space diagnostic. SO(3) labels therefore use the .delta[i] components shown by :meth:summary.

Source code in manta/fit/_map.py

def linear_contrast_posterior_sigma(
    self, coefficients: dict[str, float]
) -> float:
    """Posterior sigma of a local linear parameter combination.

    ``coefficients`` names decision-space component labels from
    ``parameter_labels``. The covariance is marginalized over window-local
    initial states, so this can distinguish a well-constrained relative
    quantity (for example one mount offset minus another) from two weak
    absolute parameters. A contrast touching an information-null
    direction reports infinity rather than false confidence.

    This is a local tangent-space diagnostic. SO(3) labels therefore use
    the ``.delta[i]`` components shown by :meth:`summary`.
    """
    if not self.posterior_computed:
        raise RuntimeError(
            "linear contrast requires compute_posterior=True"
        )
    if not coefficients:
        raise ValueError("linear contrast needs at least one coefficient")
    indices = {label: index for index, label in enumerate(self.labels)}
    unknown = set(coefficients) - set(self.parameter_labels)
    if unknown:
        raise KeyError(
            f"linear contrast has unknown parameter labels {sorted(unknown)}"
        )
    contrast = np.zeros(len(self.labels), dtype=float)
    for label, coefficient in coefficients.items():
        value = float(coefficient)
        if not np.isfinite(value):
            raise ValueError("linear contrast coefficients must be finite")
        contrast[indices[label]] = value

    information = self.JtJ + self.prior_information
    try:
        values, vectors = np.linalg.eigh(
            0.5 * (information + information.T)
        )
    except np.linalg.LinAlgError:
        return float("inf")
    largest = float(values[-1]) if len(values) else 0.0
    identified = values > max(largest, 0.0) * 1e-12
    projections = vectors.T @ contrast
    if np.any(np.abs(projections[~identified]) > 1e-12):
        return float("inf")
    variance = float(np.sum(np.square(projections[identified]) / values[identified]))
    return math.sqrt(max(variance, 0.0))

parameter_posterior_covariance

parameter_posterior_covariance(labels)

Marginal posterior covariance for selected parameter tangents.

The returned block is taken from the inverse joint information over all fitted parameters and window-local initial states, so nuisance variables are marginalized rather than held fixed. A requested coordinate touching an information-null direction is refused instead of receiving a spuriously finite pseudoinverse covariance.

Source code in manta/fit/_map.py

def parameter_posterior_covariance(
    self, labels: list[str] | tuple[str, ...]
) -> np.ndarray:
    """Marginal posterior covariance for selected parameter tangents.

    The returned block is taken from the inverse joint information over
    all fitted parameters and window-local initial states, so nuisance
    variables are marginalized rather than held fixed.  A requested
    coordinate touching an information-null direction is refused instead
    of receiving a spuriously finite pseudoinverse covariance.
    """
    if not self.posterior_computed:
        raise RuntimeError(
            "parameter covariance requires compute_posterior=True"
        )
    selected = tuple(labels)
    if not selected or len(selected) != len(set(selected)):
        raise ValueError(
            "parameter covariance needs unique selected labels"
        )
    index_by_label = {
        label: index for index, label in enumerate(self.labels)
    }
    unknown = set(selected) - set(self.parameter_labels)
    if unknown:
        raise KeyError(
            f"parameter covariance has unknown parameter labels "
            f"{sorted(unknown)}"
        )
    indices = np.asarray(
        [index_by_label[label] for label in selected], dtype=int
    )
    information = self.JtJ + self.prior_information
    try:
        values, vectors = np.linalg.eigh(
            0.5 * (information + information.T)
        )
    except np.linalg.LinAlgError as exc:
        raise ValueError(
            "posterior information eigendecomposition failed"
        ) from exc
    largest = float(values[-1]) if len(values) else 0.0
    identified = values > max(largest, 0.0) * 1e-12
    if np.any(np.abs(vectors[indices][:, ~identified]) > 1e-12):
        raise ValueError(
            "selected parameter covariance touches an unidentified "
            "posterior direction"
        )
    selected_vectors = vectors[indices][:, identified]
    covariance = (
        (selected_vectors / values[identified]) @ selected_vectors.T
        if np.any(identified)
        else np.zeros((len(indices), len(indices)), dtype=float)
    )
    return np.asarray(
        0.5 * (covariance + covariance.T), dtype=float
    )

apply

apply()

Write the fitted values — tied parameters derived through their affine maps — back onto the world's Part instances. A transform built afterwards (Sim(world), EKF(world), a C++ deploy) bakes them in as constants.

Source code in manta/fit/_map.py

def apply(self) -> None:
    """Write the fitted values — tied parameters derived through
    their affine maps — back onto the world's Part instances. A
    transform built afterwards (`Sim(world)`, `EKF(world)`, a C++
    deploy) bakes them in as constants."""
    if not self.converged:
        raise RuntimeError(
            "FitResult.apply refuses to write an unconverged solve")
    staged = self._staged_updates(self._world)
    for part, pname, value in staged:
        setattr(part, pname, value)

fitted_world

fitted_world()

An editable copy of the authoring world with the fitted values (tied parameters derived) written in — what derive() freezes and evidence() predicts with. Refuses an unconverged solve.

Source code in manta/fit/_map.py

def fitted_world(self):
    """An editable copy of the authoring world with the fitted values
    (tied parameters derived) written in — what `derive()` freezes and
    `evidence()` predicts with. Refuses an unconverged solve."""
    if not self.converged:
        raise RuntimeError(
            "FitResult.fitted_world refuses an unconverged solve")
    derived = copy.deepcopy(self._world)
    for part, name, value in self._staged_updates(derived):
        setattr(part, name, value)
    return derived

evidence

evidence(held_out, *, sensor, criteria=None, lag_count=20, selection=(), configuration_id=None, channel_contract_id=None)

Held-out evidence for the fitted model (see held_out_evidence).

held_out must be untouched by the fit: any window whose content matches a training window is refused. The result is what derive(evidence=...) attaches and what a ModelForce consumes.

Source code in manta/fit/_map.py

def evidence(self, held_out: list[Window], *, sensor: str,
             criteria: FitAcceptanceCriteria | None = None,
             lag_count: int = 20,
             selection: list[Window] = (),
             configuration_id: str | None = None,
             channel_contract_id: str | None = None) -> FitEvidence:
    """Held-out evidence for the fitted model (see `held_out_evidence`).

    ``held_out`` must be untouched by the fit: any window whose content
    matches a training window is refused. The result is what
    `derive(evidence=...)` attaches and what a `ModelForce` consumes.
    """
    candidate = self._candidate_artifact()
    candidate_sim = Sim(candidate)
    candidate_module = candidate_sim.module()
    candidate_u_fields = candidate_module.port("u").fields
    selection_digests = tuple(window_digest(w) for w in selection)
    selection_default_fills = tuple(
        fill
        for window, digest in zip(selection, selection_digests, strict=True)
        for fill in default_fills_for_window(
            candidate_sim.world,
            candidate_module.spec,
            window,
            dataset_role="selection",
            window_digest=digest,
            input_names=[field.name for field in candidate_u_fields],
            input_defaults=[field.default for field in candidate_u_fields],
            input_fields=candidate_u_fields,
        )
    )
    return held_out_evidence(
        candidate, held_out, sensor=sensor,
        criteria=criteria, lag_count=lag_count,
        training=self._training_digests,
        selection=selection_digests,
        source_model_id=self._source_model_id,
        source_artifact_id=self._source_artifact_id,
        configuration_id=(self._source_model_id
                          if configuration_id is None
                          else configuration_id),
        profile_id=self._profile_id,
        channel_contract_id=channel_contract_id,
        training_default_fills=self._training_default_fills,
        selection_default_fills=selection_default_fills)

derive

derive(*, evidence=None)

Return a new validated model revision carrying fit provenance.

evidence is the typed held-out artifact from evidence(); its criteria-derived accepted decision travels with the revision. Omitting it preserves exploratory fitting while making the resulting artifact visibly unaccepted — a model-aided estimator refuses it.

Source code in manta/fit/_map.py

def derive(self, *, evidence: FitEvidence | None = None):
    """Return a new validated model revision carrying fit provenance.

    ``evidence`` is the typed held-out artifact from `evidence()`; its
    criteria-derived ``accepted`` decision travels with the revision.
    Omitting it preserves exploratory fitting while making the
    resulting artifact visibly unaccepted — a model-aided estimator
    refuses it.
    """
    artifact = self._candidate_artifact()
    if evidence is not None:
        if not isinstance(evidence, FitEvidence):
            raise TypeError("FitResult.derive evidence must be a "
                            "FitEvidence")
        binding = evidence.binding
        if binding is None:
            raise ValueError("FitResult.derive refuses unbound evidence; "
                             "use this result's evidence(...) method")
        expected = {
            "fitted_model_id": artifact.model_id,
            "fitted_artifact_id": artifact.artifact_id,
            "source_model_id": self._source_model_id,
            "source_artifact_id": self._source_artifact_id,
            "profile_id": self._profile_id,
            "training_window_digests": self._training_digests,
        }
        mismatch = [name for name, value in expected.items()
                    if getattr(binding, name) != value]
        evidence_training_fills = tuple(
            fill for fill in evidence.default_fills
            if fill.dataset_role == "training"
        )
        if evidence_training_fills != self._training_default_fills:
            mismatch.append("training_default_fills")
        if mismatch:
            raise ValueError("FitResult.derive evidence was issued for a "
                             "different fit/model scope: "
                             f"{', '.join(mismatch)}")
    report = derivation_report(
        "parameter_fit", self._source_artifact_id, self.objective,
        self.values, evidence,
        (self._training_default_fills if evidence is None
         else evidence.default_fills))
    return artifact.with_derivation("fit", report)

summary

summary()

Per-component table: fitted value, prior σ vs posterior σ. post/prior ≈ 1 flags a component the data did not inform — its fitted value is your prior talking, not the flight. Tied parameters follow, showing their derived values and source.

Source code in manta/fit/_map.py

def summary(self) -> str:
    """Per-component table: fitted value, prior σ vs posterior σ.
    `post/prior ≈ 1` flags a component the data did not inform —
    its fitted value is your prior talking, not the flight. Tied
    parameters follow, showing their derived values and source."""
    rows = [("parameter", "fitted", "prior σ", "post σ", "post/prior")]
    i = 0
    for b in self._diagnostic_blocks:
        decision = self.v[b.offset:b.offset + b.dim]
        theta = (
            b.diagnostic_of_v(decision)
            if isinstance(b, _Block)
            else b.theta_of_v(decision)
        )
        for j, lbl in enumerate(b.labels()):
            pri, post = self.prior_sigma[i], self.posterior_sigma[i]
            ratio = ("—" if not np.isfinite(pri) or not np.isfinite(post)
                     else f"{post / pri:.3f}")
            unit = " (rel)" if b.log else ""
            rows.append((lbl, f"{theta[j]:.6g}",
                         ("inf" if np.isinf(pri)
                          else f"{pri:.3g}{unit}"),
                         ("not computed" if np.isnan(post) else
                          "inf" if np.isinf(post)
                          else f"{post:.3g}{unit}"),
                         ratio))
            i += 1
    # Tied parameters carry no decision variable of their own, so
    # they get ONE row each (the derived vector, not a component per
    # line) naming the source that does.
    for full, dim in self._fields:
        src = self._tie_sources.get(full)
        if src is None:
            continue
        theta = np.atleast_1d(self.values[full])
        val = (f"{theta[0]:.6g}" if dim == 1
               else "[" + " ".join(f"{v:.4g}" for v in theta) + "]")
        rows.append((full, val, f"← {src}", "", ""))
    return (convergence_line(self.converged, self.stats) + "\n"
            + format_table(rows))

manta.FitProgress dataclass

FitProgress(iteration, objective, best_objective, initial_objective, values)

One accepted IPOPT iteration exposed at the fitting boundary.

values contains ambient model parameter values with structural ties already resolved, plus any decision-only :class:Free values. It is the retained best iterate, not necessarily IPOPT's current trial point, so a caller can safely checkpoint it. Return False from a progress callback to request an orderly early stop; None or True continues.

FitResult.evidence(held_out, sensor=...) computes the typed held-out evidence (below) on windows the fit never saw; FitResult.derive(evidence=...) returns an immutable ModelArtifact with the source revision, objective, fitted values, and that evidence. apply() remains the mutable alternative for iterative authoring.

Noise fit

manta.NoiseFit

NoiseFit(world, noise, *, sensors=None, estimator=None)

Innovation-NLL fit of noise σ values (see module docstring).

Args: world — the model (dynamics/geometry at their — ideally already fitted — declared values). noise — {channel name/suffix: Prior | None}. Channel names are the declaration names (drone.imu.gyro_noise, drone.imu.gyro_bias); priors are relative (log-space), None = flat. sensors — measurement outputs the filter consumes (default: all with traces required in every window). estimator — optional model-derived estimator transform. Supplying an INS reuses its strapdown transition, selected sensor set, IMU prediction inputs, and measurement-source mapping.

Source code in manta/fit/_nll.py

def __init__(self, world, noise: dict, *,
             sensors: list[str] | None = None,
             estimator=None) -> None:
    source = world
    self.world = (world.world_copy()
                  if isinstance(world, ModelArtifact) else world)
    if estimator is None:
        self.estimator = None
        self.sys = LinearizedSystem(source, sensors=sensors)
    else:
        self.estimator = _resolve_estimator(source, estimator)
        self.sys = self.estimator.sys
        if sensors is not None:
            chosen = {
                resolve_suffix(name, list(self.sys.sensors),
                               label="sensor", who="NoiseFit")
                for name in sensors
            }
            if chosen != set(self.sys.sensors):
                raise ValueError(
                    "NoiseFit: when estimator= is supplied, select its "
                    "sensor set on the estimator transform")
    self.model_world = self.sys.world
    sys = self.sys

    # Resolve requested channels against the tick's noise vector.
    aliases = []
    for spec in sys.noise_specs:
        aliases.append(spec.full.removesuffix("_driver"))
    chosen: dict[int, Prior | None] = {}
    for key, prior in noise.items():
        alias = resolve_suffix(key, aliases, label="noise channel",
                               who="NoiseFit")
        chosen[aliases.index(alias)] = prior
    self.channels = [
        _Channel(sys.noise_specs[idx], k, prior)
        for k, (idx, prior) in enumerate(sorted(chosen.items()))]
    if not self.channels:
        raise ValueError(
            "NoiseFit: no noise channels selected — name at least one "
            "channel in noise={...}.")
    self._chan_by_spec = {c.spec.full: c for c in self.channels}
    self.n_s = len(self.channels)

    self._step_fn = self._build_step()
    self._fold_cache: dict[int, ca.Function] = {}
    self._window_nll_cache: dict[int, ca.Function] = {}
    self._validate_R0()

solve

solve(windows, *, P0=1e-06, solver='ipopt', verbose=False, progress=None, batched_options=None, ipopt_options=None, hessian_diagnostics='finite-difference')

Minimize the windows' total innovation NLL + prior over log-σ.

Args: windows — recorded data; every chosen sensor needs a trace in every window. P0 — initial tangent covariance per window, P0 · I. Keep small when x0 is trusted (synthetic truth); grow it for estimator-seeded initial states.

Source code in manta/fit/_nll.py

def solve(self, windows: list[Window], *, P0: float = 1e-6,
          solver: str = "ipopt",
          verbose: bool = False,
          progress: Callable[[NoiseFitProgress], bool | None] | None = None,
          batched_options: dict | None = None,
          ipopt_options: dict | None = None,
          hessian_diagnostics: str = "finite-difference") -> NoiseFitResult:
    """Minimize the windows' total innovation NLL + prior over log-σ.

    Args:
        windows — recorded data; every chosen sensor needs a trace
                  in every window.
        P0      — initial tangent covariance per window, `P0 · I`.
                  Keep small when `x0` is trusted (synthetic truth);
                  grow it for estimator-seeded initial states.
    """
    if not windows:
        raise ValueError("NoiseFit.solve: needs at least one Window.")
    if solver not in ("ipopt", "batched"):
        raise ValueError("NoiseFit solver must be 'ipopt' or 'batched'")
    if hessian_diagnostics not in ("finite-difference", "exact"):
        raise ValueError(
            "NoiseFit hessian_diagnostics must be 'finite-difference' "
            "or 'exact'"
        )
    sys = self.sys
    spec = sys.spec
    if solver == "batched":
        if hessian_diagnostics == "exact":
            raise ValueError(
                "batched NoiseFit intentionally avoids exact symbolic Hessians"
            )
        s_opt, objective, stats, hessian = self._solve_batched(
            windows,
            P0=P0,
            progress=progress,
            options=batched_options,
        )
        expanded = False
    else:
        tan = spec.tangent_dim
        s = ca.MX.sym("s", self.n_s, 1)
        total = ca.MX(0.0)
        for w in windows:
            x0, U, Z, M, K = self._window_arrays(w)
            fold = self._fold(K)
            res = fold(ca.DM(x0),
                       ca.DM((P0 * np.eye(tan)).reshape(-1, 1)),
                       ca.DM(U) if U.size else ca.DM(0, K),
                       ca.DM(Z),
                       ca.DM(M),
                       ca.repmat(s, 1, K),
                       ca.repmat(ca.DM(float(w.dt)), 1, K),
                       ca.DM(np.array([[w.t0 + i * w.dt
                                        for i in range(K)]])))
            total = total + ca.sum2(res[2])

        # ½‖(s − s̄)/σ‖² prior (skipped for flat-prior channels).
        total = total + prior_penalty(s, self.channels, weight=0.5)

        def emit_progress(
            iteration,
            _current,
            current_objective,
            best,
            best_objective,
            initial_objective,
        ):
            if progress is None:
                return None
            return progress(NoiseFitProgress(
                iteration=iteration,
                objective=float(current_objective),
                best_objective=float(best_objective),
                initial_objective=float(initial_objective),
                values={
                    channel.alias: float(np.exp(best[channel.offset]))
                    for channel in self.channels
                },
            ))

        s_opt, objective, stats, expanded = solve_blocks_nlp(
            "noise_fit", s, total, self.channels,
            verbose=verbose, ipopt_options=ipopt_options,
            retain_best=True, progress=emit_progress)

    if solver == "ipopt" and hessian_diagnostics == "exact":
        # Exact symbolic second derivatives can be useful for small bench
        # problems, but duplicate a large folded EKF graph in production.
        H_fn = ca.Function("H", [s], [ca.hessian(total, s)[0]])
        H_fn = expand_or_none(H_fn) or H_fn
        hessian = np.asarray(ca.DM(H_fn(s_opt)))
    elif solver == "ipopt":
        # Noise fits usually have only a handful of log-sigma decisions.
        # Central differences of the analytic first derivative avoid
        # constructing the folded filter's symbolic second derivative.
        gradient_fn = ca.Function("noise_fit_gradient", [s], [ca.gradient(total, s)])
        gradient_fn = expand_or_none(gradient_fn) or gradient_fn
        hessian = np.zeros((self.n_s, self.n_s), dtype=float)
        for index in range(self.n_s):
            step = 1e-5 * max(1.0, abs(float(s_opt[index])))
            plus = np.asarray(s_opt, dtype=float).copy()
            minus = np.asarray(s_opt, dtype=float).copy()
            plus[index] += step
            minus[index] -= step
            g_plus = np.asarray(ca.DM(gradient_fn(plus))).ravel()
            g_minus = np.asarray(ca.DM(gradient_fn(minus))).ravel()
            hessian[:, index] = (g_plus - g_minus) / (2.0 * step)
        hessian = 0.5 * (hessian + hessian.T)

    res = NoiseFitResult(self.channels, s_opt, hessian, objective,
                         stats, self.world, self.sys.model.model_id,
                         self.sys.model.artifact_id,
                         self.sys.model.derivation)
    res.expanded = expanded
    res._training_digests = tuple(window_digest(w) for w in windows)
    prediction = (() if self.estimator is None else tuple(
        self.estimator.module().metadata.get("prediction_inputs", ())
    ))
    input_fields = getattr(sys, "input_fields", None)
    res._training_default_fills = tuple(sorted((
        fill
        for window, digest in zip(
            windows, res._training_digests, strict=True
        )
        for fill in default_fills_for_window(
            self.model_world,
            spec,
            window,
            dataset_role="training",
            window_digest=digest,
            input_names=sys.input_names,
            input_defaults=sys.u_defaults,
            input_fields=input_fields,
            recorded_inputs={
                **window.u,
                **{name: 0.0 for name in prediction},
            },
        )
    ), key=lambda fill: (
        fill.dataset_role, fill.window_digest, fill.source, fill.name
    )))
    return res

manta.NoiseFitResult

NoiseFitResult(channels, s_opt, hessian, objective, stats, world, source_model_id, source_artifact_id, source_derivation)

Fitted σ per channel + Laplace posterior diagnostics.

prior_sigma / posterior_sigma are RELATIVE (log-space) widths; posterior ≈ prior means the data didn't inform that σ. converged is IPOPT's success flag — False ⇒ the values are the failed solve's final iterate (a RuntimeWarning was emitted), not an optimum. expanded records whether the NLL ran SX-expanded (False = a Linsol node kept it on the slower interpreted MX path; a RuntimeWarning said so at solve time).

Source code in manta/fit/_nll.py

def __init__(self, channels, s_opt, hessian, objective, stats,
             world, source_model_id, source_artifact_id,
             source_derivation) -> None:
    self._channels = channels
    self._world = world
    self._source_model_id = source_model_id
    self._source_artifact_id = source_artifact_id
    self._source_derivation = dict(source_derivation)
    self.s = np.asarray(s_opt, dtype=float).ravel()
    self.objective = float(objective)
    self.stats = stats
    self.converged = solver_converged(stats, who="NoiseFit")
    self.values = {c.alias: float(np.exp(self.s[c.offset]))
                   for c in channels}
    self.labels = [c.alias for c in channels]
    self.prior_sigma = np.concatenate([c.sigma for c in channels])
    # eigh-based: a non-PD direction (indefinite/near-singular Laplace
    # Hessian) reports inf — never a fake "perfectly identified" 0.
    self.posterior_sigma = laplace_sigma(hessian)
    self.posterior_ratio = tuple(
        float(post / prior)
        if np.isfinite(prior) and prior > 0.0 else (
            0.0 if np.isfinite(post) else float("inf")
        )
        for prior, post in zip(
            self.prior_sigma, self.posterior_sigma, strict=True
        )
    )
    self.active_bounds = tuple(
        channel.alias
        for channel in channels
        if (
            np.isfinite(channel.lower[0])
            and self.s[channel.offset] - channel.lower[0]
            <= 1e-6 * max(1.0, abs(self.s[channel.offset]))
        ) or (
            np.isfinite(channel.upper[0])
            and channel.upper[0] - self.s[channel.offset]
            <= 1e-6 * max(1.0, abs(self.s[channel.offset]))
        )
    )
    self._profile_id = hashlib.sha256(
        b"manta-noise-fit-profile-v1\0" + canonical_derivation_bytes({
            "source_model_id": self._source_model_id,
            "source_artifact_id": self._source_artifact_id,
            "objective": self.objective,
            "values": self.values,
        })).hexdigest()

apply

apply()

Write the fitted σ back onto the owning parts (<channel>_sigma attributes); transforms built afterwards (an EKF(world)'s auto-Q/R, a NoiseDriverd truth sim) use them.

Source code in manta/fit/_nll.py

def apply(self) -> None:
    """Write the fitted σ back onto the owning parts
    (`<channel>_sigma` attributes); transforms built afterwards
    (an `EKF(world)`'s auto-Q/R, a `NoiseDriver`d truth sim) use
    them."""
    if not self.converged:
        raise RuntimeError(
            "NoiseFitResult.apply refuses to write an unconverged solve")
    staged = self._staged_updates(self._world)
    for owner, attr, value in staged:
        setattr(owner, attr, value)

fitted_world

fitted_world()

An editable copy of the authoring world with the fitted σ values written in — what derive() freezes and evidence() predicts with. Refuses an unconverged solve.

Source code in manta/fit/_nll.py

def fitted_world(self):
    """An editable copy of the authoring world with the fitted σ values
    written in — what `derive()` freezes and `evidence()` predicts
    with. Refuses an unconverged solve."""
    if not self.converged:
        raise RuntimeError(
            "NoiseFitResult.fitted_world refuses an unconverged solve")
    derived = copy.deepcopy(self._world)
    for owner, attr, value in self._staged_updates(derived):
        setattr(owner, attr, value)
    return derived

evidence

evidence(held_out, *, sensor, criteria=None, lag_count=20, selection=(), configuration_id=None, channel_contract_id=None)

Held-out evidence for the fitted model (see held_out_evidence); windows that entered the fit are refused.

Source code in manta/fit/_nll.py

def evidence(self, held_out: list[Window], *, sensor: str,
             criteria: FitAcceptanceCriteria | None = None,
             lag_count: int = 20,
             selection: list[Window] = (),
             configuration_id: str | None = None,
             channel_contract_id: str | None = None) -> FitEvidence:
    """Held-out evidence for the fitted model (see `held_out_evidence`);
    windows that entered the fit are refused."""
    from ..sim import Sim

    candidate = self._candidate_artifact()
    candidate_sim = Sim(candidate)
    candidate_module = candidate_sim.module()
    candidate_u_fields = candidate_module.port("u").fields
    selection_digests = tuple(window_digest(w) for w in selection)
    selection_default_fills = tuple(
        fill
        for window, digest in zip(selection, selection_digests, strict=True)
        for fill in default_fills_for_window(
            candidate_sim.world,
            candidate_module.spec,
            window,
            dataset_role="selection",
            window_digest=digest,
            input_names=[field.name for field in candidate_u_fields],
            input_defaults=[field.default for field in candidate_u_fields],
            input_fields=candidate_u_fields,
        )
    )
    return held_out_evidence(
        candidate, held_out, sensor=sensor,
        criteria=criteria, lag_count=lag_count,
        training=self._training_digests,
        selection=selection_digests,
        source_model_id=self._source_model_id,
        source_artifact_id=self._source_artifact_id,
        configuration_id=(self._source_model_id
                          if configuration_id is None
                          else configuration_id),
        profile_id=self._profile_id,
        channel_contract_id=channel_contract_id,
        training_default_fills=self._training_default_fills,
        selection_default_fills=selection_default_fills)

derive

derive(*, evidence=None)

Return a structurally validated model revision carrying the typed held-out evidence (or none — visibly unaccepted).

Source code in manta/fit/_nll.py

def derive(self, *, evidence: FitEvidence | None = None):
    """Return a structurally validated model revision carrying the
    typed held-out evidence (or none — visibly unaccepted)."""
    artifact = self._candidate_artifact()
    if evidence is not None:
        if not isinstance(evidence, FitEvidence):
            raise TypeError("NoiseFitResult.derive evidence must be a "
                            "FitEvidence")
        binding = evidence.binding
        if binding is None:
            raise ValueError("NoiseFitResult.derive refuses unbound "
                             "evidence; use this result's evidence(...) "
                             "method")
        expected = {
            "fitted_model_id": artifact.model_id,
            "fitted_artifact_id": artifact.artifact_id,
            "source_model_id": self._source_model_id,
            "source_artifact_id": self._source_artifact_id,
            "profile_id": self._profile_id,
            "training_window_digests": self._training_digests,
        }
        mismatch = [name for name, value in expected.items()
                    if getattr(binding, name) != value]
        evidence_training_fills = tuple(
            fill for fill in evidence.default_fills
            if fill.dataset_role == "training"
        )
        if evidence_training_fills != self._training_default_fills:
            mismatch.append("training_default_fills")
        if mismatch:
            raise ValueError("NoiseFitResult.derive evidence was issued "
                             "for a different fit/model scope: "
                             f"{', '.join(mismatch)}")
    report = derivation_report(
        "noise_fit", self._source_artifact_id, self.objective,
        self.values, evidence,
        (self._training_default_fills if evidence is None
         else evidence.default_fills))
    return artifact.with_derivation("noise_fit", report)

Noise fits support the same derive() / apply() split.

manta.FitDerivationReport dataclass

FitDerivationReport(method, source_artifact_id, objective, values, evidence, default_fill_policy_id=DEFAULT_FILL_POLICY_ID, default_fills=())

Provenance of one derived model revision.

evidence is the typed held-out artifact (FitEvidence) or None for an exploratory derivation that computed none; there is no untyped form. accepted is the evidence's own criteria-derived decision and is never set by a caller — a report without evidence is not accepted. default_fills records every model initial-state/control value used for omitted window data; exploratory reports retain their training fills too.

Held-out evidence

The doctrine's artifact channel: a fitted model's held-out residual bias and its time-correlated process covariance are typed evidence that model-aided estimators consume explicitly — never an implicit zero. hold_out splits the log, held_out_evidence (or FitResult.evidence / NoiseFitResult.evidence) computes the artifact, FitAcceptanceCriteria declares the thresholds that decide FitEvidence.accepted, and ModelForce(evidence=...) consumes it.

FitEvidence.binding scopes that decision to the exact evaluated fitted model and artifact, pre-fit source model and artifact, opaque configuration and profile IDs, disjoint training/selection/acceptance dataset digests, and the qualified Manta channel shape/cadence contract. An integration layer can also provide channel_contract_id for its schema/frame/unit contract. FitResult.derive() and NoiseFitResult.derive() reject evidence issued for another result; model-aided INS rejects unbound evidence when consuming it.

Manta deliberately remains permissive when a window omits initial-state or control fields: it fills them from the model. Every such substitution is a typed FitDefaultFill in the derivation report and evidence, including its training/selection/acceptance role, window digest, source, field name, shape, and exact finite numeric value. These records are canonical artifact identity and never acceptance checks; FitEvidence.accepted still depends only on the declared residual criteria. dt and t0 are not substitutions: every Window already has concrete values for both and window_digest binds them exactly.

manta.FitDefaultFill dataclass

FitDefaultFill(dataset_role, window_digest, source, name, shape, values)

One model value substituted for missing fit-window data.

The record is deliberately numeric and shape-explicit so model artifact provenance has one deterministic JSON representation. dataset_role is one of training, selection, or acceptance; source is model_initial_state or model_control_default.

manta.fit.hold_out

hold_out(windows, *, fraction=0.3)

Deterministic training / held-out split: the last ceil(fraction · n) windows (in the order given) are held out and must never enter the fit. Both sides must be non-empty.

Source code in manta/fit/_evidence.py

def hold_out(windows: Sequence[Window], *, fraction: float = 0.3
             ) -> tuple[list[Window], list[Window]]:
    """Deterministic training / held-out split: the last
    ``ceil(fraction · n)`` windows (in the order given) are held out and
    must never enter the fit. Both sides must be non-empty."""
    windows = list(windows)
    fraction = _finite(fraction, name="hold_out fraction", minimum=0.0,
                       strict=True)
    if fraction >= 1.0:
        raise ValueError(f"hold_out fraction must be < 1, got {fraction!r}")
    n_held = math.ceil(fraction * len(windows))
    if n_held < 1 or n_held >= len(windows):
        raise ValueError(
            f"hold_out: {len(windows)} window(s) at fraction {fraction:g} "
            f"leaves {n_held} held out — both sides need at least one window")
    return windows[:-n_held], windows[-n_held:]

manta.fit.held_out_evidence

held_out_evidence(model, windows, *, sensor, criteria=None, lag_count=20, correlation_confidence=0.99, training=(), selection=(), source_model_id=None, source_artifact_id=None, configuration_id=None, profile_id='manta.held_out_replay.v1', channel_contract_id=None, training_default_fills=(), selection_default_fills=())

Compute :class:FitEvidence for sensor on untouched held-out windows.

Args: model — the fitted model: a World or ModelArtifact. Its mean prediction (noise zeroed) is folded from each window's x0 over the recorded controls. windows — the held-out windows; each needs a z trace for sensor with more than lag_count observed samples and all must share dt. A regular z_mask cadence is supported; irregular masks are refused because the autocorrelation model assumes a fixed sample interval. sensor — the measured output whose residual z − h is the model error (full name or unique suffix). criteria — acceptance thresholds (default :class:FitAcceptanceCriteria). lag_count — autocorrelation lags used for the τ/σ² fit (default 20; cover a few correlation times at the sample rate). correlation_confidence — the χ²(2) confidence a Gauss–Markov component must reach over the white model to be kept (default 0.99, a limit of 9.21); below it the axis records the white fallback with this number in its reason. training — content digests (window_digest) of the windows the fit was solved on; a held-out window among them is refused (the acceptance set must be untouched). selection — content digests used to choose the model/profile. They must be disjoint from both training and acceptance. source_model_id/source_artifact_id — identity of the pre-fit source; defaults to the evaluated model for direct replay. configuration_id/profile_id — opaque identities supplied by the integration layer. Manta binds but does not interpret their policy. channel_contract_id — optional external schema/frame/unit digest. The Manta channel name, shape, and cadence are always bound separately.

Source code in manta/fit/_evidence.py

def held_out_evidence(model, windows: Sequence[Window], *, sensor: str,
                      criteria: FitAcceptanceCriteria | None = None,
                      lag_count: int = 20,
                      correlation_confidence: float = 0.99,
                      training: Sequence[str] = (),
                      selection: Sequence[str] = (),
                      source_model_id: str | None = None,
                      source_artifact_id: str | None = None,
                      configuration_id: str | None = None,
                      profile_id: str = "manta.held_out_replay.v1",
                      channel_contract_id: str | None = None,
                      training_default_fills: Sequence[FitDefaultFill] = (),
                      selection_default_fills: Sequence[FitDefaultFill] = (),
                      ) -> FitEvidence:
    """Compute :class:`FitEvidence` for ``sensor`` on untouched held-out
    windows.

    Args:
        model    — the fitted model: a `World` or `ModelArtifact`. Its
                   mean prediction (noise zeroed) is folded from each
                   window's ``x0`` over the recorded controls.
        windows  — the held-out windows; each needs a ``z`` trace for
                   ``sensor`` with more than ``lag_count`` observed samples
                   and all must share ``dt``. A regular ``z_mask`` cadence is
                   supported; irregular masks are refused because the
                   autocorrelation model assumes a fixed sample interval.
        sensor   — the measured output whose residual ``z − h`` is the
                   model error (full name or unique suffix).
        criteria — acceptance thresholds (default
                   :class:`FitAcceptanceCriteria`).
        lag_count — autocorrelation lags used for the τ/σ² fit (default
                   20; cover a few correlation times at the sample rate).
        correlation_confidence — the χ²(2) confidence a Gauss–Markov
                   component must reach over the white model to be kept
                   (default 0.99, a limit of 9.21); below it the axis
                   records the white fallback with this number in its
                   reason.
        training — content digests (`window_digest`) of the windows the
                   fit was solved on; a held-out window among them is
                   refused (the acceptance set must be untouched).
        selection — content digests used to choose the model/profile. They
                   must be disjoint from both training and acceptance.
        source_model_id/source_artifact_id — identity of the pre-fit source;
                   defaults to the evaluated model for direct replay.
        configuration_id/profile_id — opaque identities supplied by the
                   integration layer. Manta binds but does not interpret
                   their policy.
        channel_contract_id — optional external schema/frame/unit digest.
                   The Manta channel name, shape, and cadence are always
                   bound separately.
    """
    from ..estimation.consistency import chi2_quantile
    from ..sim import Sim

    if not windows:
        raise ValueError("held_out_evidence: needs at least one held-out "
                         "Window")
    lag_count = _count(lag_count, name="held_out_evidence lag_count",
                       minimum=1)
    confidence = _finite(correlation_confidence,
                         name="held_out_evidence correlation_confidence")
    if not 0.0 < confidence < 1.0:
        raise ValueError(
            "held_out_evidence correlation_confidence must lie in (0, 1), "
            f"got {correlation_confidence!r}")
    chi2_limit = float(chi2_quantile(2, confidence))
    training = tuple(str(value) for value in training)
    selection = tuple(str(value) for value in selection)
    training_set = set(training)
    selection_set = set(selection)
    if training_set & selection_set:
        raise ValueError("held_out_evidence: training and selection datasets "
                         "must be distinct")
    digests = []
    for index, w in enumerate(windows):
        if not isinstance(w, Window):
            raise TypeError(
                f"held_out_evidence: windows[{index}] is not a Window")
        digest = window_digest(w)
        if digest in training_set:
            raise ValueError(
                f"held_out_evidence: windows[{index}] is a training window "
                "(same content) — the acceptance set must be untouched by "
                "the fit")
        if digest in selection_set:
            raise ValueError(
                f"held_out_evidence: windows[{index}] is a selection window "
                "(same content) — the acceptance set must be untouched by "
                "model/profile selection")
        digests.append(digest)
    dts = {float(w.dt) for w in windows}
    if len(dts) != 1:
        raise ValueError(
            f"held_out_evidence: held-out windows must share dt, got "
            f"{sorted(dts)}")
    base_dt = dts.pop()

    sim = Sim(model)
    # Held-out windows already encode acquisition events in z_mask. Replay
    # against the all-inline observation graph so rate-gated sensors remain
    # available to the batch evaluator without fabricating held samples.
    module = sim.inline_module()
    spec = module.spec
    ep = module.entry("step")
    meas_names = [pt.name for pt in module.ports_by_role(Role.MEASUREMENT)]
    full = resolve_suffix(sensor, meas_names, label="sensor",
                          who="held_out_evidence")
    channel_port = module.port(full)
    dim = channel_port.size
    dims = {n: module.port(n).size for n in meas_names}
    u_fields = module.port("u").fields
    n_noise = module.port("noise").size
    params = next((pt for pt in module.ports if pt.name == "params"), None)
    out_index = 1 + ep.returns.index(full)
    step = module.functions["step"]

    axis_names = _AXES3 if dim == 3 else tuple(str(i) for i in range(dim))
    segments: list[list[np.ndarray]] = [[] for _ in range(dim)]
    default_fills = [*training_default_fills, *selection_default_fills]
    total = 0
    observed_stride: int | None = None
    for index, w in enumerate(windows):
        traces, K = resolve_traces(w.z, meas_names, dims,
                                   who="held_out_evidence")
        if full not in traces:
            raise ValueError(
                f"held_out_evidence: windows[{index}] has no z trace for "
                f"{full!r}")
        masks = resolve_trace_masks(
            w.z_mask, traces, meas_names, K, who="held_out_evidence"
        )
        observed = np.flatnonzero(masks[full])
        if observed.size <= lag_count:
            raise ValueError(
                f"held_out_evidence: windows[{index}] has "
                f"{observed.size} observed samples; the "
                f"autocorrelation fit over {lag_count} lags needs more than "
                f"{lag_count}")
        differences = np.diff(observed)
        stride = int(differences[0])
        if np.any(differences != stride):
            raise ValueError(
                f"held_out_evidence: windows[{index}] has an irregular "
                f"z_mask cadence for {full!r}; fixed-step autocorrelation "
                "evidence requires regularly spaced observations"
            )
        if observed_stride is None:
            observed_stride = stride
        elif stride != observed_stride:
            raise ValueError(
                "held_out_evidence: held-out windows must share one "
                f"observation cadence, got strides {observed_stride} and "
                f"{stride}"
            )
        x0 = pack_x0(sim.world, spec, w)
        default_fills.extend(default_fills_for_window(
            sim.world,
            spec,
            w,
            dataset_role="acceptance",
            window_digest=digests[index],
            input_names=[field.name for field in u_fields],
            input_defaults=[field.default for field in u_fields],
            input_fields=u_fields,
        ))
        U = pack_u_trace(
            w.u, [f.name for f in u_fields],
            [float(np.asarray(f.default).ravel()[0]) for f in u_fields],
            K, who="held_out_evidence")
        call_args = {
            "x": x0,
            "u": U if U.size else np.zeros((0, K)),
            "noise": np.zeros((n_noise, K)),
            "dt": np.full((1, K), base_dt),
            "t": np.array([[w.t0 + i * base_dt for i in range(K)]]),
        }
        if params is not None:
            p = np.concatenate([np.atleast_1d(np.asarray(
                f.default, dtype=float)).ravel() for f in params.fields])
            call_args["params"] = np.tile(p.reshape(-1, 1), (1, K))
        ordered = []
        for a in ep.args:
            key = a.name if isinstance(a, PortRef) else "x"
            if key not in call_args:
                raise KeyError(
                    f"held_out_evidence: step entry takes unknown port arg "
                    f"{key!r}")
            ordered.append(call_args[key])
        outs = step.mapaccum(f"evidence_x{K}", K, [0], [0])(*ordered)
        predicted = np.asarray(outs[out_index], dtype=float).reshape(dim, K)
        residual = traces[full][observed] - predicted.T[observed]
        if not np.all(np.isfinite(residual)):
            raise ValueError(
                f"held_out_evidence: windows[{index}] produced a non-finite "
                f"residual for {full!r}")
        for i in range(dim):
            segments[i].append(residual[:, i])
        total += observed.size

    assert observed_stride is not None
    evidence_dt = base_dt * observed_stride
    axes = [_axis_evidence(axis_names[i], segments[i], dt=evidence_dt,
                           lag_count=lag_count, chi2_limit=chi2_limit)
            for i in range(dim)]
    held = HeldOutWindow(window_count=len(windows), sample_count=total,
                         dt=evidence_dt, window_digests=tuple(digests))
    evaluated = sim.model
    source_model_id = evaluated.model_id if source_model_id is None \
        else source_model_id
    source_artifact_id = evaluated.artifact_id \
        if source_artifact_id is None else source_artifact_id
    configuration_id = evaluated.model_id if configuration_id is None \
        else configuration_id
    if channel_contract_id is None:
        observed_rate_hz = 1.0 / evidence_dt
        contract = repr((full, channel_port.role.value, channel_port.shape,
                         observed_rate_hz)).encode()
        channel_contract_id = hashlib.sha256(
            b"manta-channel-contract-v1\0" + contract).hexdigest()
    observed_rate_hz = 1.0 / evidence_dt
    binding = FitEvidenceBinding(
        fitted_model_id=evaluated.model_id,
        fitted_artifact_id=evaluated.artifact_id,
        source_model_id=source_model_id,
        source_artifact_id=source_artifact_id,
        configuration_id=configuration_id,
        profile_id=profile_id,
        training_window_digests=training,
        selection_window_digests=selection,
        acceptance_window_digests=tuple(digests),
        channel_shape=channel_port.shape,
        channel_rate_hz=observed_rate_hz,
        channel_contract_id=channel_contract_id,
    )
    return FitEvidence.evaluate(channel=full, held_out=held, axes=axes,
                                criteria=criteria, binding=binding,
                                default_fills=default_fills)

manta.FitEvidence dataclass

FitEvidence(channel, held_out, axes, criteria, checks, accepted, binding=None, default_fill_policy_id=DEFAULT_FILL_POLICY_ID, default_fills=())

The typed held-out fit-evidence artifact.

Construct through :meth:evaluate; checks and accepted are a pure function of axes and criteria and construction refuses any other value. The artifact is a frozen dataclass of scalars, strings and tuples, so ModelArtifact's canonical derivation hashing covers it field by field. Missing-window substitutions are carried separately as default_fills and never participate in acceptance checks.

evaluate classmethod

evaluate(*, channel, held_out, axes, criteria=None, binding=None, default_fills=())

Build the artifact, deciding accepted from criteria (default :class:FitAcceptanceCriteria).

Source code in manta/fit/_evidence.py

@classmethod
def evaluate(cls, *, channel: str, held_out: HeldOutWindow,
             axes: Sequence[AxisFitEvidence],
             criteria: FitAcceptanceCriteria | None = None,
             binding: FitEvidenceBinding | None = None,
             default_fills: Sequence[FitDefaultFill] = (),
             ) -> FitEvidence:
    """Build the artifact, deciding ``accepted`` from ``criteria``
    (default :class:`FitAcceptanceCriteria`)."""
    criteria = FitAcceptanceCriteria() if criteria is None else criteria
    axes = tuple(axes)
    checks = _evaluate_checks(axes, criteria)
    return cls(channel=channel, held_out=held_out, axes=axes,
               criteria=criteria, checks=checks,
               accepted=all(c.passed for c in checks), binding=binding,
               default_fills=tuple(default_fills))

manta.FitEvidenceBinding dataclass

FitEvidenceBinding(fitted_model_id, fitted_artifact_id, source_model_id, source_artifact_id, configuration_id, profile_id, training_window_digests, selection_window_digests, acceptance_window_digests, channel_shape, channel_rate_hz, channel_contract_id)

Exact identity scope in which held-out evidence is valid.

The opaque configuration/profile identifiers are deliberately generic: Manta binds them but does not interpret vehicle or release policy. Dataset identities are content digests, and the channel contract is the qualified Manta port name, shape, cadence, and caller's optional external schema digest.

manta.AxisFitEvidence dataclass

AxisFitEvidence(axis, sample_count, residual_bias, residual_bias_stderr, residual_rms, lag_one_autocorrelation, lag_count, fitted_tau, fitted_correlated_fraction, correlation_chi2, correlation_chi2_limit, white_floor_fraction, noise_model, white_sigma, autocorrelation_rmse, white_fallback, white_fallback_reason)

Held-out evidence for one residual axis.

residual_bias is the held-out mean residual mean(z − h) with its standard error (effective sample count corrected for lag-one autocorrelation); residual_rms the raw root-mean-square residual. The autocorrelation fit ρ(l) ≈ a·exp(−l·dt/τ) over lag_count lags yields fitted_tau and fitted_correlated_fraction (a). correlation_chi2 is the fit's significance statistic n·(SS_white − SS_fit) — the reduction in squared autocorrelation misfit the two-parameter model buys over the white model, which is χ²(2)-distributed for white residuals — and correlation_chi2_limit the declared quantile it had to exceed. The decision that followed is explicit: noise_model is the Gauss–Markov model when the correlated component is significant and the fitted τ is at or above the sample interval, otherwise the white model with white_fallback set and white_fallback_reason naming why. white_sigma is the uncorrelated per-sample floor (equal to noise_model.sigma for a white model). For a Gauss–Markov model the white fraction of the variance is max(1 − a, white_floor_fraction) with white_floor_fraction = 1/√n: a sample autocorrelation over n points scatters by about 1/√n, so it cannot resolve a white component smaller than that — the floor is the data's resolving power (recorded here), never a default, and it keeps the pseudo-measurement covariance away from the singular R = 0 a saturated fit would imply. autocorrelation_rmse is the RMS misfit between the empirical autocorrelation and the chosen model's.

total_sigma property

total_sigma

Stationary 1-σ of the modelled residual: white floor plus the correlated component (for a random walk, the floor alone — its variance is unbounded).

bias_ratio property

bias_ratio

|held-out bias| relative to the modelled residual σ.

manta.ProcessNoiseModel dataclass

ProcessNoiseModel(kind, sigma, tau=None)

One axis' fitted process-noise model.

Args: kind — "white" (per-sample σ), "gauss_markov" (stationary σ with correlation time tau seconds), or "random_walk" (σ/√Hz drift density). sigma — 1-σ magnitude in the residual's units (≥ 0). tau — correlation time in seconds; required (finite, > 0) for "gauss_markov" and forbidden otherwise.

manta.HeldOutWindow dataclass

HeldOutWindow(window_count, sample_count, dt, window_digests)

Definition of the held-out (acceptance) set the evidence was computed on: how many windows, how many samples, at which step, and the content digest of every window (the identity a consumer can check against its training set).

manta.FitAcceptanceCriteria dataclass

FitAcceptanceCriteria(max_bias_ratio=0.5, max_autocorrelation_rmse=0.15, min_samples=200, max_residual_rms=None)

The numeric thresholds that decide FitEvidence.accepted.

Every criterion applies per axis; the artifact records each check's value, limit, and outcome. Defaults:

• max_bias_ratio = 0.5 — |held-out bias| ≤ 0.5 × the modelled residual σ. A bias larger than that is systematic model error, not noise the filter can absorb.
• max_autocorrelation_rmse = 0.15 — the empirical autocorrelation over the fitted lags must match the chosen noise model to within 0.15 RMS (white noise over N ≥ 200 samples scatters at ≈ 1/√N ≈ 0.07, so this accepts sampling scatter and rejects a mis-modelled spectrum).
• min_samples = 200 — held-out samples per axis; fewer cannot support the autocorrelation fit.
• max_residual_rms = None — optional absolute ceiling on the raw held-out residual RMS in the channel's units (None = no ceiling; set it from the vehicle's risk policy).

manta.fit.AcceptanceCheck dataclass

AcceptanceCheck(criterion, axis, value, limit, passed)

One criterion evaluated on one axis.

manta.fit.window_digest

window_digest(window)

Content identity of a Window (sha256 over every trace, dt, t0) — what HeldOutWindow.window_digests records and what the untouched-acceptance-set check compares.

Source code in manta/fit/_evidence.py

def window_digest(window: Window) -> str:
    """Content identity of a `Window` (sha256 over every trace, `dt`,
    `t0`) — what `HeldOutWindow.window_digests` records and what the
    untouched-acceptance-set check compares."""
    if not isinstance(window, Window):
        raise TypeError(f"window_digest expects a Window, got "
                        f"{type(window).__name__}")
    digest = hashlib.sha256(b"manta-window-v2\0")

    def feed(tag: str, mapping: dict) -> None:
        digest.update(tag.encode())
        for key in sorted(mapping, key=str):
            value = mapping[key]
            if isinstance(value, dict):
                feed(f"{tag}/{key}", value)
                continue
            array = np.asarray(value, dtype=float)
            digest.update(repr((str(key), array.shape)).encode())
            digest.update(np.ascontiguousarray(array).tobytes())

    feed("x0", window.x0)
    feed("x0_sigma", window.x0_sigma)
    feed("u", window.u)
    feed("x", window.x)
    feed("x_scale", window.x_scale)
    feed("z", window.z)
    feed("z_mask", window.z_mask)
    digest.update(repr((float(window.dt), float(window.t0))).encode())
    return digest.hexdigest()

Residual covariance

bartlett_hac_residual_statistics is the dimension-generic mathematical boundary used by fitting and reduction pipelines after they produce residual sequences. It keeps independent windows separate, reports bias explicitly, and returns both instantaneous sample covariance and a positive-semidefinite Bartlett/Newey–West long-run covariance suitable as white-equivalent process noise. Vehicle replay, acceptance thresholds, and release policy remain with the caller.

manta.bartlett_hac_residual_statistics

bartlett_hac_residual_statistics(sequences, *, reference_dt_s, correlation_horizon_s)

Estimate bias, sample covariance, and Bartlett-HAC covariance.

Each input is one independent (samples, dimension) sequence. Lagged products never cross sequence boundaries, so unrelated fitting windows do not acquire artificial adjacency. All sequences must share a dimension; the calculation is otherwise dimension-agnostic.

Source code in manta/fit/_residuals.py

def bartlett_hac_residual_statistics(
        sequences: Sequence[NDArray[np.float64]], *,
        reference_dt_s: float,
        correlation_horizon_s: float) -> ResidualStatistics:
    """Estimate bias, sample covariance, and Bartlett-HAC covariance.

    Each input is one independent ``(samples, dimension)`` sequence. Lagged
    products never cross sequence boundaries, so unrelated fitting windows do
    not acquire artificial adjacency. All sequences must share a dimension;
    the calculation is otherwise dimension-agnostic.
    """
    dt = float(reference_dt_s)
    horizon = float(correlation_horizon_s)
    if not math.isfinite(dt) or dt <= 0.0:
        raise ValueError("reference_dt_s must be finite and positive")
    if not math.isfinite(horizon) or horizon < 0.0:
        raise ValueError("correlation_horizon_s must be finite and "
                         "non-negative")
    normalized: list[NDArray[np.float64]] = []
    dimension: int | None = None
    for index, sequence in enumerate(sequences):
        array = np.asarray(sequence, dtype=float)
        if array.ndim != 2 or array.shape[0] < 1 or array.shape[1] < 1:
            raise ValueError(
                f"residual sequences[{index}] must have non-empty "
                "(samples, dimension) shape")
        if not np.all(np.isfinite(array)):
            raise ValueError(f"residual sequences[{index}] must be finite")
        if dimension is None:
            dimension = array.shape[1]
        elif array.shape[1] != dimension:
            raise ValueError("residual sequences must share one dimension")
        normalized.append(array)
    if not normalized:
        raise ValueError("residual sequences may not be empty")

    residuals = np.vstack(normalized)
    bias = np.mean(residuals, axis=0)
    centered_sequences = [sequence - bias for sequence in normalized]
    centered = np.vstack(centered_sequences)
    sample_count = len(centered)
    instantaneous = centered.T @ centered / max(1, sample_count - 1)

    maximum_lag = min(len(sequence) - 1 for sequence in centered_sequences)
    lag_steps = min(maximum_lag, max(0, round(horizon / dt)))
    long_run = sum(
        (sequence.T @ sequence for sequence in centered_sequences),
        np.zeros_like(instantaneous),
    ) / sample_count
    for lag in range(1, lag_steps + 1):
        cross = sum(
            (sequence[lag:].T @ sequence[:-lag]
             for sequence in centered_sequences),
            np.zeros_like(instantaneous),
        ) / sample_count
        weight = 1.0 - lag / (lag_steps + 1.0)
        long_run += weight * (cross + cross.T)

    (
        instantaneous,
        instantaneous_minimum,
        instantaneous_correction_norm,
        instantaneous_correction_count,
    ) = _positive_semidefinite(
        instantaneous, name="instantaneous residual covariance"
    )
    (
        long_run,
        long_run_minimum,
        long_run_correction_norm,
        long_run_correction_count,
    ) = _positive_semidefinite(
        long_run, name="Bartlett-HAC residual covariance"
    )
    instantaneous_diagonal = np.diag(instantaneous)
    long_run_diagonal = np.diag(long_run)
    effective_unclipped = np.full(int(dimension), float(sample_count))
    np.divide(
        sample_count * instantaneous_diagonal,
        long_run_diagonal,
        out=effective_unclipped,
        where=long_run_diagonal > np.finfo(float).tiny,
    )
    effective = np.clip(effective_unclipped, 1.0, float(sample_count))
    return ResidualStatistics(
        bias=bias,
        instantaneous_covariance=instantaneous,
        white_equivalent_covariance=long_run,
        reference_dt_s=dt,
        correlation_lag_steps=lag_steps,
        correlation_horizon_s=lag_steps * dt,
        samples=sample_count,
        windows=len(normalized),
        effective_sample_size=effective,
        effective_sample_size_unclipped=effective_unclipped,
        instantaneous_raw_min_eigenvalue=instantaneous_minimum,
        instantaneous_psd_correction_norm=instantaneous_correction_norm,
        instantaneous_psd_correction_count=instantaneous_correction_count,
        white_equivalent_raw_min_eigenvalue=long_run_minimum,
        white_equivalent_psd_correction_norm=long_run_correction_norm,
        white_equivalent_psd_correction_count=long_run_correction_count,
    )

manta.ResidualStatistics dataclass

ResidualStatistics(bias, instantaneous_covariance, white_equivalent_covariance, reference_dt_s, correlation_lag_steps, correlation_horizon_s, samples, windows, effective_sample_size, effective_sample_size_unclipped, instantaneous_raw_min_eigenvalue, instantaneous_psd_correction_norm, instantaneous_psd_correction_count, white_equivalent_raw_min_eigenvalue, white_equivalent_psd_correction_norm, white_equivalent_psd_correction_count)

Bias and covariance evidence from independent residual sequences.

instantaneous_covariance describes individual residual samples. white_equivalent_covariance is the Bartlett/Newey–West long-run covariance: using it as independent per-step noise reproduces the asymptotic integrated-error growth of the observed correlated sequence. Bias remains separate and is never folded into zero-mean covariance. Raw minimum eigenvalues and correction norm/count fields expose any roundoff-scale PSD projection. Materially indefinite estimates are refused. effective_sample_size_unclipped preserves the estimate before the public effective sample size is bounded to [1, samples].

covariance property

covariance

Estimator-ready white-equivalent covariance.

Inputs

manta.Window dataclass

Window(x0, x0_sigma=dict(), u=dict(), x=dict(), x_scale=dict(), z=dict(), z_mask=dict(), dt=0.01, t0=0.0)

One fitting window: a short recorded rollout.

Args: x0 — nested initial state dict (the sim.state shape: {craft: {slot: value}}). Slots omitted fall back to the world's initial state and are recorded as FitDefaultFill provenance. x0_sigma — optional per-slot tangent-space standard deviations for multiple shooting. A named slot must be explicitly present in x0; its value is the prior mean and the fitter optimizes a manifold-aware initial perturbation for that window. Scalars are broadcast across the slot tangent dimension. An SO(3) value is therefore a three-component rotation-vector sigma, never a four-component quaternion sigma. u — recorded controls: {input name/suffix: scalar | (K,)}. A scalar is held for the whole window; inputs omitted hold their model default and are recorded as FitDefaultFill provenance. x — recorded state trajectories: nested or flat mapping from state slot name to (K, ambient_dim) values. Row k is the state after step k. Only named slots enter Fit; quaternion slots are compared on their SO(3) tangent manifold, not componentwise. x_scale — optional positive physical scale per recorded state slot. When present, Fit scores that slot as a trajectory mean-square sum(error²) / (K · scale²) instead of a raw sample sum. This is the explicit mixed-unit normalization boundary: callers choose meaningful floors/tolerances in each slot's native units. z — recorded sensor readings: {sensor name/suffix: (K, dim) | (K,)}. Row k is the reading produced by step k (the step taken FROM state k). For Fit, only sensors present here enter the loss; for NoiseFit, every chosen sensor needs a trace. z_mask — optional explicit availability masks for multi-rate observations. Each key names a trace in z and carries a boolean (K,) array; only true rows enter the fit. Values at false rows are storage placeholders and are never observations. dt — fixed step, seconds. t0 — world-clock time of x0.

dt and t0 are always concrete values and are part of :func:window_digest; FitDefaultFill records only actual model-value substitutions for omitted x0 and u fields.

manta.Prior dataclass

Prior(sigma=None, mean=None, log=False, lower=None, upper=None)

Gaussian prior on one fitted parameter.

Args: sigma — 1-σ width. Scalar (isotropic across the parameter's components) or a per-component sequence. With log=True it is RELATIVE (log-space): sigma=0.3 ≈ ±30%. mean — prior mean. None (default) → the model's declared value. log — fit log(p) instead of p, elementwise. Strictly- positive parameters only (mass, moi, a thrust magnitude along one axis). Keeps every component positive with no constraint and makes pure scale ambiguities linear. NoiseFit ignores this flag — noise σ is ALWAYS fit in log-space, and sigma there is always relative. lower — hard lower bound, AMBIENT space (a value, not a log). upper — hard upper bound, ambient. Scalar or per-component. Enforced as IPOPT box constraints — the sanity rails that keep a physically absurd optimum off the table (a thruster gain that can't exceed the motor's rating, a mount that must stay inside the hull). The prior pulls softly; bounds are walls. In NoiseFit they bound σ itself. The declared/starting value must satisfy them.

Structure

Symmetry and sanity are declared, not hoped for: tie identical or mirrored parameters to one decision variable (Tied), introduce shared geometry as an auxiliary variable (Free), and wall off absurd values with Prior(lower=, upper=) box bounds.

manta.Tied dataclass

Tied(source, scale=None, offset=None)

Structural tie: this promoted parameter is a fixed affine function of another fitted parameter (or a Free variable), not a decision variable of its own — p = scale · p_source + offset, in AMBIENT space (after any log-reparam of the source).

This is how symmetry is enforced rather than hoped for: identical actuators share one gain, mirrored mounts share one geometry. Fewer decision variables ⇒ better-conditioned fits, and every data point that touches any tied copy informs the shared source.

Args: source — name/suffix of the fitted parameter (or Free name) this one derives from. Must itself be a decision variable — chains of ties are not supported; tie all copies to the same source. scale — None (identity — identical copies), a scalar, a per-component sequence (elementwise, e.g. a mirror's sign flips (-1, 1, 1)), or a full (target_dim, source_dim) matrix (e.g. a scalar arm length mapped to a 3-vector mount position). offset — additive constant: None (zero), scalar, or length-target_dim.

Examples::

# four identical motors — one fitted gain
"t1.force_quad": Prior(sigma=3.0),
"t2.force_quad": Tied("t1.force_quad"),

# mirrored mount across the y-z plane
"t2.mount_offset": Tied("t1.mount_offset", scale=(-1, 1, 1)),

# scalar arm length -> the four X-frame mount positions
"arm": Free(0.12, prior=Prior(sigma=0.02, lower=0.0)),
"t1.mount_offset": Tied("arm", scale=[[1], [1], [0]]),
"t2.mount_offset": Tied("arm", scale=[[-1], [1], [0]]),

manta.Free dataclass

Free(init, prior=None)

Auxiliary decision variable that is NOT a promoted parameter of the model — it exists to be the source of Tied entries (a shared arm length, a common incidence angle). Its key in Fit(parameters=) is a fresh name, not a part parameter.

Args: init — starting value (scalar or vector); also the prior mean unless the prior says otherwise. prior — optional Prior (sigma/mean/log/bounds), same semantics as for a promoted parameter.