Skip to content

System Identification

System identification fits per-DOF dynamics parameters of an entity (damping, armature, stiffness, frictionloss, and the PD gains kp / kd; masses are not fittable) so the simulation reproduces recorded trajectories. A SystemIdentificationBase rolls candidate ParameterSets out in parallel environments, scores them with a residual against Trajectory recordings, and runs an optimizer (CMA-ES or SciPy least squares). Excitation signals drive the robot while you record, and DeploymentRecorder captures real-robot trajectories during a deployment. The planning and sysid guide walks through a fit.

from eden.extensions.sysid import CMAES, Parameter, ParameterSet, SYSID_REGISTRY, Trajectory
from eden.options.extensions.sysid import SystemIdentificationOptions

Base class, registry and result

Base class and fit-result type for system identification.

Attributes

SYSID_REGISTRY module-attribute 

SYSID_REGISTRY = Registry('SYSID')

Classes

FitResult dataclass 

FitResult(cost: float, nfev: int, x: ndarray, jacobian: ndarray | None = None, message: str = '', history: ndarray | None = None, extras: dict[str, Any] = dict())

Structured diagnostics returned by :meth:SystemIdentificationBase.fit.

ATTRIBUTE DESCRIPTION
cost

Final 0.5 * ||residual||^2.

TYPE: float

nfev

Number of residual evaluations performed.

TYPE: int

x

Flat optimised parameter vector (matches ParameterSet.as_vector).

TYPE: ndarray

jacobian

Jacobian at the optimum, when the backend provides one.

TYPE: ndarray | None

message

Backend-specific termination message.

TYPE: str

history

Per-iteration cost history, when available.

TYPE: ndarray | None

extras

Backend-specific additional diagnostics.

TYPE: dict[str, Any]

Attributes

cost instance-attribute 
cost: float
nfev instance-attribute 
nfev: int
x instance-attribute 
x: ndarray
jacobian class-attribute instance-attribute 
jacobian: ndarray | None = None
message class-attribute instance-attribute 
message: str = ''
history class-attribute instance-attribute 
history: ndarray | None = None
extras class-attribute instance-attribute 
extras: dict[str, Any] = field(default_factory=dict)

Methods:

SystemIdentificationBase 

SystemIdentificationBase(env: 'EnvBase', options: SystemIdentificationOptions)

Bases: ConfigurableMixin[SystemIdentificationOptions], ABC

Base class for system-identification backends.

Subclasses override :meth:fit to implement the optimisation loop. The base class provides :meth:evaluate for one-shot residual evaluation on the current parameter vector.

PARAMETER DESCRIPTION
entity_name

Name of the entity whose parameters are being identified.
signals

Measurement signals included in the residual (SignalName values).
signal_weights

Per-signal weights (missing signals default to 1.0).
normalize

If True, normalise each signal block by its measured RMS.
max_iters

Maximum optimiser iterations.
verbose

Print per-iteration status.
Notes

The environment passed in should be a dedicated sysid env: a standalone RLEnvBase built with env_options.num_envs equal to the intended batched-candidate population (or 1 for serial backends). The base class avoids RLEnvBase.step entirely so termination, reward, reset, command, event, and recorder managers do not fire during replay.

Source code in eden/extensions/sysid/base.py 
 97
 98
 99
100
101
102
103
104
def __init__(self, env: "EnvBase", options: SystemIdentificationOptions) -> None:
    super().__init__(options)
    self._env = env
    resolve_named_entity(
        env.entities,
        self.entity_name,
        hint="SystemIdentificationOptions.entity_name names it; set it to one of those.",
    )

Attributes

entity_name class-attribute instance-attribute 
entity_name: str = 'robot'
signals class-attribute instance-attribute 
signals: Sequence[str] = ('dofs_pos', 'dofs_vel', 'dofs_torque')
signal_weights class-attribute instance-attribute 
signal_weights: dict[str, float] = {}
normalize class-attribute instance-attribute 
normalize: bool = True
max_iters class-attribute instance-attribute 
max_iters: int = 200
verbose class-attribute instance-attribute 
verbose: bool = True

Methods:

evaluate 
evaluate(params: ParameterSet, trajectories: Sequence[Trajectory]) -> ndarray

Rollout once per trajectory and return the flattened residual.

Source code in eden/extensions/sysid/base.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def evaluate(
    self,
    params: ParameterSet,
    trajectories: Sequence[Trajectory],
) -> np.ndarray:
    """Rollout once per trajectory and return the flattened residual."""
    preds = [
        single_candidate_rollout(
            self._env,
            params,
            traj,
            entity_name=self.entity_name,
            signals=self.signals,
        )
        for traj in trajectories
    ]
    return multi_trajectory_residual(
        preds,
        trajectories,
        signals=self.signals,
        weights=self.signal_weights,
        normalize=self.normalize,
    )
fit abstractmethod 
fit(params: ParameterSet, trajectories: Sequence[Trajectory]) -> tuple[ParameterSet, FitResult]

Run the optimisation loop and return the best ParameterSet + FitResult.

Source code in eden/extensions/sysid/base.py 
130
131
132
133
134
135
136
137
@abstractmethod
def fit(
    self,
    params: ParameterSet,
    trajectories: Sequence[Trajectory],
) -> tuple[ParameterSet, FitResult]:
    """Run the optimisation loop and return the best ParameterSet + FitResult."""
    raise NotImplementedError

Functions:

Options

Configuration options for system-identification extensions.

Attributes

DofProperty module-attribute 

DofProperty = Literal['damping', 'armature', 'stiffness', 'frictionloss', 'kp', 'kd']

SignalName module-attribute 

SignalName = Literal['dofs_pos', 'dofs_vel', 'dofs_torque', 'base_quat', 'base_ang_vel']

Classes

SystemIdentificationOptions 

SystemIdentificationOptions(**data)

Bases: DynamicConfigurableOptions

Options for sysid extensions.

PARAMETER DESCRIPTION
entity_name

Name of the entity being identified (must exist in the scene).
signals

Subset of signals included in the residual. Defaults to proprioception.
signal_weights

Per-signal multiplicative weights. Missing signals default to 1.0.
normalize

Divide each signal block by its measured RMS before weighting.
max_iters

Maximum optimizer iterations / evaluations.
verbose

Log per-iteration progress.
Source code in eden/options/options.py 
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
def __init__(self, **data):
    # Seed type metadata as construction data so BaseModel validates it into
    # __pydantic_extra__ (no-op for extra="forbid" classes, which use
    # the serialization envelope instead).
    data.update(option_metadata(self))
    # NOTE: all of the option modifications/validations should be done in the __init__ method
    try:
        BaseModel.__init__(self, **data)
    except ValidationError as exc:
        # A misspelled field on an extra="forbid" class should teach the field set, not just
        # name the typo. Only rewritten when every error is extra_forbidden.
        enriched = _enrich_extra_forbidden(type(self), exc)
        if enriched is None:
            raise
        raise enriched from None

Attributes

entity_name class-attribute instance-attribute 
entity_name: str = 'robot'
signals class-attribute instance-attribute 
signals: Sequence[SignalName] = ('dofs_pos', 'dofs_vel', 'dofs_torque')
signal_weights class-attribute instance-attribute 
signal_weights: dict[str, float] = {}
normalize class-attribute instance-attribute 
normalize: bool = True
max_iters class-attribute instance-attribute 
max_iters: int = 200
verbose class-attribute instance-attribute 
verbose: bool = True

Parameters and trajectories

Parameter and parameter-set definitions for system identification.

Attributes

Classes

Parameter 

Parameter(name: str, property: DofProperty, dof_names: Sequence[str], entity_name: str = 'robot', per_dof: bool = False, nominal: float | Sequence[float] = 0.0, min_value: float | Sequence[float] = 0.0, max_value: float | Sequence[float] = 1.0, frozen: bool = False)

A single sysid parameter with nominal value, box bounds, and a DOF-property target.

A parameter drives one of the per-DOF solver fields (damping, armature, stiffness, frictionloss) or one of the per-DOF PD gains (kp, kd) on the entity named entity_name for the listed dof_names.

If per_dof is True, value is a length-len(dof_names) vector. Otherwise value is a length-1 vector that is broadcast to all listed DOFs when applied to the entity.

Source code in eden/extensions/sysid/parameter.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def __init__(
    self,
    name: str,
    property: DofProperty,
    dof_names: Sequence[str],
    entity_name: str = "robot",
    per_dof: bool = False,
    nominal: float | Sequence[float] = 0.0,
    min_value: float | Sequence[float] = 0.0,
    max_value: float | Sequence[float] = 1.0,
    frozen: bool = False,
) -> None:
    if not dof_names:
        raise ValueError(f"Parameter '{name}' must target at least one DOF.")

    self.name = name
    self.property = property
    self.entity_name = entity_name
    self.dof_names = tuple(dof_names)
    self.per_dof = per_dof
    self.frozen = frozen

    expected_size = len(self.dof_names) if per_dof else 1
    self.nominal = _as_vector(nominal, expected_size, f"{name}.nominal")
    self.min_value = _as_vector(min_value, expected_size, f"{name}.min_value")
    self.max_value = _as_vector(max_value, expected_size, f"{name}.max_value")
    if np.any(self.min_value > self.max_value):
        raise ValueError(f"Parameter '{name}' has min_value > max_value.")
    if np.any(self.nominal < self.min_value) or np.any(self.nominal > self.max_value):
        raise ValueError(f"Parameter '{name}' nominal is outside [min, max].")
    self.value = self.nominal.copy()

Attributes

name instance-attribute 
name = name
property instance-attribute 
property = property
entity_name instance-attribute 
entity_name = entity_name
dof_names instance-attribute 
dof_names = tuple(dof_names)
per_dof instance-attribute 
per_dof = per_dof
frozen instance-attribute 
frozen = frozen
nominal instance-attribute 
nominal = _as_vector(nominal, expected_size, f'{name}.nominal')
min_value instance-attribute 
min_value = _as_vector(min_value, expected_size, f'{name}.min_value')
max_value instance-attribute 
max_value = _as_vector(max_value, expected_size, f'{name}.max_value')
value instance-attribute 
value = self.nominal.copy()

Methods:

size 
size() -> int
Source code in eden/extensions/sysid/parameter.py 
73
74
75
@property
def size(self) -> int:
    return int(self.nominal.size)
reset 
reset() -> None
Source code in eden/extensions/sysid/parameter.py 
77
78
def reset(self) -> None:
    self.value = self.nominal.copy()
clip 
clip() -> None
Source code in eden/extensions/sysid/parameter.py 
80
81
def clip(self) -> None:
    np.clip(self.value, self.min_value, self.max_value, out=self.value)
as_dof_vector 
as_dof_vector(n_dofs: int) -> ndarray

Broadcast value to a length-n_dofs vector for set_dofs_.

Source code in eden/extensions/sysid/parameter.py 
83
84
85
86
87
88
89
def as_dof_vector(self, n_dofs: int) -> np.ndarray:
    """Broadcast ``value`` to a length-``n_dofs`` vector for set_dofs_<prop>."""
    if n_dofs != len(self.dof_names):
        raise ValueError(f"Parameter '{self.name}' targets {len(self.dof_names)} DOFs but got n_dofs={n_dofs}.")
    if self.per_dof:
        return self.value.astype(np.float64, copy=False)
    return np.full(n_dofs, float(self.value[0]), dtype=np.float64)

ParameterSet 

ParameterSet(parameters: Sequence[Parameter] | None = None)

Ordered collection of Parameter objects with vectorised access.

Frozen parameters are skipped by as_vector / get_bounds / update_from_vector so the decision-variable dimension seen by the optimiser equals the number of free scalars.

Source code in eden/extensions/sysid/parameter.py 
133
134
135
136
def __init__(self, parameters: Sequence[Parameter] | None = None) -> None:
    self._parameters: dict[str, Parameter] = {}
    for p in parameters or ():
        self.add(p)

Attributes

size property 
size: int

Methods:

add 
add(parameter: Parameter) -> None
Source code in eden/extensions/sysid/parameter.py 
138
139
140
141
def add(self, parameter: Parameter) -> None:
    if parameter.name in self._parameters:
        raise ValueError(f"Duplicate parameter name: {parameter.name!r}")
    self._parameters[parameter.name] = parameter
copy 
copy() -> 'ParameterSet'
Source code in eden/extensions/sysid/parameter.py 
143
144
def copy(self) -> "ParameterSet":
    return copy.deepcopy(self)
values 
values() -> list[Parameter]
Source code in eden/extensions/sysid/parameter.py 
158
159
def values(self) -> list[Parameter]:
    return list(self._parameters.values())
as_vector 
as_vector() -> ndarray
Source code in eden/extensions/sysid/parameter.py 
165
166
167
def as_vector(self) -> np.ndarray:
    free = [p.value.ravel() for p in self if not p.frozen]
    return np.concatenate(free) if free else np.zeros(0, dtype=np.float64)
as_nominal_vector 
as_nominal_vector() -> ndarray
Source code in eden/extensions/sysid/parameter.py 
169
170
171
def as_nominal_vector(self) -> np.ndarray:
    free = [p.nominal.ravel() for p in self if not p.frozen]
    return np.concatenate(free) if free else np.zeros(0, dtype=np.float64)
update_from_vector 
update_from_vector(vector: ndarray) -> None
Source code in eden/extensions/sysid/parameter.py 
173
174
175
176
177
178
179
180
181
182
183
def update_from_vector(self, vector: np.ndarray) -> None:
    vector = np.asarray(vector, dtype=np.float64).ravel()
    if vector.size != self.size:
        raise ValueError(f"Expected vector of size {self.size}, got {vector.size}.")
    start = 0
    for p in self:
        if p.frozen:
            continue
        end = start + p.size
        p.value = vector[start:end].reshape(p.nominal.shape).copy()
        start = end
get_bounds 
get_bounds() -> tuple[ndarray, ndarray]
Source code in eden/extensions/sysid/parameter.py 
185
186
187
188
189
190
191
192
193
194
def get_bounds(self) -> tuple[np.ndarray, np.ndarray]:
    lo, hi = [], []
    for p in self:
        if p.frozen:
            continue
        lo.append(p.min_value.ravel())
        hi.append(p.max_value.ravel())
    if not lo:
        return np.zeros(0), np.zeros(0)
    return np.concatenate(lo), np.concatenate(hi)
reset 
reset() -> None
Source code in eden/extensions/sysid/parameter.py 
196
197
198
def reset(self) -> None:
    for p in self:
        p.reset()
sample 
sample(rng: Generator | None = None) -> ndarray
Source code in eden/extensions/sysid/parameter.py 
200
201
202
203
def sample(self, rng: np.random.Generator | None = None) -> np.ndarray:
    rng = rng or np.random.default_rng()
    lo, hi = self.get_bounds()
    return rng.uniform(lo, hi)
save_yaml 
save_yaml(path: str | Path) -> None
Source code in eden/extensions/sysid/parameter.py 
205
206
207
def save_yaml(self, path: str | pathlib.Path) -> None:
    payload = {p.name: p.__getstate__() for p in self}
    pathlib.Path(path).write_text(yaml.safe_dump(payload, sort_keys=False))
load_yaml classmethod 
load_yaml(path: str | Path) -> 'ParameterSet'
Source code in eden/extensions/sysid/parameter.py 
209
210
211
212
213
214
215
216
217
218
219
@classmethod
def load_yaml(cls, path: str | pathlib.Path) -> "ParameterSet":
    payload = yaml.safe_load(pathlib.Path(path).read_text())
    params: list[Parameter] = []
    for name, state in payload.items():
        p = Parameter.__new__(Parameter)
        p.__setstate__(state)
        if p.name != name:
            p.name = name
        params.append(p)
    return cls(params)

Trajectory container for system identification.

Classes

Trajectory dataclass 

Trajectory(times: ndarray, action: ndarray | None = None, dofs_pos: ndarray | None = None, dofs_vel: ndarray | None = None, dofs_torque: ndarray | None = None, base_quat: ndarray | None = None, base_ang_vel: ndarray | None = None, base_lin_acc: ndarray | None = None, dof_names: tuple[str, ...] = (), initial_state: dict[str, ndarray] = dict(), extra: dict[str, Any] = dict())

Measured-or-simulated trajectory for sysid.

All arrays are 2-D (n_steps, dim) (times is 1-D). Fields absent from the measurement are stored as None and skipped by the residual.

ATTRIBUTE DESCRIPTION
times

Time stamps, shape (n_steps,).

TYPE: ndarray

action

Applied action / command, shape (n_steps, n_action_dim). This is the pre-decimation action that the action manager receives; the rollout feeds it to env.action_manager.compute once per control step.

TYPE: ndarray

dofs_pos, dofs_vel, dofs_torque

Per-DOF measurements, shape (n_steps, n_dofs). In dofs_name order.
base_quat, base_ang_vel, base_lin_acc

Floating-base signals in wxyz / body-frame IMU convention. Each shape (n_steps, 3) (4 for quat). May be None for fixed-base robots.
dof_names

Names in matching column order with dofs_*.

TYPE: tuple[str, ...]

initial_state

Snapshot of the entity state at t=0 for rollout reset. Keys: qpos, dofs_vel, optionally pos, quat.

TYPE: dict[str, ndarray]

extra

Free-form attachments (e.g. fingertip tactile, metadata).

TYPE: dict[str, Any]

Attributes

times instance-attribute 
times: ndarray
action class-attribute instance-attribute 
action: ndarray | None = None
dofs_pos class-attribute instance-attribute 
dofs_pos: ndarray | None = None
dofs_vel class-attribute instance-attribute 
dofs_vel: ndarray | None = None
dofs_torque class-attribute instance-attribute 
dofs_torque: ndarray | None = None
base_quat class-attribute instance-attribute 
base_quat: ndarray | None = None
base_ang_vel class-attribute instance-attribute 
base_ang_vel: ndarray | None = None
base_lin_acc class-attribute instance-attribute 
base_lin_acc: ndarray | None = None
dof_names class-attribute instance-attribute 
dof_names: tuple[str, ...] = ()
initial_state class-attribute instance-attribute 
initial_state: dict[str, ndarray] = field(default_factory=dict)
extra class-attribute instance-attribute 
extra: dict[str, Any] = field(default_factory=dict)
dt property 
dt: float

Methods:

signal 
signal(name: str) -> ndarray | None
Source code in eden/extensions/sysid/trajectory.py 
89
90
def signal(self, name: str) -> np.ndarray | None:
    return getattr(self, name, None)
save 
save(path: str | Path) -> None
Source code in eden/extensions/sysid/trajectory.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def save(self, path: str | pathlib.Path) -> None:
    payload: dict[str, np.ndarray] = {"times": self.times}
    for f in fields(self):
        if f.name in ("times", "dof_names", "initial_state", "extra"):
            continue
        val = getattr(self, f.name)
        if val is not None:
            payload[f.name] = val
    payload["dof_names"] = np.asarray(self.dof_names)
    for k, v in self.initial_state.items():
        payload[f"initial_state/{k}"] = np.asarray(v)
    np.savez(path, **payload)
load classmethod 
load(path: str | Path) -> 'Trajectory'
Source code in eden/extensions/sysid/trajectory.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
@classmethod
def load(cls, path: str | pathlib.Path) -> "Trajectory":
    with np.load(path, allow_pickle=False) as npz:
        keys = set(npz.files)
        times = npz["times"]
        kwargs: dict[str, Any] = {"times": times}
        for name in ("action", "dofs_pos", "dofs_vel", "dofs_torque", "base_quat", "base_ang_vel", "base_lin_acc"):
            if name in keys:
                kwargs[name] = npz[name]
        if "dof_names" in keys:
            kwargs["dof_names"] = tuple(str(s) for s in npz["dof_names"])
        initial_state = {k.split("/", 1)[1]: npz[k] for k in keys if k.startswith("initial_state/")}
        kwargs["initial_state"] = initial_state
    return cls(**kwargs)
from_recorder_episode classmethod 
from_recorder_episode(path: str | Path, dof_names: Sequence[str] | None = None, demo: str | None = None) -> 'Trajectory'

Load a Trajectory from a RecorderManager npz produced by SysIDRecorder.

Reads only the sysid/* key schema. NPZFileHandler writes one or more episodes under demo_{n}/... keys; pass demo="demo_0" to pick a specific episode, or leave it to auto-select the first one. A flat npz whose keys are already under sysid/... (no demo_ prefix) is also supported.

Source code in eden/extensions/sysid/trajectory.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
@classmethod
def from_recorder_episode(
    cls,
    path: str | pathlib.Path,
    dof_names: Sequence[str] | None = None,
    demo: str | None = None,
) -> "Trajectory":
    """Load a Trajectory from a ``RecorderManager`` npz produced by ``SysIDRecorder``.

    Reads only the ``sysid/*`` key schema. ``NPZFileHandler`` writes one or
    more episodes under ``demo_{n}/...`` keys; pass ``demo="demo_0"`` to
    pick a specific episode, or leave it to auto-select the first one.
    A flat npz whose keys are already under ``sysid/...`` (no ``demo_``
    prefix) is also supported.
    """
    with np.load(path, allow_pickle=False) as npz:
        keys = set(npz.files)

        prefix = ""
        flat_stamp = _TIME_KEY
        if flat_stamp not in keys:
            # Recorder-manager output stores episodes under ``demo_N/``.
            demo_names = sorted({k.split("/", 1)[0] for k in keys if k.startswith("demo_")})
            if not demo_names:
                raise ValueError(f"Recorder episode {path} is missing '{_TIME_KEY}'.")
            chosen = demo if demo is not None else demo_names[0]
            if chosen not in demo_names:
                raise ValueError(f"{path} has no episode '{chosen}' (available: {demo_names}).")
            prefix = f"{chosen}/"
            if f"{prefix}{_TIME_KEY}" not in keys:
                raise ValueError(f"Recorder episode {path}:{chosen} is missing '{_TIME_KEY}'.")

        # ``times`` may arrive as (n_steps,), (n_steps, 1), (n_steps, num_envs),
        # or (n_steps, num_envs, 1) depending on which recorder layout produced
        # the npz. Squeeze trailing singletons and pick env 0 explicitly so a
        # blind reshape(-1) can't interleave envs and steps for multi-env runs.
        raw_times = np.asarray(npz[f"{prefix}{_TIME_KEY}"])
        env_axis_size = raw_times.shape[1] if raw_times.ndim >= 2 else 1
        if raw_times.ndim == 1:
            times = raw_times
        elif raw_times.ndim == 2:
            times = raw_times[:, 0] if raw_times.shape[1] >= 1 else raw_times.reshape(-1)
        else:  # ndim >= 3
            times = raw_times[:, 0, 0] if raw_times.shape[1] >= 1 else raw_times.reshape(-1)
        n_steps = int(times.shape[0])

        def _take(key: str) -> np.ndarray | None:
            full = f"{prefix}{key}"
            if full not in keys:
                return None
            arr = np.asarray(npz[full])
            # (n_steps, num_envs, dim) → take env 0; (n_steps, num_envs) where
            # num_envs > 1 → take env 0 column. Single-env (n_steps, 1) layouts
            # fall through unchanged and are 2-D-OK for the Trajectory dataclass.
            if arr.ndim == 3 and arr.shape[0] == n_steps:
                arr = arr[:, 0, :]
            elif arr.ndim == 2 and env_axis_size > 1 and arr.shape == (n_steps, env_axis_size):
                arr = arr[:, 0:1]
            return arr

        kwargs: dict[str, Any] = {"times": times}
        for name, candidates in _SIGNAL_KEYS.items():
            for candidate in candidates:
                arr = _take(candidate)
                if arr is not None:
                    kwargs[name] = arr
                    break
        action = _take(_ACTION_KEY)
        if action is not None:
            kwargs["action"] = action

        initial_state: dict[str, np.ndarray] = {}
        for key in _INITIAL_STATE_KEYS:
            full_key = f"{prefix}sysid/initial_state/{key}"
            if full_key not in keys:
                continue
            arr = np.asarray(npz[full_key])
            if arr.ndim == 2:
                arr = arr[0]
            initial_state[key] = arr
        kwargs["initial_state"] = initial_state

    if dof_names is not None:
        kwargs["dof_names"] = tuple(dof_names)
    return cls(**kwargs)

Optimizers

CMA-ES optimizer for system identification.

Attributes

Classes

CMAES 

CMAES(env: 'EnvBase', options: SystemIdentificationOptions)

Bases: SystemIdentificationBase

CMA-ES identifier for noisy or contact-heavy residuals (e.g. dex hand).

By default evaluates candidates sequentially on the current env — each candidate ask() writes its own parameter vector across all envs and runs one replay per trajectory. This works with any env.num_envs and matches the serial assumption used by SciPyLeastSquares.

To enable batched candidate evaluation set batched=True and build the sysid env with env_options.num_envs == population_size; a single replay then scores the whole population, reducing the per-generation wall-clock by a factor close to the population size.

Extra options

sigma0: float Initial step size in the unit-hypercube reparameterisation. 0.25 explores ~50% of each bound interval on the first generation. population_size: int | None CMA population. Defaults to the library's recommendation 4 + floor(3*log(n)). batched: bool If True, score the whole population in one replay per trajectory; requires env.num_envs == population_size.

Source code in eden/extensions/sysid/base.py 
 97
 98
 99
100
101
102
103
104
def __init__(self, env: "EnvBase", options: SystemIdentificationOptions) -> None:
    super().__init__(options)
    self._env = env
    resolve_named_entity(
        env.entities,
        self.entity_name,
        hint="SystemIdentificationOptions.entity_name names it; set it to one of those.",
    )

Attributes

sigma0 class-attribute instance-attribute 
sigma0: float = 0.25
population_size class-attribute instance-attribute 
population_size: int | None = None
batched class-attribute instance-attribute 
batched: bool = False

Methods:

fit 
fit(params: ParameterSet, trajectories: Sequence[Trajectory]) -> tuple[ParameterSet, FitResult]
Source code in eden/extensions/sysid/optimizers/cmaes.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def fit(
    self,
    params: ParameterSet,
    trajectories: Sequence[Trajectory],
) -> tuple[ParameterSet, FitResult]:
    try:
        import cma
    except ImportError as exc:  # pragma: no cover
        raise RuntimeError("CMAES backend requires the 'cma' package.") from exc

    opt_params = params.copy()
    n = opt_params.size
    if n == 0:
        return opt_params, FitResult(cost=0.0, nfev=0, x=np.zeros(0), message="no free parameters")

    lo, hi = opt_params.get_bounds()
    x0 = opt_params.as_vector()
    x0_scaled = _to_unit(x0, lo, hi)

    pop = self.population_size or (4 + int(3 * np.log(max(n, 1))))
    if self.batched and pop != self._env.num_envs:
        raise ValueError(
            f"CMAES(batched=True) requires env.num_envs == population_size "
            f"(got num_envs={self._env.num_envs}, pop={pop})."
        )

    es = cma.CMAEvolutionStrategy(
        x0_scaled,
        self.sigma0,
        {
            "bounds": [[0.0] * n, [1.0] * n],
            "popsize": pop,
            "maxiter": self.max_iters,
            # Quiet cma's internal chatter — we print our own per-generation line below.
            "verbose": -9,
        },
    )

    history: list[float] = []
    score_fn = self._score_batched if self.batched else self._score_serial
    best_ever = float("inf")
    t_start = time.monotonic()
    if self.verbose:
        mode = "batched" if self.batched else "serial"
        en.logger.info(
            f"[CMAES] n_params={n}, popsize={pop}, mode={mode}, "
            f"num_envs={self._env.num_envs}, max_iters={self.max_iters}"
        )
        en.logger.info(f"{'gen':>4}  {'nfev':>7}  {'best':>12}  {'mean':>12}  {'sigma':>10}  {'elapsed':>8}")
    gen = 0
    while not es.stop():
        t_gen = time.monotonic()
        unit_candidates = np.asarray(es.ask())
        candidates = np.stack([_from_unit(c, lo, hi) for c in unit_candidates])
        if self.verbose:
            en.logger.info(
                f"[gen {gen + 1:>3d}] scoring {candidates.shape[0]} candidates "
                f"over {len(trajectories)} trajectories …"
            )
        costs = score_fn(opt_params, candidates, trajectories)
        es.tell(list(unit_candidates), list(costs.tolist()))
        gen_best = float(np.min(costs))
        history.append(gen_best)
        best_ever = min(best_ever, gen_best)
        gen += 1
        if self.verbose:
            gen_dt = time.monotonic() - t_gen
            elapsed = time.monotonic() - t_start
            en.logger.info(
                f"{gen:>4d}  {int(es.result.evaluations):>7d}  "
                f"{best_ever:>12.4g}  {float(np.mean(costs)):>12.4g}  "
                f"{float(es.sigma):>10.4g}  {elapsed:>6.1f}s  (+{gen_dt:.1f}s)"
            )

    best_unit = np.asarray(es.result.xbest)
    best = _from_unit(best_unit, lo, hi)
    opt_params.update_from_vector(best)

    return opt_params, FitResult(
        cost=float(es.result.fbest),
        nfev=int(es.result.evaluations),
        x=best,
        message="cma completed",
        history=np.asarray(history) if history else None,
    )

Functions:

SciPy least-squares optimizer for system identification.

Attributes

Classes

SciPyLeastSquares 

SciPyLeastSquares(env: 'EnvBase', options: SystemIdentificationOptions)

Bases: SystemIdentificationBase

Trust-Region Reflective least-squares with box bounds.

Operates on a unit-hypercube reparameterisation of the free parameters — decision variables live in [0, 1] regardless of their physical scale — so the finite-difference Jacobian uses a step of diff_step fraction of each parameter's bound range (not sqrt(eps) of its absolute value, which is below the sim's numerical floor for most solver fields). This is the same reparameterisation the CMA-ES backend uses.

Extra options

diff_step: float Relative FD step in unit space. 1e-2 = 1 % of each bound range. Too small → FD underflows to numerical noise and the optimiser terminates at xtol with no progress; too large → the Jacobian becomes a secant rather than a derivative. parallel_fd: bool If True, write n_params + 1 FD columns into as many envs and score them all in one replay per trajectory via :func:batched_candidate_rollout. Requires env.num_envs == n_params + 1 (the batched rollout writes one candidate per env). When False, scipy's serial 2-point FD is used.

Source code in eden/extensions/sysid/base.py 
 97
 98
 99
100
101
102
103
104
def __init__(self, env: "EnvBase", options: SystemIdentificationOptions) -> None:
    super().__init__(options)
    self._env = env
    resolve_named_entity(
        env.entities,
        self.entity_name,
        hint="SystemIdentificationOptions.entity_name names it; set it to one of those.",
    )

Attributes

diff_step class-attribute instance-attribute 
diff_step: float = 0.01
parallel_fd class-attribute instance-attribute 
parallel_fd: bool = False

Methods:

fit 
fit(params: ParameterSet, trajectories: Sequence[Trajectory]) -> tuple[ParameterSet, FitResult]
Source code in eden/extensions/sysid/optimizers/scipy_ls.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def fit(
    self,
    params: ParameterSet,
    trajectories: Sequence[Trajectory],
) -> tuple[ParameterSet, FitResult]:
    opt_params = params.copy()
    n = opt_params.size
    if n == 0:
        return opt_params, FitResult(cost=0.0, nfev=0, x=np.zeros(0), message="no free parameters")

    lo, hi = opt_params.get_bounds()
    span = np.where(hi > lo, hi - lo, 1.0)
    x0 = opt_params.as_vector()
    x0_unit = np.clip((x0 - lo) / span, 0.0, 1.0)

    def _to_physical(x_unit: np.ndarray) -> np.ndarray:
        return lo + np.clip(x_unit, 0.0, 1.0) * span

    def residual_fn(x_unit: np.ndarray) -> np.ndarray:
        opt_params.update_from_vector(_to_physical(x_unit))
        return self.evaluate(opt_params, trajectories)

    jac_arg = "2-point"
    K = n + 1  # center + one perturbation per parameter
    if self.parallel_fd:
        if self._env.num_envs != K:
            raise ValueError(
                f"SciPyLeastSquares(parallel_fd=True) requires env.num_envs == {K} "
                f"(got {self._env.num_envs}). Build the sysid env with num_envs = n_params + 1."
            )
        jac_arg = self._build_parallel_jacobian(opt_params, trajectories, lo, span, K)

    result = least_squares(
        residual_fn,
        x0_unit,
        bounds=(np.zeros(n), np.ones(n)),
        max_nfev=self.max_iters,
        verbose=2 if self.verbose else 0,
        x_scale=1.0,
        diff_step=self.diff_step,
        jac=jac_arg,
    )

    x_star = _to_physical(result.x)
    opt_params.update_from_vector(x_star)

    return opt_params, FitResult(
        cost=float(result.cost),
        nfev=int(result.nfev),
        x=x_star,
        jacobian=getattr(result, "jac", None),
        message=str(result.message),
        extras={
            "optimality": float(result.optimality),
            "status": int(result.status),
            "x_unit": np.asarray(result.x),
        },
    )

Functions:

Excitation signals

Excitation signal generators (chirp, PRBS, playback).

Classes

Excitation

Bases: ABC

Generator of per-DOF position commands used to excite the robot for sysid.

Implementations return a length-num_dofs array at each call. The driver (DeploymentRecorder) adds this to its center pose (the robot's pose at recording start, or center_dofs_pos) before sending to hardware.

Attributes

num_dofs instance-attribute 
num_dofs: int
duration abstractmethod property 
duration: float

Total excitation duration in seconds.

ChirpExcitation 

ChirpExcitation(num_dofs: int, dof_indices: Sequence[int], f_start: float, f_end: float, duration: float, amplitude: float | Sequence[float], stagger_phase: float = 0.0)

Bases: Excitation

Linear-swept sine — the canonical sysid input for linear-system ID.

Produces amplitude * sin(2*pi * phase(t)) where the instantaneous frequency ramps from f_start to f_end over duration. When dof_indices lists multiple DOFs, each is phase-shifted by stagger_phase * (idx / num_active) so simultaneous excitation doesn't create degenerate rank-1 inputs.

Source code in eden/extensions/sysid/excitation.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def __init__(
    self,
    num_dofs: int,
    dof_indices: Sequence[int],
    f_start: float,
    f_end: float,
    duration: float,
    amplitude: float | Sequence[float],
    stagger_phase: float = 0.0,
) -> None:
    self.num_dofs = num_dofs
    self.dof_indices = np.asarray(list(dof_indices), dtype=np.int64)
    self.f_start = float(f_start)
    self.f_end = float(f_end)
    self._duration = float(duration)
    self.stagger_phase = float(stagger_phase)
    amp = np.atleast_1d(np.asarray(amplitude, dtype=np.float64))
    if amp.size == 1:
        amp = np.full(self.dof_indices.size, float(amp[0]))
    if amp.size != self.dof_indices.size:
        raise ValueError("amplitude size must be 1 or len(dof_indices).")
    self.amplitude = amp

Attributes

num_dofs instance-attribute 
num_dofs = num_dofs
dof_indices instance-attribute 
dof_indices = np.asarray(list(dof_indices), dtype=np.int64)
f_start instance-attribute 
f_start = float(f_start)
f_end instance-attribute 
f_end = float(f_end)
stagger_phase instance-attribute 
stagger_phase = float(stagger_phase)
amplitude instance-attribute 
amplitude = amp
duration property 
duration: float

Methods:

PRBSExcitation 

PRBSExcitation(num_dofs: int, dof_indices: Sequence[int], period: float, duration: float, amplitude: float | Sequence[float], seed: int = 0)

Bases: Excitation

Pseudo-random binary sequence — good for stiction / friction characterisation.

Every period seconds each DOF's command takes a fresh random sign, +amplitude or -amplitude. The DOFs draw independent sequences (from one seed), which avoids correlated excitation.

Source code in eden/extensions/sysid/excitation.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def __init__(
    self,
    num_dofs: int,
    dof_indices: Sequence[int],
    period: float,
    duration: float,
    amplitude: float | Sequence[float],
    seed: int = 0,
) -> None:
    self.num_dofs = num_dofs
    self.dof_indices = np.asarray(list(dof_indices), dtype=np.int64)
    self.period = float(period)
    self._duration = float(duration)
    amp = np.atleast_1d(np.asarray(amplitude, dtype=np.float64))
    if amp.size == 1:
        amp = np.full(self.dof_indices.size, float(amp[0]))
    if amp.size != self.dof_indices.size:
        raise ValueError("amplitude size must be 1 or len(dof_indices).")
    self.amplitude = amp
    n_flips = max(int(np.ceil(self._duration / max(self.period, 1e-6))) + 1, 1)
    rng = np.random.default_rng(seed)
    self._flips = rng.choice([-1.0, 1.0], size=(n_flips, self.dof_indices.size))

Attributes

num_dofs instance-attribute 
num_dofs = num_dofs
dof_indices instance-attribute 
dof_indices = np.asarray(list(dof_indices), dtype=np.int64)
period instance-attribute 
period = float(period)
amplitude instance-attribute 
amplitude = amp
duration property 
duration: float

Methods:

PlaybackExcitation 

PlaybackExcitation(offsets: ndarray, dt: float)

Bases: Excitation

Replay a pre-recorded trajectory (e.g. captured earlier in sim).

offsets is (n_steps, num_dofs) of offsets from the center pose sampled at dt. At arbitrary t the nearest earlier sample is returned (zero-order hold) — robust to clock skew between generator and deployment loop.

Source code in eden/extensions/sysid/excitation.py 
137
138
139
140
141
142
143
def __init__(self, offsets: np.ndarray, dt: float) -> None:
    offsets = np.asarray(offsets, dtype=np.float64)
    if offsets.ndim != 2:
        raise ValueError("offsets must be 2-D (n_steps, num_dofs).")
    self.num_dofs = int(offsets.shape[1])
    self._offsets = offsets
    self._dt = float(dt)

Attributes

num_dofs instance-attribute 
num_dofs = int(offsets.shape[1])
duration property 
duration: float

Methods:

Rollouts, residuals and applying parameters

Rollout helpers for replaying and evaluating system-id candidates.

Classes

Functions:

reset_to_initial_state 

reset_to_initial_state(env: 'EnvBase', trajectory: Trajectory, entity_name: str) -> None

Reset all envs to the measurement's initial state by writing entity state directly.

Bypasses _reset_idx on purpose: firing _RESET events would re-run domain-randomisation terms that may mutate the very parameters being identified. The caller is responsible for first placing the sysid parameters via :func:apply_parameters / :func:apply_candidates.

For floating-base entities, qpos from a sim-side recording already includes the 7-DOF base prefix (because :class:SysIDRecorder writes it using entity.qs_idx_local). Deployment recordings, however, store DOF positions only — pos and quat come through as separate initial_state keys, and we apply them after set_qpos so the base actually starts at the measured pose.

Source code in eden/extensions/sysid/rollout.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def reset_to_initial_state(env: "EnvBase", trajectory: Trajectory, entity_name: str) -> None:
    """Reset all envs to the measurement's initial state by writing entity state directly.

    Bypasses ``_reset_idx`` on purpose: firing ``_RESET`` events would re-run
    domain-randomisation terms that may mutate the very parameters being
    identified. The caller is responsible for first placing the sysid
    parameters via :func:`apply_parameters` / :func:`apply_candidates`.

    For floating-base entities, ``qpos`` from a sim-side recording already
    includes the 7-DOF base prefix (because :class:`SysIDRecorder` writes it
    using ``entity.qs_idx_local``). Deployment recordings, however, store
    DOF positions only — ``pos`` and ``quat`` come through as separate
    ``initial_state`` keys, and we apply them after ``set_qpos`` so the
    base actually starts at the measured pose.
    """
    entity = env.entities[entity_name]
    if "qpos" not in trajectory.initial_state:
        raise ValueError("Trajectory.initial_state missing required 'qpos' field.")

    qpos = torch.as_tensor(trajectory.initial_state["qpos"], dtype=torch.float32, device=env.device)
    qpos = qpos.unsqueeze(0).expand(env.num_envs, -1).contiguous()
    entity.set_qpos(qpos, zero_velocity=True)

    if "dofs_vel" in trajectory.initial_state:
        dofs_vel = torch.as_tensor(trajectory.initial_state["dofs_vel"], dtype=torch.float32, device=env.device)
        dofs_vel = dofs_vel.unsqueeze(0).expand(env.num_envs, -1).contiguous()
        entity.set_dofs_vel(dofs_vel)

    # Apply base pose if it was recorded separately. Entities with no base DOFs don't accept
    # set_pos/set_quat — fixed-base ones reject it, and a welded child's pose is FK output that
    # discards the write — so guard on has_immobile_base when available.
    has_immobile_base = bool(getattr(entity, "has_immobile_base", False))
    if not has_immobile_base:
        if "pos" in trajectory.initial_state:
            pos = torch.as_tensor(trajectory.initial_state["pos"], dtype=torch.float32, device=env.device)
            if pos.ndim == 1:
                pos = pos.unsqueeze(0).expand(env.num_envs, -1).contiguous()
            entity.set_pos(pos)
        if "quat" in trajectory.initial_state:
            quat = torch.as_tensor(trajectory.initial_state["quat"], dtype=torch.float32, device=env.device)
            if quat.ndim == 1:
                quat = quat.unsqueeze(0).expand(env.num_envs, -1).contiguous()
            entity.set_quat(quat)

replay_rollout 

replay_rollout(env: 'EnvBase', trajectory: Trajectory, entity_name: str, signals: Sequence[str] = ('dofs_pos', 'dofs_vel', 'dofs_torque')) -> dict[str, Tensor]

Replay the measured action trajectory under the currently-applied parameters.

Uses the narrow physics-only stepper so that termination, reward, reset, command, event, and recorder managers do not fire — a measured trajectory longer than max_episode_length would otherwise be corrupted by auto-reset.

All env.num_envs envs step on every tick (Genesis steps the full batch). For K-candidate batched identification, build a dedicated env with env_options.num_envs = K and use :func:batched_candidate_rollout; otherwise all envs receive the same state (but may hold different parameters for Monte-Carlo variance estimation).

Returns a dict {signal_name: tensor(num_envs, n_steps, dim)}.

Source code in eden/extensions/sysid/rollout.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def replay_rollout(
    env: "EnvBase",
    trajectory: Trajectory,
    entity_name: str,
    signals: Sequence[str] = ("dofs_pos", "dofs_vel", "dofs_torque"),
) -> dict[str, torch.Tensor]:
    """Replay the measured action trajectory under the currently-applied parameters.

    Uses the narrow physics-only stepper so that termination, reward,
    reset, command, event, and recorder managers do not fire — a measured
    trajectory longer than ``max_episode_length`` would otherwise be
    corrupted by auto-reset.

    All ``env.num_envs`` envs step on every tick (Genesis steps the full
    batch). For K-candidate batched identification, build a dedicated env
    with ``env_options.num_envs = K`` and use :func:`batched_candidate_rollout`;
    otherwise all envs receive the same state (but may hold different
    parameters for Monte-Carlo variance estimation).

    Returns a dict ``{signal_name: tensor(num_envs, n_steps, dim)}``.
    """
    if trajectory.action is None:
        raise ValueError("Trajectory.action is required for replay.")
    for s in signals:
        if s not in _SUPPORTED_SIGNALS:
            raise ValueError(f"Unsupported rollout signal: {s!r}. Expected one of {_SUPPORTED_SIGNALS}.")

    entity = env.entities[entity_name]
    n_steps = len(trajectory)
    signal_dims = {
        "dofs_pos": entity.num_dofs,
        "dofs_vel": entity.num_dofs,
        "dofs_torque": entity.num_dofs,
        "base_quat": 4,
        "base_ang_vel": 3,
    }
    buffers: dict[str, torch.Tensor] = {
        s: torch.zeros(env.num_envs, n_steps, signal_dims[s], device=env.device) for s in signals
    }

    reset_to_initial_state(env, trajectory, entity_name)

    actions = torch.as_tensor(trajectory.action, dtype=torch.float32, device=env.device)
    for t in range(n_steps):
        action_batch = actions[t].unsqueeze(0).expand(env.num_envs, -1).contiguous()
        _physics_step(env, action_batch)
        if "dofs_pos" in buffers:
            buffers["dofs_pos"][:, t] = entity.get_dofs_pos()
        if "dofs_vel" in buffers:
            buffers["dofs_vel"][:, t] = entity.get_dofs_vel()
        if "dofs_torque" in buffers:
            buffers["dofs_torque"][:, t] = entity.get_dofs_control_force()
        if "base_quat" in buffers:
            buffers["base_quat"][:, t] = entity.get_quat()
        if "base_ang_vel" in buffers:
            buffers["base_ang_vel"][:, t] = entity.get_ang(frame="body")

    return buffers

single_candidate_rollout 

single_candidate_rollout(env: 'EnvBase', params: ParameterSet, trajectory: Trajectory, entity_name: str, signals: Sequence[str] = ('dofs_pos', 'dofs_vel', 'dofs_torque')) -> dict[str, ndarray]

Apply every free parameter's value to all envs, replay once, return env-0 predictions.

This is the serial primitive used by every optimiser that evaluates one parameter vector per step (scipy least-squares with FD Jacobian, and CMA-ES unless batched mode is opted into).

Source code in eden/extensions/sysid/rollout.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def single_candidate_rollout(
    env: "EnvBase",
    params: ParameterSet,
    trajectory: Trajectory,
    entity_name: str,
    signals: Sequence[str] = ("dofs_pos", "dofs_vel", "dofs_torque"),
) -> dict[str, np.ndarray]:
    """Apply every free parameter's ``value`` to all envs, replay once, return env-0 predictions.

    This is the serial primitive used by every optimiser that evaluates one
    parameter vector per step (scipy least-squares with FD Jacobian, and
    CMA-ES unless batched mode is opted into).
    """
    apply_parameters(env, params)
    pred = replay_rollout(env, trajectory, entity_name=entity_name, signals=signals)
    return {k: v[0].detach().cpu().numpy() for k, v in pred.items()}

batched_candidate_rollout 

batched_candidate_rollout(env: 'EnvBase', params: ParameterSet, candidates: ndarray, trajectory: Trajectory, entity_name: str, signals: Sequence[str] = ('dofs_pos', 'dofs_vel', 'dofs_torque')) -> dict[str, ndarray]

Evaluate K candidates in one rollout on a K-sized env.

Requires env.num_envs == K. Each candidate row is written into the matching env via per-env DOF setters; a single replay then produces predictions for all K candidates simultaneously.

Returns {signal_name: array(shape=(K, n_steps, dim))}.

Source code in eden/extensions/sysid/rollout.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def batched_candidate_rollout(
    env: "EnvBase",
    params: ParameterSet,
    candidates: np.ndarray,
    trajectory: Trajectory,
    entity_name: str,
    signals: Sequence[str] = ("dofs_pos", "dofs_vel", "dofs_torque"),
) -> dict[str, np.ndarray]:
    """Evaluate K candidates in one rollout on a K-sized env.

    Requires ``env.num_envs == K``. Each candidate row is written into the
    matching env via per-env DOF setters; a single replay then produces
    predictions for all K candidates simultaneously.

    Returns ``{signal_name: array(shape=(K, n_steps, dim))}``.
    """
    K = int(candidates.shape[0])
    if K != env.num_envs:
        raise ValueError(
            f"batched_candidate_rollout requires env.num_envs == K (got K={K}, "
            f"num_envs={env.num_envs}). Build the sysid env with "
            f"env_options.num_envs = K or use single_candidate_rollout."
        )
    apply_candidates(env, params, candidates)
    pred = replay_rollout(env, trajectory, entity_name=entity_name, signals=signals)
    return {k: v.detach().cpu().numpy() for k, v in pred.items()}

Signal-residual functions for system-identification objectives.

Classes

Functions:

signal_residual 

signal_residual(predicted: Mapping[str, ndarray], measured: Trajectory, signals: Sequence[str], weights: Mapping[str, float] | None = None, normalize: bool = True) -> ndarray

Flatten per-signal residuals into a single 1-D vector.

predicted[s] has shape (n_steps, dim). The measured value is read from measured.signal(s); the two are compared over the shorter of their lengths. A signal missing from either side is skipped. Each block is multiplied by its weight and, when normalize=True, divided by its measured RMS so heterogeneous units contribute comparably.

Source code in eden/extensions/sysid/residual.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def signal_residual(
    predicted: Mapping[str, np.ndarray],
    measured: Trajectory,
    signals: Sequence[str],
    weights: Mapping[str, float] | None = None,
    normalize: bool = True,
) -> np.ndarray:
    """Flatten per-signal residuals into a single 1-D vector.

    ``predicted[s]`` has shape ``(n_steps, dim)``. The measured value is read
    from ``measured.signal(s)``; the two are compared over the shorter of
    their lengths. A signal missing from either side is skipped. Each block is
    multiplied by its weight and, when ``normalize=True``, divided by its
    measured RMS so heterogeneous units contribute comparably.
    """
    weights = weights or {}
    pieces: list[np.ndarray] = []
    for s in signals:
        meas = measured.signal(s)
        if meas is None:
            continue
        pred = predicted.get(s)
        if pred is None:
            continue
        n = min(pred.shape[0], meas.shape[0])
        diff = pred[:n] - meas[:n]
        w = float(weights.get(s, 1.0))
        if w != 1.0:
            diff = diff * w
        if normalize:
            rms = float(np.sqrt(np.mean(meas[:n] ** 2)))
            if rms > 1e-8:
                diff = diff / rms
        pieces.append(diff.ravel())
    if not pieces:
        raise ValueError("No overlapping signals between predicted and measured.")
    return np.concatenate(pieces)

multi_trajectory_residual 

multi_trajectory_residual(predicted_trajectories: Sequence[Mapping[str, ndarray]], measured_trajectories: Sequence[Trajectory], signals: Sequence[str], weights: Mapping[str, float] | None = None, normalize: bool = True) -> ndarray

Concatenate residuals from multiple (predicted, measured) pairs.

Source code in eden/extensions/sysid/residual.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def multi_trajectory_residual(
    predicted_trajectories: Sequence[Mapping[str, np.ndarray]],
    measured_trajectories: Sequence[Trajectory],
    signals: Sequence[str],
    weights: Mapping[str, float] | None = None,
    normalize: bool = True,
) -> np.ndarray:
    """Concatenate residuals from multiple (predicted, measured) pairs."""
    if len(predicted_trajectories) != len(measured_trajectories):
        raise ValueError("predicted and measured trajectory lists have different lengths.")
    blocks = [
        signal_residual(p, m, signals=signals, weights=weights, normalize=normalize)
        for p, m in zip(predicted_trajectories, measured_trajectories)
    ]
    return np.concatenate(blocks)

Apply identified parameters / candidates to entities.

Attributes

DOF_PROPERTY_SETTERS module-attribute 

DOF_PROPERTY_SETTERS: dict[str, str] = {'damping': 'set_dofs_damping', 'armature': 'set_dofs_armature', 'stiffness': 'set_dofs_stiffness', 'frictionloss': 'set_dofs_frictionloss', 'kp': 'set_dofs_kp', 'kd': 'set_dofs_kd'}

DOF_PROPERTY_GETTERS module-attribute 

DOF_PROPERTY_GETTERS: dict[str, str] = {'damping': 'get_dofs_damping', 'armature': 'get_dofs_armature', 'stiffness': 'get_dofs_stiffness', 'frictionloss': 'get_dofs_frictionloss', 'kp': 'get_dofs_kp', 'kd': 'get_dofs_kd'}

Classes

Functions:

make_parameter_from_default 

make_parameter_from_default(env: 'EnvBase', name: str, property: str, dof_names: Sequence[str], entity_name: str = 'robot', min_scale: float = 0.1, max_scale: float = 10.0, per_dof: bool = False, frozen: bool = False) -> Parameter

Construct a Parameter using the entity's current value as the nominal.

Reads the current per-DOF value (env 0) via the entity getter and uses it as nominal — per DOF with per_dof, else their mean. Bounds are min_scale / max_scale times nominal, the natural parameterisation for the supported non-negative properties (damping, armature, stiffness, frictionloss, kp, kd). When the default is zero (e.g. frictionloss), the multiplicative bounds collapse — construct the Parameter explicitly in that case.

Source code in eden/extensions/sysid/modifier.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def make_parameter_from_default(
    env: "EnvBase",
    name: str,
    property: str,
    dof_names: Sequence[str],
    entity_name: str = "robot",
    min_scale: float = 0.1,
    max_scale: float = 10.0,
    per_dof: bool = False,
    frozen: bool = False,
) -> Parameter:
    """Construct a Parameter using the entity's current value as the nominal.

    Reads the current per-DOF value (env 0) via the entity getter and uses it
    as ``nominal`` — per DOF with ``per_dof``, else their mean. Bounds are
    ``min_scale`` / ``max_scale`` times nominal, the natural parameterisation
    for the supported non-negative properties (``damping``, ``armature``,
    ``stiffness``, ``frictionloss``, ``kp``, ``kd``). When the default is zero
    (e.g. ``frictionloss``), the multiplicative bounds collapse — construct the
    ``Parameter`` explicitly in that case.
    """
    if property not in DOF_PROPERTY_GETTERS:
        raise ValueError(f"Unknown DOF property: {property!r}. Expected one of {list(DOF_PROPERTY_GETTERS)}.")
    entity = env.entities[entity_name]
    _, dof_indices = entity.find_named_dofs_idx_local(list(dof_names), preserve_order=True)
    # Env 0 is the nominal: the getters carry an env axis, and nothing has randomized yet.
    values = getattr(entity, DOF_PROPERTY_GETTERS[property])(dofs_idx_local=dof_indices)[0]
    values_np = values.detach().cpu().numpy().astype(np.float64, copy=False)
    if per_dof:
        nominal = values_np
        lo = nominal * min_scale
        hi = nominal * max_scale
    else:
        nominal_scalar = float(values_np.mean())
        nominal = np.array([nominal_scalar], dtype=np.float64)
        lo = np.array([nominal_scalar * min_scale], dtype=np.float64)
        hi = np.array([nominal_scalar * max_scale], dtype=np.float64)
    return Parameter(
        name=name,
        property=property,
        dof_names=dof_names,
        entity_name=entity_name,
        per_dof=per_dof,
        nominal=nominal,
        min_value=lo,
        max_value=hi,
        frozen=frozen,
    )

apply_parameters 

apply_parameters(env: 'EnvBase', params: ParameterSet) -> None

Write every free parameter's current value to all envs.

Broadcasts the parameter vector to (num_envs, n_dofs) so the solver tensors are consistent across the batch.

Source code in eden/extensions/sysid/modifier.py 
85
86
87
88
89
90
91
92
93
94
def apply_parameters(env: "EnvBase", params: ParameterSet) -> None:
    """Write every free parameter's current value to all envs.

    Broadcasts the parameter vector to ``(num_envs, n_dofs)`` so the
    solver tensors are consistent across the batch.
    """
    for param in params:
        if param.frozen:
            continue
        _apply_one(env, param, envs_idx=None, per_env_values=None)

apply_candidates 

apply_candidates(env: 'EnvBase', params: ParameterSet, candidates: ndarray) -> None

Write K distinct candidate parameter vectors into K environments.

candidates has shape (K, n_free_params) and requires K == env.num_envs. Each row is written to env k as the per-env value of the targeted solver tensor. Frozen parameters are untouched.

Source code in eden/extensions/sysid/modifier.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def apply_candidates(env: "EnvBase", params: ParameterSet, candidates: np.ndarray) -> None:
    """Write K distinct candidate parameter vectors into K environments.

    ``candidates`` has shape ``(K, n_free_params)`` and requires
    ``K == env.num_envs``. Each row is written to env ``k`` as the per-env
    value of the targeted solver tensor. Frozen parameters are untouched.
    """
    K, n_free = candidates.shape
    if n_free != params.size:
        raise ValueError(f"Candidates width {n_free} != ParameterSet.size {params.size}.")
    if K != env.num_envs:
        raise ValueError(f"apply_candidates requires K == env.num_envs (got K={K}, num_envs={env.num_envs}).")

    original = params.as_vector()
    try:
        per_env = [candidates[k] for k in range(K)]
        start = 0
        for param in params:
            if param.frozen:
                continue
            end = start + param.size
            values = np.stack([vec[start:end] for vec in per_env])
            start = end
            _apply_one(env, param, envs_idx=None, per_env_values=values)
    finally:
        params.update_from_vector(original)

Recording on the robot

Recorder that captures real-robot trajectories for system identification.

Classes

DeploymentRecorder 

DeploymentRecorder(deployer: 'DeploymentBase', excitation: Excitation, kp_scale: float = 1.0, kd_scale: float = 1.0, safety_limits: ndarray | None = None, center_dofs_pos: ndarray | None = None, dofs_limit: tuple[ndarray, ndarray] | None = None)

Drive a DeploymentBase with an Excitation and record a Trajectory.

The loop writes the excitation offset on top of a center pose at every control tick, captures the returned RobotState, and after the duration serialises everything to a Trajectory compatible with the sysid rollout loader.

PARAMETER DESCRIPTION
deployer

Active deployment backend, connected and already at its start pose (an INIT that has finished: reset(), request_init(), then step(None) until init_done). The recorder writes the wire directly (send_payload / read_state), outside the operator state machine.

TYPE: 'DeploymentBase'

excitation

Excitation generator. Called once per control tick with the elapsed seconds since run was invoked.

TYPE: Excitation

kp_scale

Multipliers on deployer.default_dof_kp / ..._kd used for this recording. Lower kp is safer and surfaces compliance/damping in the response; higher kp stiffens the reference-tracking.

TYPE: float DEFAULT: 1.0

kd_scale

Multipliers on deployer.default_dof_kp / ..._kd used for this recording. Lower kp is safer and surfaces compliance/damping in the response; higher kp stiffens the reference-tracking.

TYPE: float DEFAULT: 1.0

safety_limits

Optional per-DOF symmetric clamp on the excitation offset (radians). Defaults to no additional clamp beyond whatever the backend already enforces.

TYPE: ndarray | None DEFAULT: None

center_dofs_pos

Optional absolute joint-position center for the excitation. Defaults to the robot's actual position at recording start.

TYPE: ndarray | None DEFAULT: None

dofs_limit

Optional per-DOF lower and upper position limits used for diagnostics.

TYPE: tuple[ndarray, ndarray] | None DEFAULT: None

Source code in eden/extensions/sysid/deployment_recorder.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def __init__(
    self,
    deployer: "DeploymentBase",
    excitation: Excitation,
    kp_scale: float = 1.0,
    kd_scale: float = 1.0,
    safety_limits: np.ndarray | None = None,
    center_dofs_pos: np.ndarray | None = None,
    dofs_limit: tuple[np.ndarray, np.ndarray] | None = None,
) -> None:
    if excitation.num_dofs != deployer.num_dofs:
        raise ValueError(f"Excitation has {excitation.num_dofs} DOFs but deployer expects {deployer.num_dofs}.")
    self.deployer = deployer
    self.excitation = excitation
    self.kp = deployer.default_dof_kp * float(kp_scale)
    self.kd = deployer.default_dof_kd * float(kd_scale)
    self.safety_limits = np.asarray(safety_limits, dtype=np.float64) if safety_limits is not None else None
    if self.safety_limits is not None and self.safety_limits.size != deployer.num_dofs:
        raise ValueError("safety_limits size must equal deployer.num_dofs.")
    self.center_dofs_pos = np.asarray(center_dofs_pos, dtype=np.float64) if center_dofs_pos is not None else None
    if self.center_dofs_pos is not None and self.center_dofs_pos.size != deployer.num_dofs:
        raise ValueError("center_dofs_pos size must equal deployer.num_dofs.")
    if dofs_limit is None:
        self.dofs_limit = None
    else:
        lo, hi = dofs_limit
        self.dofs_limit = (np.asarray(lo, dtype=np.float64), np.asarray(hi, dtype=np.float64))
        if self.dofs_limit[0].size != deployer.num_dofs or self.dofs_limit[1].size != deployer.num_dofs:
            raise ValueError("dofs_limit arrays must equal deployer.num_dofs.")
    # Own our own rate limiter rather than reusing the deployer's, so that
    # data collection can run at its own cadence and does not race with
    # whatever other loop may hold the deployer's limiter.
    self._rate = RateLimiter(deployer.control_freq) if deployer.control_freq else None

Attributes

deployer instance-attribute 
deployer = deployer
excitation instance-attribute 
excitation = excitation
kp instance-attribute 
kp = deployer.default_dof_kp * float(kp_scale)
kd instance-attribute 
kd = deployer.default_dof_kd * float(kd_scale)
safety_limits instance-attribute 
safety_limits = np.asarray(safety_limits, dtype=np.float64) if safety_limits is not None else None
center_dofs_pos instance-attribute 
center_dofs_pos = np.asarray(center_dofs_pos, dtype=np.float64) if center_dofs_pos is not None else None
dofs_limit instance-attribute 
dofs_limit = None

Methods:

run 
run() -> Trajectory

Execute the excitation for excitation.duration seconds.

By default, the excitation offset is added on top of the robot's actual resting pose (read_state().dofs_pos). Callers can pass center_dofs_pos to use a known center pose instead, such as the midpoint of each joint's position limits.

Returns the recorded Trajectory. The timestamp column starts at zero (relative to loop start); the caller can shift externally if absolute alignment with other logs is needed.

Source code in eden/extensions/sysid/deployment_recorder.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def run(self) -> Trajectory:
    """Execute the excitation for ``excitation.duration`` seconds.

    By default, the excitation offset is added on top of the robot's
    **actual resting pose** (``read_state().dofs_pos``). Callers can pass
    ``center_dofs_pos`` to use a known center pose instead, such as the
    midpoint of each joint's position limits.

    Returns the recorded Trajectory. The timestamp column starts at
    zero (relative to loop start); the caller can shift externally if
    absolute alignment with other logs is needed.
    """
    deployer = self.deployer
    duration = self.excitation.duration

    stamps: list[float] = []
    actions: list[np.ndarray] = []
    dofs_pos: list[np.ndarray] = []
    dofs_vel: list[np.ndarray] = []
    dofs_torque: list[np.ndarray] = []
    base_quat: list[np.ndarray] = []
    base_ang_vel: list[np.ndarray] = []
    base_lin_acc: list[np.ndarray] = []

    initial_state = deployer.read_state()
    q0 = np.asarray(initial_state.dofs_pos, dtype=np.float64).copy()
    center = q0 if self.center_dofs_pos is None else self.center_dofs_pos

    # Warn if the requested excitation range leaves the known joint limits.
    try:
        if self.dofs_limit is None:
            entity = deployer._env.entities[deployer.entity_name]
            q_lim_lo, q_lim_hi = entity.get_dofs_limit()
            q_lim_lo = q_lim_lo[0].detach().cpu().numpy()
            q_lim_hi = q_lim_hi[0].detach().cpu().numpy()
        else:
            q_lim_lo, q_lim_hi = self.dofs_limit
        # Peak excitation magnitude per DOF, sampled at 64 points over the duration (best-effort).
        peak = np.zeros(deployer.num_dofs, dtype=np.float64)
        for t in np.linspace(0.0, duration, 64):
            peak = np.maximum(peak, np.abs(self.excitation(t)))
        if self.safety_limits is not None:
            peak = np.minimum(peak, self.safety_limits)
        below = (center - peak) < q_lim_lo
        above = (center + peak) > q_lim_hi
        if np.any(below | above):
            names = list(deployer.dofs_name)
            for i in range(deployer.num_dofs):
                if below[i] or above[i]:
                    en.logger.warning(
                        f"DOF '{names[i]}' excitation range "
                        f"[{center[i] - peak[i]:+.3f}, {center[i] + peak[i]:+.3f}] "
                        f"exceeds URDF limits [{q_lim_lo[i]:+.3f}, {q_lim_hi[i]:+.3f}]. "
                        f"The real hand will clip at the mechanical stop and this joint "
                        f"will be hard to identify. Lower --dofs-range-ratio."
                    )
    except Exception as exc:  # pragma: no cover — diagnostic, don't crash collection
        en.logger.debug(f"joint-limit check skipped: {exc!r}")

    t0 = time.monotonic()
    while True:
        t = time.monotonic() - t0
        if t >= duration:
            break

        offset = self.excitation(t)
        if self.safety_limits is not None:
            offset = np.clip(offset, -self.safety_limits, self.safety_limits)
        target = center + offset

        cmd = RobotCommand(
            dofs_pos=target,
            dofs_vel=np.zeros(deployer.num_dofs),
            dofs_torque=np.zeros(deployer.num_dofs),
            dofs_kp=self.kp,
            dofs_kd=self.kd,
        )
        deployer.send_payload(cmd)
        if self._rate is not None:
            self._rate.sleep()
        state = deployer.read_state()

        stamps.append(t)
        actions.append(target.astype(np.float64, copy=True))
        dofs_pos.append(np.asarray(state.dofs_pos, dtype=np.float64))
        dofs_vel.append(np.asarray(state.dofs_vel, dtype=np.float64))
        dofs_torque.append(np.asarray(state.dofs_torque, dtype=np.float64))
        if state.base_quat is not None:
            base_quat.append(np.asarray(state.base_quat, dtype=np.float64))
        if state.base_ang_vel is not None:
            base_ang_vel.append(np.asarray(state.base_ang_vel, dtype=np.float64))
        if state.base_lin_acc is not None:
            base_lin_acc.append(np.asarray(state.base_lin_acc, dtype=np.float64))

    traj = Trajectory(
        times=np.asarray(stamps),
        action=_stack_or_none(actions),
        dofs_pos=_stack_or_none(dofs_pos),
        dofs_vel=_stack_or_none(dofs_vel),
        dofs_torque=_stack_or_none(dofs_torque),
        base_quat=_stack_or_none(base_quat),
        base_ang_vel=_stack_or_none(base_ang_vel),
        base_lin_acc=_stack_or_none(base_lin_acc),
        dof_names=tuple(deployer.dofs_name),
        initial_state={"qpos": q0, "dofs_vel": np.zeros_like(q0)},
    )
    return traj

Reporting

Reporting helpers for system identification (tables, plots, summary).

Classes

Functions:

parameter_table 

parameter_table(initial: ParameterSet, identified: ParameterSet, digits: int = 4) -> str

Human-readable comparison of initial vs identified parameters.

Source code in eden/extensions/sysid/report.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def parameter_table(
    initial: ParameterSet,
    identified: ParameterSet,
    digits: int = 4,
) -> str:
    """Human-readable comparison of initial vs identified parameters."""
    lines = [f"{'name':<32}{'initial':>14}{'identified':>14}{'rel Δ':>10}"]
    for p in identified:
        init = initial[p.name].value
        for i in range(p.size):
            label = p.name if p.size == 1 else f"{p.name}[{i}]"
            v0, v1 = float(init.ravel()[i]), float(p.value.ravel()[i])
            rel = (v1 - v0) / v0 if abs(v0) > 1e-12 else float("nan")
            lines.append(f"{label:<32}{v0:>14.{digits}g}{v1:>14.{digits}g}{rel * 100:>9.1f}%")
    return "\n".join(lines)

signal_plots 

signal_plots(predicted: Mapping[str, ndarray] | Mapping[str, Mapping[str, ndarray]], measured: Trajectory, signals: Sequence[str], save_dir: str | Path, dof_names: Sequence[str] | None = None) -> list[Path]

Write per-signal matplotlib plots comparing one or more predictions against measurement.

predicted accepts two forms: - Single trace: {signal_name: array(n_steps, dim)} — one dashed "predicted" line is plotted alongside the measurement. - Multiple named traces: {label: {signal_name: array(...)}} — each label is drawn with its own line style, e.g. {"initial": pred0, "identified": pred1} to visualise the pre- vs post-optimisation fit on the same axes.

Each signal gets one figure with one subplot per DOF column. Requires matplotlib; returns the list of written file paths.

Source code in eden/extensions/sysid/report.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def signal_plots(
    predicted: Mapping[str, np.ndarray] | Mapping[str, Mapping[str, np.ndarray]],
    measured: Trajectory,
    signals: Sequence[str],
    save_dir: str | pathlib.Path,
    dof_names: Sequence[str] | None = None,
) -> list[pathlib.Path]:
    """Write per-signal matplotlib plots comparing one or more predictions against measurement.

    ``predicted`` accepts two forms:
    - **Single trace**: ``{signal_name: array(n_steps, dim)}`` — one
      dashed "predicted" line is plotted alongside the measurement.
    - **Multiple named traces**: ``{label: {signal_name: array(...)}}``
      — each label is drawn with its own line style, e.g.
      ``{"initial": pred0, "identified": pred1}`` to visualise the
      pre- vs post-optimisation fit on the same axes.

    Each signal gets one figure with one subplot per DOF column. Requires
    matplotlib; returns the list of written file paths.
    """
    try:
        import matplotlib.pyplot as plt
    except ImportError as exc:  # pragma: no cover
        raise RuntimeError("signal_plots requires matplotlib.") from exc

    save_dir = pathlib.Path(save_dir)
    save_dir.mkdir(parents=True, exist_ok=True)

    times = measured.times
    dof_names = tuple(dof_names) if dof_names is not None else measured.dof_names

    # Normalise predicted -> dict[label, dict[signal, array]].
    first_value = next(iter(predicted.values())) if predicted else None
    is_multi = isinstance(first_value, Mapping)
    if is_multi:
        traces: dict[str, Mapping[str, np.ndarray]] = dict(predicted)  # type: ignore[assignment]
    else:
        traces = {"predicted": predicted}  # type: ignore[dict-item]

    # Measurement = dotted reference. Predictions = continuous; semantically
    # meaningful labels get fixed colors so pre/post-fit are instantly
    # readable across runs. Unknown labels fall back to the default cycle.
    label_colors: dict[str, str] = {
        "initial": "lightblue",
        "baseline": "lightblue",
        "identified": "red",
    }
    written: list[pathlib.Path] = []

    for signal in signals:
        meas = measured.signal(signal)
        signal_traces = [(label, pred.get(signal)) for label, pred in traces.items()]
        signal_traces = [(label, arr) for label, arr in signal_traces if arr is not None]
        if meas is None or not signal_traces:
            continue
        n = min(min(arr.shape[0] for _, arr in signal_traces), meas.shape[0])
        dim = signal_traces[0][1].shape[-1]

        fig, axes = plt.subplots(dim, 1, figsize=(10, 1.2 * max(dim, 2)), sharex=True)
        if dim == 1:
            axes = [axes]
        for j in range(dim):
            axes[j].plot(
                times[:n],
                meas[:n, j],
                label="measured",
                linewidth=1.0,
                linestyle=":",
                color="black",
            )
            for label, pred in signal_traces:
                axes[j].plot(
                    times[:n],
                    pred[:n, j],
                    label=label,
                    linewidth=1.0,
                    linestyle="-",
                    color=label_colors.get(label),
                )
            title = dof_names[j] if dof_names and j < len(dof_names) else f"{signal}[{j}]"
            axes[j].set_ylabel(title, fontsize=8)
        axes[0].legend(loc="upper right", fontsize=8)
        axes[-1].set_xlabel("time [s]")
        fig.suptitle(signal)
        fig.tight_layout()
        out = save_dir / f"{signal}.png"
        fig.savefig(out, dpi=120)
        plt.close(fig)
        written.append(out)
    return written

write_summary 

write_summary(save_dir: str | Path, initial: ParameterSet, identified: ParameterSet, result: 'FitResult') -> None

Write identified parameters as YAML plus a human-readable summary.txt.

Source code in eden/extensions/sysid/report.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def write_summary(
    save_dir: str | pathlib.Path,
    initial: ParameterSet,
    identified: ParameterSet,
    result: "FitResult",
) -> None:
    """Write identified parameters as YAML plus a human-readable summary.txt."""
    save_dir = pathlib.Path(save_dir)
    save_dir.mkdir(parents=True, exist_ok=True)

    identified.save_yaml(save_dir / "params_identified.yaml")
    initial.save_yaml(save_dir / "params_initial.yaml")

    table = parameter_table(initial, identified)
    header = f"cost={result.cost:.6g} nfev={result.nfev} {result.message}".strip()
    (save_dir / "summary.txt").write_text(f"{header}\n\n{table}\n")

See also