Scenario API

skelarm.scenario

Load a full control scenario (robot + task + simulator + controller) from one TOML file.

A combined config mirrors the per-section schema used by :meth:Skeleton.from_toml: [skeleton] and [initial] describe the robot and its start state, [task] the goal (for a reach, a task-space target with a movement duration), [simulator] the integration settings, and [controller] which controller drives the motion. :func:load_scenario reads them all.

Example::

[task]
type = "reaching"  # the task kind (required)
target = [0.55, 1.21]  # endpoint goal (x, y) in meters (required for reaching)
duration = 2.0  # simulated time (s)

[simulator]
dt = 0.002  # simulation and control step (s)

[controller]
type = "computed_torque"
kp = 200.0
kd = 30.0

Scenario dataclass

A loaded control scenario: the robot, the task, the simulator, and the controller.

source_config keeps the original combined config (the parsed [skeleton] / [initial] / [task] / [simulator] / [controller] tables) when the scenario was loaded from a file, so a run can embed it for an exact, editable re-run; it is None for scenarios built programmatically.

Source code in src/skelarm/scenario.py

@dataclass
class Scenario:
    """A loaded control scenario: the robot, the task, the simulator, and the controller.

    ``source_config`` keeps the original combined config (the parsed ``[skeleton]`` /
    ``[initial]`` / ``[task]`` / ``[simulator]`` / ``[controller]`` tables) when the
    scenario was loaded from a file, so a run can embed it for an exact, editable re-run;
    it is ``None`` for scenarios built programmatically.
    """

    skeleton: Skeleton
    task: Task
    controller: Controller
    simulator: Simulator = field(default_factory=Simulator)
    source_config: Mapping[str, Any] | None = None

Simulator dataclass

Numerical-integration parameters for the dynamics, from the [simulator] table.

These are run/execution concerns, kept separate from the task (the desired motion).

Parameters:

Name	Type	Description	Default
dt	float	Control / integration step (seconds). Also the MPC prediction step.	0.002
enforce_limits	bool	Apply the joint limits as hard stops in the dynamics (default). When False, the limits constrain only the kinematics (posing and inverse kinematics), not the simulated motion.	True

Source code in src/skelarm/scenario.py

@dataclass
class Simulator:
    """Numerical-integration parameters for the dynamics, from the ``[simulator]`` table.

    These are run/execution concerns, kept separate from the task (the desired motion).

    Parameters
    ----------
    dt : float
        Control / integration step (seconds). Also the MPC prediction step.
    enforce_limits : bool
        Apply the joint limits as hard stops in the dynamics (default). When ``False``,
        the limits constrain only the kinematics (posing and inverse kinematics), not the
        simulated motion.
    """

    dt: float = 0.002
    enforce_limits: bool = True

    @classmethod
    def from_dict(cls, section: Mapping[str, Any]) -> Simulator:
        """Build a Simulator from a ``[simulator]`` mapping (defaults when keys are absent)."""
        return cls(
            dt=float(section.get("dt", 0.002)),
            enforce_limits=bool(section.get("enforce_limits", True)),
        )

from_dict(section) classmethod

Build a Simulator from a [simulator] mapping (defaults when keys are absent).

Source code in src/skelarm/scenario.py

@classmethod
def from_dict(cls, section: Mapping[str, Any]) -> Simulator:
    """Build a Simulator from a ``[simulator]`` mapping (defaults when keys are absent)."""
    return cls(
        dt=float(section.get("dt", 0.002)),
        enforce_limits=bool(section.get("enforce_limits", True)),
    )

Task dataclass

A task and the run conditions for it.

The required field is type — the kind of task, which determines what else is needed. reaching requires a target. The other built-in types (multi_target_reaching, periodic_curve, trajectory_tracking, and joint_trajectory_tracking) and custom types registered with :func:register_task_type carry their own data on :attr:params and may omit the target.

Parameters:

Name	Type	Description	Default
type	str	The task kind (the [task].type config key). Required.	required
target	NDArray[float64] | None	The endpoint goal (x, y) in meters. Required for reaching; None for task types that do not use a single point goal.	None
label	str | None	Optional name for the target (for multi_target_reaching, the active candidate's label; the candidate itself is chosen by the active index).	None
color	str	Marker color for the target (any Qt/SVG color name or #rrggbb).	_DEFAULT_TARGET_COLOR
tolerance	float | None	Optional success distance in meters: the reach succeeds when the tip is within this distance of the target, and the target marker's ring is drawn at this radius. None applies no success criterion.	None
duration	float	Total simulated time / planned-motion horizon (seconds).	2.0
schedule	str	Time-scaling schedule for planned-trajectory controllers.	'minimum_jerk'
params	dict[str, Any]	Any extra [task] keys not recognized above, kept verbatim. A custom task type carries its own parameters here for a matching controller builder to read (e.g. task.params["radius"]).	dict()

Source code in src/skelarm/scenario.py

@dataclass
class Task:
    """A task and the run conditions for it.

    The required field is ``type`` — the kind of task, which determines what else is
    needed. ``reaching`` requires a ``target``. The other built-in types
    (``multi_target_reaching``, ``periodic_curve``, ``trajectory_tracking``, and
    ``joint_trajectory_tracking``) and custom types registered with
    :func:`register_task_type` carry their own data on :attr:`params` and may omit the
    target.

    Parameters
    ----------
    type : str
        The task kind (the ``[task].type`` config key). Required.
    target : NDArray[np.float64] | None
        The endpoint goal ``(x, y)`` in meters. Required for ``reaching``; ``None``
        for task types that do not use a single point goal.
    label : str | None
        Optional name for the target (for ``multi_target_reaching``, the active
        candidate's label; the candidate itself is chosen by the ``active`` index).
    color : str
        Marker color for the target (any Qt/SVG color name or ``#rrggbb``).
    tolerance : float | None
        Optional success distance in meters: the reach succeeds when the tip is
        within this distance of the target, and the target marker's ring is drawn
        at this radius. ``None`` applies no success criterion.
    duration : float
        Total simulated time / planned-motion horizon (seconds).
    schedule : str
        Time-scaling schedule for planned-trajectory controllers.
    params : dict[str, Any]
        Any extra ``[task]`` keys not recognized above, kept verbatim. A custom
        task type carries its own parameters here for a matching controller builder
        to read (e.g. ``task.params["radius"]``).
    """

    type: str  # the task kind (the [task].type config key); the required discriminator
    target: NDArray[np.float64] | None = None
    label: str | None = None
    color: str = _DEFAULT_TARGET_COLOR
    tolerance: float | None = None
    duration: float = 2.0
    schedule: str = "minimum_jerk"
    params: dict[str, Any] = field(default_factory=dict)

    def require_target(self) -> NDArray[np.float64]:
        """Return the task target, raising if this task type has none.

        Use this from a controller that needs a point goal (e.g. the reaching
        controllers), so a task wired without a target fails with a clear message.

        Raises
        ------
        ValueError
            If the task has no target.
        """
        if self.target is None:
            msg = f"task type {self.type!r} has no target, but this controller requires one"
            raise ValueError(msg)
        return self.target

    def to_dict(self) -> dict[str, Any]:
        """Return the ``[task]`` mapping that :meth:`from_dict` builds this task from.

        The inverse of :meth:`from_dict`: the target becomes a table with its
        position, color, and (when set) label and tolerance; the duration, the
        schedule, and the :attr:`params` follow. Values are plain lists, numbers,
        and strings, so the mapping can be written to TOML or stored in a log.

        Returns
        -------
        dict[str, Any]
            The ``[task]`` table.
        """
        section: dict[str, Any] = {"type": self.type}
        if self.target is not None:
            target: dict[str, Any] = {"pos": [float(value) for value in self.target], "color": self.color}
            if self.label is not None:
                target["label"] = self.label
            if self.tolerance is not None:
                target["tolerance"] = self.tolerance
            section["target"] = target
        section["duration"] = self.duration
        section["schedule"] = self.schedule
        return section | self.params

    @classmethod
    def from_toml(cls, file_path: str | Path) -> Task:
        """Build a Task from the ``[task]`` table of a TOML file."""
        with Path(file_path).open("rb") as f:
            data = tomllib.load(f)
        if "task" not in data:
            msg = f"no [task] section in {file_path}"
            raise ValueError(msg)
        return cls.from_dict(data["task"])

    @classmethod
    def from_dict(cls, section: Mapping[str, Any]) -> Task:
        """Build a Task from a ``[task]`` mapping.

        ``type`` is required; it selects the task kind. The ``reaching`` type also
        requires a ``target`` — either a ``[x, y]`` array (position only) or a table
        with a ``pos`` and optional ``label`` / ``color`` / ``tolerance``. Other task
        types may omit the target. Any keys beyond the recognized ones are kept on
        :attr:`params` for custom task types.

        Raises
        ------
        ValueError
            If ``type`` is missing or unknown, or a ``reaching`` task has no/ill-shaped
            ``target``.
        """
        if "type" not in section:
            msg = "[task] requires a 'type' (e.g. type = \"reaching\")"
            raise ValueError(msg)
        task_type = str(section["type"])
        if task_type not in _TASK_TYPES:
            msg = f"unknown task type {task_type!r}; choose from {sorted(_TASK_TYPES)} (or register_task_type)"
            raise ValueError(msg)
        moved = [key for key in _MOVED_TO_SIMULATOR if key in section]
        if moved:
            msg = f"[task] keys {moved} moved to the [simulator] table; set them under [simulator] instead"
            raise ValueError(msg)

        pos: NDArray[np.float64] | None
        if "target" in section:
            pos, label, color, tolerance = _parse_target(section["target"])
        elif task_type == _REACHING_TYPE:
            msg = "[task] of type 'reaching' requires a 'target' endpoint"
            raise ValueError(msg)
        else:
            pos, label, color, tolerance = None, None, _DEFAULT_TARGET_COLOR, None

        params = {key: value for key, value in section.items() if key not in _KNOWN_TASK_KEYS}
        return cls(
            type=task_type,
            target=pos,
            label=label,
            color=color,
            tolerance=tolerance,
            duration=float(section.get("duration", 2.0)),
            schedule=str(section.get("schedule", "minimum_jerk")),
            params=params,
        )

from_dict(section) classmethod

Build a Task from a [task] mapping.

type is required; it selects the task kind. The reaching type also requires a target — either a [x, y] array (position only) or a table with a pos and optional label / color / tolerance. Other task types may omit the target. Any keys beyond the recognized ones are kept on :attr:params for custom task types.

Raises:

Type	Description
ValueError	If type is missing or unknown, or a reaching task has no/ill-shaped target.

Source code in src/skelarm/scenario.py

@classmethod
def from_dict(cls, section: Mapping[str, Any]) -> Task:
    """Build a Task from a ``[task]`` mapping.

    ``type`` is required; it selects the task kind. The ``reaching`` type also
    requires a ``target`` — either a ``[x, y]`` array (position only) or a table
    with a ``pos`` and optional ``label`` / ``color`` / ``tolerance``. Other task
    types may omit the target. Any keys beyond the recognized ones are kept on
    :attr:`params` for custom task types.

    Raises
    ------
    ValueError
        If ``type`` is missing or unknown, or a ``reaching`` task has no/ill-shaped
        ``target``.
    """
    if "type" not in section:
        msg = "[task] requires a 'type' (e.g. type = \"reaching\")"
        raise ValueError(msg)
    task_type = str(section["type"])
    if task_type not in _TASK_TYPES:
        msg = f"unknown task type {task_type!r}; choose from {sorted(_TASK_TYPES)} (or register_task_type)"
        raise ValueError(msg)
    moved = [key for key in _MOVED_TO_SIMULATOR if key in section]
    if moved:
        msg = f"[task] keys {moved} moved to the [simulator] table; set them under [simulator] instead"
        raise ValueError(msg)

    pos: NDArray[np.float64] | None
    if "target" in section:
        pos, label, color, tolerance = _parse_target(section["target"])
    elif task_type == _REACHING_TYPE:
        msg = "[task] of type 'reaching' requires a 'target' endpoint"
        raise ValueError(msg)
    else:
        pos, label, color, tolerance = None, None, _DEFAULT_TARGET_COLOR, None

    params = {key: value for key, value in section.items() if key not in _KNOWN_TASK_KEYS}
    return cls(
        type=task_type,
        target=pos,
        label=label,
        color=color,
        tolerance=tolerance,
        duration=float(section.get("duration", 2.0)),
        schedule=str(section.get("schedule", "minimum_jerk")),
        params=params,
    )

from_toml(file_path) classmethod

Build a Task from the [task] table of a TOML file.

Source code in src/skelarm/scenario.py

@classmethod
def from_toml(cls, file_path: str | Path) -> Task:
    """Build a Task from the ``[task]`` table of a TOML file."""
    with Path(file_path).open("rb") as f:
        data = tomllib.load(f)
    if "task" not in data:
        msg = f"no [task] section in {file_path}"
        raise ValueError(msg)
    return cls.from_dict(data["task"])

require_target()

Return the task target, raising if this task type has none.

Use this from a controller that needs a point goal (e.g. the reaching controllers), so a task wired without a target fails with a clear message.

Raises:

Type	Description
ValueError	If the task has no target.

Source code in src/skelarm/scenario.py

def require_target(self) -> NDArray[np.float64]:
    """Return the task target, raising if this task type has none.

    Use this from a controller that needs a point goal (e.g. the reaching
    controllers), so a task wired without a target fails with a clear message.

    Raises
    ------
    ValueError
        If the task has no target.
    """
    if self.target is None:
        msg = f"task type {self.type!r} has no target, but this controller requires one"
        raise ValueError(msg)
    return self.target

to_dict()

Return the [task] mapping that :meth:from_dict builds this task from.

The inverse of :meth:from_dict: the target becomes a table with its position, color, and (when set) label and tolerance; the duration, the schedule, and the :attr:params follow. Values are plain lists, numbers, and strings, so the mapping can be written to TOML or stored in a log.

Returns:

Type	Description
dict[str, Any]	The [task] table.

Source code in src/skelarm/scenario.py

def to_dict(self) -> dict[str, Any]:
    """Return the ``[task]`` mapping that :meth:`from_dict` builds this task from.

    The inverse of :meth:`from_dict`: the target becomes a table with its
    position, color, and (when set) label and tolerance; the duration, the
    schedule, and the :attr:`params` follow. Values are plain lists, numbers,
    and strings, so the mapping can be written to TOML or stored in a log.

    Returns
    -------
    dict[str, Any]
        The ``[task]`` table.
    """
    section: dict[str, Any] = {"type": self.type}
    if self.target is not None:
        target: dict[str, Any] = {"pos": [float(value) for value in self.target], "color": self.color}
        if self.label is not None:
            target["label"] = self.label
        if self.tolerance is not None:
            target["tolerance"] = self.tolerance
        section["target"] = target
    section["duration"] = self.duration
    section["schedule"] = self.schedule
    return section | self.params

active_target_index(task)

The configured active-target index of a multi-target task (default 0).

Source code in src/skelarm/scenario.py

def active_target_index(task: Task) -> int:
    """The configured active-target index of a multi-target task (default 0)."""
    return int(task.params.get("active", 0))

apply_active_target(task)

Set a multi-target task's first-class target (and marker attrs) to its active candidate.

Lets the active goal flow through the ordinary reaching pipeline (controller build, marker drawing); the GUI switches it live by reassigning task.target / the controller's target.

Raises:

Type	Description
ValueError	If the active index is out of range.

Source code in src/skelarm/scenario.py

def apply_active_target(task: Task) -> None:
    """Set a multi-target task's first-class ``target`` (and marker attrs) to its active candidate.

    Lets the active goal flow through the ordinary reaching pipeline (controller build,
    marker drawing); the GUI switches it live by reassigning ``task.target`` / the
    controller's target.

    Raises
    ------
    ValueError
        If the active index is out of range.
    """
    specs = multi_target_specs(task)
    index = active_target_index(task)
    if not 0 <= index < len(specs):
        msg = f"active target index {index} is out of range for {len(specs)} targets"
        raise ValueError(msg)
    task.target, task.label, task.color, task.tolerance = specs[index]

build_controller(source, *, skeleton, task, dt=0.002)

Construct the controller described by a [controller] config.

Parameters:

Name	Type	Description	Default
source	str | Path | Mapping[str, Any]	A TOML file path (its [controller] table is read) or the table itself.	required
skeleton	Skeleton	The robot, posed at its start state; used to build joint references.	required
task	Task	The reaching task supplying the target and run conditions.	required
dt	float	The control / integration step (from [simulator]). It is injected into the params mapping handed to the builder as params["dt"], so a controller that needs the step (e.g. MPC, whose prediction step must match it) can read it.	0.002

Returns:

Type	Description
Controller	The configured controller.

Raises:

Type	Description
ValueError	If the config has no type or an unknown controller type.

Source code in src/skelarm/scenario.py

def build_controller(
    source: str | Path | Mapping[str, Any],
    *,
    skeleton: Skeleton,
    task: Task,
    dt: float = 0.002,
) -> Controller:
    """Construct the controller described by a ``[controller]`` config.

    Parameters
    ----------
    source : str | Path | Mapping[str, Any]
        A TOML file path (its ``[controller]`` table is read) or the table itself.
    skeleton : Skeleton
        The robot, posed at its start state; used to build joint references.
    task : Task
        The reaching task supplying the target and run conditions.
    dt : float, optional
        The control / integration step (from ``[simulator]``). It is injected into the
        ``params`` mapping handed to the builder as ``params["dt"]``, so a controller that
        needs the step (e.g. MPC, whose prediction step must match it) can read it.

    Returns
    -------
    Controller
        The configured controller.

    Raises
    ------
    ValueError
        If the config has no ``type`` or an unknown controller ``type``.
    """
    config = source if isinstance(source, Mapping) else _read_controller_section(source)
    controller_type = config.get("type")
    if controller_type not in _BUILDERS:
        msg = f"unknown controller type {controller_type!r}; choose from {list(controller_types())}"
        raise ValueError(msg)
    params = {key: value for key, value in config.items() if key != "type"}
    params["dt"] = dt  # the control step, available to controller builders that need it
    return _BUILDERS[controller_type](params, skeleton, task)

controller_types()

Return the registered [controller].type values, sorted.

Source code in src/skelarm/scenario.py

def controller_types() -> tuple[str, ...]:
    """Return the registered ``[controller].type`` values, sorted."""
    return tuple(sorted(_BUILDERS))

export_scenario_toml(log, path)

Write the log's embedded scenario config to an editable TOML file.

The output is a standard combined config ([skeleton] / [initial] / [task] / [simulator] / [controller]) that :func:load_scenario reads back. It is the resolved scenario config the run used (section overrides merged, reference samples inlined — not necessarily the input TOML byte-for-byte): re-running it unedited reproduces a headless, config-driven run exactly for the deterministic controllers, but call-time overrides (a duration= or enforce_limits= argument) live only in the run metadata and are not exported, and an interactive run's mouse-applied forces are not part of the config. Individual values can be edited for comparison studies.

Parameters:

Name	Type	Description	Default
log	StateLog	A log carrying an embedded source config (written by :func:run_scenario or an interactive scenario simulator).	required
path	str | Path	Destination .toml path.	required

Raises:

Type	Description
ValueError	If the log carries no embedded scenario config.

Source code in src/skelarm/scenario.py

def export_scenario_toml(log: StateLog, path: str | Path) -> None:
    """Write the log's embedded scenario config to an editable TOML file.

    The output is a standard combined config (``[skeleton]`` / ``[initial]`` /
    ``[task]`` / ``[simulator]`` / ``[controller]``) that :func:`load_scenario`
    reads back. It is the **resolved scenario config the run used** (section
    overrides merged, reference samples inlined — not necessarily the input TOML
    byte-for-byte): re-running it unedited reproduces a headless, config-driven
    run exactly for the deterministic controllers, but call-time overrides (a
    ``duration=`` or
    ``enforce_limits=`` argument) live only in the run metadata and are not
    exported, and an interactive run's mouse-applied forces are not part of the
    config. Individual values can be edited for comparison studies.

    Parameters
    ----------
    log : StateLog
        A log carrying an embedded source config (written by :func:`run_scenario`
        or an interactive scenario simulator).
    path : str | Path
        Destination ``.toml`` path.

    Raises
    ------
    ValueError
        If the log carries no embedded scenario config.
    """
    config = log.extra.get("source_config")
    if not config:
        msg = "log has no embedded scenario config to export; it was recorded without one"
        raise ValueError(msg)
    Path(path).write_text(dump_toml(config).strip() + "\n", encoding="utf-8")

load_reference_log(path)

Load a .sklog.npz reference trajectory (e.g. from tools/trajectory_recorder.py).

Source code in src/skelarm/scenario.py

def load_reference_log(path: str | Path) -> StateLog:
    """Load a ``.sklog.npz`` reference trajectory (e.g. from ``tools/trajectory_recorder.py``)."""
    file = Path(path)
    if not file.exists():
        msg = f"reference file not found: {file}"
        raise FileNotFoundError(msg)
    return StateLog.load(file)

load_scenario(file_path)

Load a robot, task, and controller from one combined TOML file.

Source code in src/skelarm/scenario.py

def load_scenario(file_path: str | Path) -> Scenario:
    """Load a robot, task, and controller from one combined TOML file."""
    with Path(file_path).open("rb") as f:
        data = tomllib.load(f)
    return scenario_from_config(data)

multi_target_specs(task)

Parse a multi-target task's targets list into (pos, label, color, tolerance) tuples.

Raises:

Type	Description
ValueError	If the task carries no targets list.

Source code in src/skelarm/scenario.py

def multi_target_specs(task: Task) -> list[tuple[NDArray[np.float64], str | None, str, float | None]]:
    """Parse a multi-target task's ``targets`` list into ``(pos, label, color, tolerance)`` tuples.

    Raises
    ------
    ValueError
        If the task carries no ``targets`` list.
    """
    raw = task.params.get("targets")
    if not raw:
        msg = "[task] of type 'multi_target_reaching' requires a non-empty 'targets' list"
        raise ValueError(msg)
    return [_parse_target(entry) for entry in raw]

reference_builders()

Return the task types that provide a joint reference, sorted.

Source code in src/skelarm/scenario.py

def reference_builders() -> tuple[str, ...]:
    """Return the task types that provide a joint reference, sorted."""
    return tuple(sorted(_REFERENCE_BUILDERS))

register_controller(name, builder)

Register a controller builder so [controller].type = name builds it.

The builder receives the [controller] table (without type) as a mapping, the posed Skeleton, and the Task, and returns a :class:~skelarm.Controller. Re-registering an existing name replaces it. Custom controllers are also usable directly (construct the instance and pass it to :func:run_scenario / :func:~skelarm.simulate_controlled); registration is only needed for the config-driven path.

Parameters:

Name	Type	Description	Default
name	str	The [controller].type value to bind.	required
builder	ControllerBuilder	A callable (params, skeleton, task) -> Controller.	required

Source code in src/skelarm/scenario.py

def register_controller(name: str, builder: ControllerBuilder) -> None:
    """Register a controller builder so ``[controller].type = name`` builds it.

    The builder receives the ``[controller]`` table (without ``type``) as a mapping,
    the posed ``Skeleton``, and the ``Task``, and returns a :class:`~skelarm.Controller`.
    Re-registering an existing name replaces it. Custom controllers are also usable
    directly (construct the instance and pass it to :func:`run_scenario` /
    :func:`~skelarm.simulate_controlled`); registration is only needed for the
    config-driven path.

    Parameters
    ----------
    name : str
        The ``[controller].type`` value to bind.
    builder : ControllerBuilder
        A callable ``(params, skeleton, task) -> Controller``.
    """
    _BUILDERS[name] = builder

register_reference_builder(task_type, builder)

Register a joint-reference builder for a [task].type.

A tracking controller (computed torque, joint PD, inverse-dynamics PD, MPC) builds its reference by calling the builder registered for the scenario's task type. Use this to add a new trajectory-style task that drives the existing controllers.

Parameters:

Name	Type	Description	Default
task_type	str	The [task].type value to bind.	required
builder	ReferenceBuilder	A callable (skeleton, task) -> JointReference.	required

Source code in src/skelarm/scenario.py

def register_reference_builder(task_type: str, builder: ReferenceBuilder) -> None:
    """Register a joint-reference builder for a ``[task].type``.

    A tracking controller (computed torque, joint PD, inverse-dynamics PD, MPC) builds
    its reference by calling the builder registered for the scenario's task type. Use
    this to add a new trajectory-style task that drives the existing controllers.

    Parameters
    ----------
    task_type : str
        The ``[task].type`` value to bind.
    builder : ReferenceBuilder
        A callable ``(skeleton, task) -> JointReference``.
    """
    _REFERENCE_BUILDERS[task_type] = builder

register_task_type(name)

Allow name as a [task].type.

Task types are a label that a controller interprets; a custom task carries its parameters on :attr:Task.params (any extra [task] keys). Register the type so :meth:Task.from_dict accepts it, then read task.params from a matching controller built via :func:register_controller.

Parameters:

Name	Type	Description	Default
name	str	The task-type string to allow (idempotent).	required

Source code in src/skelarm/scenario.py

def register_task_type(name: str) -> None:
    """Allow ``name`` as a ``[task].type``.

    Task types are a label that a controller interprets; a custom task carries its
    parameters on :attr:`Task.params` (any extra ``[task]`` keys). Register the type
    so :meth:`Task.from_dict` accepts it, then read ``task.params`` from a matching
    controller built via :func:`register_controller`.

    Parameters
    ----------
    name : str
        The task-type string to allow (idempotent).
    """
    _TASK_TYPES.add(name)

rerun_log(log)

Reconstruct and re-simulate a scenario from its recorded config and run parameters.

Rebuilds the scenario by reparsing the embedded source config (identical input gives identical state) and re-runs with the recorded run parameters. For a headless :func:run_scenario log, deterministic controllers reproduce the recorded channels exactly (MPC, which calls :func:scipy.optimize.minimize, reproduces within a small numerical tolerance on the same platform). For an interactive simulator log the re-run is the unperturbed scenario: mouse-applied forces (the ext_force channel) and live target switches (the active_target channel) are recorded but not replayed, so the re-run reaches the configured active target.

Raises:

Type	Description
ValueError	If the log carries no reproduction metadata.

Source code in src/skelarm/scenario.py

def rerun_log(log: StateLog) -> StateLog:
    """Reconstruct and re-simulate a scenario from its recorded config and run parameters.

    Rebuilds the scenario by reparsing the embedded source config (identical input
    gives identical state) and re-runs with the recorded run parameters. For a
    headless :func:`run_scenario` log, deterministic controllers reproduce the
    recorded channels exactly (MPC, which calls :func:`scipy.optimize.minimize`,
    reproduces within a small numerical tolerance on the same platform). For an
    interactive simulator log the re-run is the *unperturbed* scenario:
    mouse-applied forces (the ``ext_force`` channel) and live target switches
    (the ``active_target`` channel) are recorded but not replayed, so the re-run
    reaches the configured active target.

    Raises
    ------
    ValueError
        If the log carries no reproduction metadata.
    """
    scenario, run = scenario_from_log(log)
    grav_vec = np.asarray(run["grav_vec"], dtype=np.float64)
    # Older logs may predate the recorded enforce_limits; fall back to the simulator's value.
    enforce_limits = run.get("enforce_limits", scenario.simulator.enforce_limits)
    return run_scenario(
        scenario, duration=run["duration"], dt=run["dt"], grav_vec=grav_vec, enforce_limits=enforce_limits
    )

run_scenario(scenario, *, duration=None, dt=None, grav_vec=None, enforce_limits=None)

Simulate a scenario and embed its full config for later reproduction.

Runs :func:~skelarm.control.simulate_controlled and, when the scenario was loaded from a config, stores the task, controller, initial state, and run parameters in the log's [extra] table so :func:rerun_log can reconstruct and re-simulate it.

Parameters:

Name	Type	Description	Default
scenario	Scenario	The robot, task, and controller to run.	required
duration	float | None	Simulated duration (seconds); defaults to scenario.task.duration.	None
dt	float | None	Control / integration step; defaults to scenario.simulator.dt.	None
grav_vec	NDArray[float64] | None	Gravity vector; defaults to zero (planar motion).	None
enforce_limits	bool | None	Apply the joint limits as hard stops in the dynamics. None (the default) uses the scenario's simulator.enforce_limits; an explicit value overrides it (e.g. a --no-joint-limits CLI flag). The resolved value is recorded in the run metadata so the override is reproduced on re-run.	None

Returns:

Type	Description
StateLog	The recorded run, with reproduction metadata when available.

Source code in src/skelarm/scenario.py

def run_scenario(
    scenario: Scenario,
    *,
    duration: float | None = None,
    dt: float | None = None,
    grav_vec: NDArray[np.float64] | None = None,
    enforce_limits: bool | None = None,
) -> StateLog:
    """Simulate a scenario and embed its full config for later reproduction.

    Runs :func:`~skelarm.control.simulate_controlled` and, when the scenario was
    loaded from a config, stores the task, controller, initial state, and run
    parameters in the log's ``[extra]`` table so :func:`rerun_log` can reconstruct
    and re-simulate it.

    Parameters
    ----------
    scenario : Scenario
        The robot, task, and controller to run.
    duration : float | None, optional
        Simulated duration (seconds); defaults to ``scenario.task.duration``.
    dt : float | None, optional
        Control / integration step; defaults to ``scenario.simulator.dt``.
    grav_vec : NDArray[np.float64] | None, optional
        Gravity vector; defaults to zero (planar motion).
    enforce_limits : bool | None, optional
        Apply the joint limits as hard stops in the dynamics. ``None`` (the default)
        uses the scenario's ``simulator.enforce_limits``; an explicit value overrides it
        (e.g. a ``--no-joint-limits`` CLI flag). The resolved value is recorded in
        the run metadata so the override is reproduced on re-run.

    Returns
    -------
    StateLog
        The recorded run, with reproduction metadata when available.
    """
    run_duration = scenario.task.duration if duration is None else float(duration)
    run_dt = scenario.simulator.dt if dt is None else float(dt)
    grav = np.zeros(_TASK_DIM, dtype=np.float64) if grav_vec is None else np.asarray(grav_vec, dtype=np.float64)
    run_enforce_limits = scenario.simulator.enforce_limits if enforce_limits is None else bool(enforce_limits)
    extra = scenario_run_metadata(
        scenario, duration=run_duration, dt=run_dt, grav_vec=grav, enforce_limits=run_enforce_limits
    )
    return simulate_controlled(
        scenario.skeleton,
        scenario.controller,
        duration=run_duration,
        dt=run_dt,
        grav_vec=grav,
        enforce_limits=run_enforce_limits,
        extra=extra,
    )

scenario_from_config(data)

Build a Scenario from a parsed combined config and keep the config for re-runs.

Reuses the real parsers (:meth:Skeleton.from_config, :meth:Task.from_dict, :func:build_controller), so rebuilding from an identical mapping reproduces the robot, initial pose, task, and controller exactly.

Raises:

Type	Description
ValueError	If the config has no [task] or [controller] table.

Source code in src/skelarm/scenario.py

def scenario_from_config(data: Mapping[str, Any]) -> Scenario:
    """Build a Scenario from a parsed combined config and keep the config for re-runs.

    Reuses the real parsers (:meth:`Skeleton.from_config`, :meth:`Task.from_dict`,
    :func:`build_controller`), so rebuilding from an identical mapping reproduces the
    robot, initial pose, task, and controller exactly.

    Raises
    ------
    ValueError
        If the config has no ``[task]`` or ``[controller]`` table.
    """
    if "task" not in data:
        msg = "no [task] section in scenario config"
        raise ValueError(msg)
    if "controller" not in data:
        msg = "no [controller] section in scenario config"
        raise ValueError(msg)
    resolved = _inline_reference_samples(data)
    skeleton = Skeleton.from_config(resolved)
    task = Task.from_dict(resolved["task"])
    if task.type == _MULTI_TARGET_TYPE:
        apply_active_target(task)  # set the first-class target to the active candidate
    simulator = Simulator.from_dict(resolved.get("simulator", {}))
    controller = build_controller(resolved["controller"], skeleton=skeleton, task=task, dt=simulator.dt)
    return Scenario(
        skeleton=skeleton, task=task, controller=controller, simulator=simulator, source_config=dict(resolved)
    )

scenario_from_log(log)

Reconstruct the scenario and run parameters embedded in a log.

Works with logs written by :func:run_scenario and by the interactive scenario simulators; run keys a log does not record (e.g. an interactive run's duration) fall back to the scenario's own values.

Returns:

Type	Description
tuple[Scenario, Mapping[str, Any]]	The rebuilt scenario (skeleton posed at the recorded initial state) and the run parameters (duration / dt / grav_vec / enforce_limits).

Raises:

Type	Description
ValueError	If the log carries no reproduction metadata (it was recorded without an embedded scenario config).

Source code in src/skelarm/scenario.py

def scenario_from_log(log: StateLog) -> tuple[Scenario, Mapping[str, Any]]:
    """Reconstruct the scenario and run parameters embedded in a log.

    Works with logs written by :func:`run_scenario` and by the interactive
    scenario simulators; run keys a log does not record (e.g. an interactive
    run's ``duration``) fall back to the scenario's own values.

    Returns
    -------
    tuple[Scenario, Mapping[str, Any]]
        The rebuilt scenario (skeleton posed at the recorded initial state) and the
        ``run`` parameters (``duration`` / ``dt`` / ``grav_vec`` / ``enforce_limits``).

    Raises
    ------
    ValueError
        If the log carries no reproduction metadata (it was recorded without an
        embedded scenario config).
    """
    config = log.extra.get("source_config")
    if not config:
        msg = "log has no reproduction metadata; it was recorded without an embedded scenario config"
        raise ValueError(msg)
    scenario = scenario_from_config(config)
    # Merge key-by-key: a GUI log records dt / grav_vec / enforce_limits but no
    # duration (the run is open-ended), and older logs may lack the table entirely.
    defaults: dict[str, Any] = {
        "duration": scenario.task.duration,
        "dt": scenario.simulator.dt,
        "grav_vec": [0.0, 0.0],
        "enforce_limits": scenario.simulator.enforce_limits,
    }
    run = {**defaults, **dict(log.extra.get("run", {}))}
    return scenario, run

scenario_run_metadata(scenario, *, dt, enforce_limits, duration=None, grav_vec=None)

Assemble the [extra] payload that lets a run be reconstructed and re-run.

Embeds the resolved scenario config (the editable [skeleton] / [initial] / [task] / [simulator] / [controller] tables as the run used them — section overrides merged and file-based reference samples inlined), the actual run parameters (including the resolved enforce_limits, so a call-time override is captured), and the package versions. Returns None when the scenario carries no source_config (e.g. a programmatically built controller).

Parameters:

Name	Type	Description	Default
scenario	Scenario	The scenario whose source config to embed.	required
dt	float	The control / integration step actually used.	required
enforce_limits	bool	The resolved joint-limit hard-stop setting actually used.	required
duration	float | None	Simulated duration (seconds). None — for open-ended runs such as the interactive simulators — omits the key; :func:scenario_from_log then falls back to the task's duration.	None
grav_vec	NDArray[float64] | None	Gravity vector actually used; None records zero (planar motion).	None

Returns:

Type	Description
dict[str, Any] | None	The [extra] payload, or None without a source config.

Source code in src/skelarm/scenario.py

def scenario_run_metadata(
    scenario: Scenario,
    *,
    dt: float,
    enforce_limits: bool,
    duration: float | None = None,
    grav_vec: NDArray[np.float64] | None = None,
) -> dict[str, Any] | None:
    """Assemble the ``[extra]`` payload that lets a run be reconstructed and re-run.

    Embeds the resolved scenario config (the editable ``[skeleton]`` / ``[initial]`` /
    ``[task]`` / ``[simulator]`` / ``[controller]`` tables as the run used them —
    section overrides merged and file-based reference samples inlined), the
    actual run parameters (including the resolved ``enforce_limits``, so a call-time
    override is captured), and the package versions. Returns ``None`` when the
    scenario carries no ``source_config`` (e.g. a programmatically built controller).

    Parameters
    ----------
    scenario : Scenario
        The scenario whose source config to embed.
    dt : float
        The control / integration step actually used.
    enforce_limits : bool
        The resolved joint-limit hard-stop setting actually used.
    duration : float | None, optional
        Simulated duration (seconds). ``None`` — for open-ended runs such as the
        interactive simulators — omits the key; :func:`scenario_from_log` then
        falls back to the task's duration.
    grav_vec : NDArray[np.float64] | None, optional
        Gravity vector actually used; ``None`` records zero (planar motion).

    Returns
    -------
    dict[str, Any] | None
        The ``[extra]`` payload, or ``None`` without a source config.
    """
    if scenario.source_config is None:
        return None
    grav = np.zeros(_TASK_DIM, dtype=np.float64) if grav_vec is None else np.asarray(grav_vec, dtype=np.float64)
    run: dict[str, Any] = {
        "dt": float(dt),
        "grav_vec": grav.tolist(),
        "enforce_limits": bool(enforce_limits),
    }
    if duration is not None:
        run["duration"] = float(duration)
    return {
        "source_config": dict(scenario.source_config),
        "run": run,
        "provenance": _provenance(),
    }

task_types()

Return the registered [task].type values, sorted.

Source code in src/skelarm/scenario.py

def task_types() -> tuple[str, ...]:
    """Return the registered ``[task].type`` values, sorted."""
    return tuple(sorted(_TASK_TYPES))