Factories

shinro.factories

---

ControllerFactory

Source code in src/shinro/factories/controller_factory.py

def __init__(self, config_path: str | None = None, config: dict | None = None):
    if (config_path is None) == (config is None):
        raise ValueError("ControllerFactory requires exactly one of config_path or config.")
    if config_path is not None:
        with open(resolve_config_path(config_path), "rb") as f:
            config = tomllib.load(f)
    self.config = config

create

create(backend: ArrayBackend = None, plant=None, derive_model: bool = False)

Construct the registered controller from the config.

Parameters:

Name	Type	Description	Default
backend	ArrayBackend	Array backend for the controller.	None
plant		Optional plant for derived-value injection — dt is filled from plant.dt when the config omits it (loud error on disagreement) and, with derive_model, A_dynamics/B_dynamics are derived when both are absent.	None
derive_model	bool	Derive the model from the plant (plant-only mode).	False

Source code in src/shinro/factories/controller_factory.py

def create(self, backend: ArrayBackend = None, plant=None, derive_model: bool = False):
    """Construct the registered controller from the config.

    Args:
        backend: Array backend for the controller.
        plant: Optional plant for derived-value injection — ``dt`` is
            filled from ``plant.dt`` when the config omits it (loud error
            on disagreement) and, with ``derive_model``,
            ``A_dynamics``/``B_dynamics`` are derived when both are absent.
        derive_model: Derive the model from the plant (plant-only mode).
    """
    cls = _CONTROLLER_REGISTRY[self.config["type"]]
    if plant is not None:
        return cls.from_config(cls.load_config(self.config, plant=plant, derive_model=derive_model), backend=backend)
    return cls.from_config(self.config, backend=backend)

---

EstimatorFactory

Source code in src/shinro/factories/estimator_factory.py

def __init__(self, config_path: str | None = None, config: dict | None = None):
    if (config_path is None) == (config is None):
        raise ValueError("EstimatorFactory requires exactly one of config_path or config.")
    if config_path is not None:
        with open(resolve_config_path(config_path), "rb") as f:
            config = tomllib.load(f)
    self.config = config

create

create(backend: ArrayBackend = None, plant=None, derive_model: bool = False)

Construct the registered estimator from the config.

Parameters:

Name	Type	Description	Default
backend	ArrayBackend	Array backend for the estimator.	None
plant		Optional plant for derived-value injection — dt is filled from plant.dt when the config omits it (loud error on disagreement) and, with derive_model, A_dynamics/B_dynamics are derived when both are absent.	None
derive_model	bool	Derive the model from the plant (plant-only mode).	False

Source code in src/shinro/factories/estimator_factory.py

def create(self, backend: ArrayBackend = None, plant=None, derive_model: bool = False):
    """Construct the registered estimator from the config.

    Args:
        backend: Array backend for the estimator.
        plant: Optional plant for derived-value injection — ``dt`` is
            filled from ``plant.dt`` when the config omits it (loud error
            on disagreement) and, with ``derive_model``,
            ``A_dynamics``/``B_dynamics`` are derived when both are absent.
        derive_model: Derive the model from the plant (plant-only mode).
    """
    cls = _ESTIMATOR_REGISTRY[self.config["type"]]
    if plant is not None:
        return cls.from_config(cls.load_config(self.config, plant=plant, derive_model=derive_model), backend=backend)
    return cls.from_config(self.config, backend=backend)

---

TrajectoryFactory

Source code in src/shinro/factories/trajectory_factory.py

def __init__(self, config_path: str):
    with open(resolve_config_path(config_path), "rb") as f:
        self.config = tomllib.load(f)

---

Scenario

Fully composed control loop plus scenario parameters.

sim is the RobotSim instance (engine + plants) for MuJoCo- backed scenarios, or None for plant-only scenarios where the plant self-integrates its analytical dynamics. plant is the primary plant driven by the controller; controller, estimator and trajectory are the other loop roles. For feedforward scenarios (e.g. phase_list pick-and-place) controller and estimator are None and the schedule itself is the control. config is the raw TOML dict (used by the runner and the tests for tolerances, noise, etc.).

A scenario is a runnable simulation: run executes the whole loop (closed-loop or feedforward, chosen by whether controller is set), driven entirely by the TOML — duration, dt, noise, adversarial faults, input limits and tolerances. iter_run yields one shinro.simulation.runner.StepRecord per step for live rendering or early stopping; reset restores the TOML initial state and clears controller/estimator state.

run

run(steps: int | None = None, seed: int | None = None)

Run the scenario to completion.

Parameters:

Name	Type	Description	Default
steps	int | None	Number of steps. Defaults to [scenario].duration / dt.	None
seed	int | None	Noise RNG seed. Overrides [noise.measurement].seed.	None

Returns:

Type	Description
	A shinro.simulation.runner.SimResult — one
	shinro.simulation.runner.StepRecord per step, plus
	check() tolerance reporting against the scenario TOML.

Source code in src/shinro/factories/scenario_factory.py

def run(self, steps: int | None = None, seed: int | None = None):
    """Run the scenario to completion.

    Args:
        steps: Number of steps. Defaults to ``[scenario].duration / dt``.
        seed: Noise RNG seed. Overrides ``[noise.measurement].seed``.

    Returns:
        A :class:`~shinro.simulation.runner.SimResult` — one
        :class:`~shinro.simulation.runner.StepRecord` per step, plus
        ``check()`` tolerance reporting against the scenario TOML.
    """
    from shinro.simulation.runner import run_phase_schedule, run_scenario

    if self.controller is None:
        return run_phase_schedule(self, steps=steps)
    return run_scenario(self, steps=steps, seed=seed)

iter_run

iter_run(steps: int | None = None, seed: int | None = None)

Iterate over the run, yielding one StepRecord per step.

For live rendering or early stopping; run collects the same records into a shinro.simulation.runner.SimResult.

Source code in src/shinro/factories/scenario_factory.py

def iter_run(self, steps: int | None = None, seed: int | None = None):
    """Iterate over the run, yielding one StepRecord per step.

    For live rendering or early stopping; :meth:`run` collects the same
    records into a :class:`~shinro.simulation.runner.SimResult`.
    """
    from shinro.simulation.runner import iter_phase_schedule, iter_scenario

    if self.controller is None:
        return iter_phase_schedule(self, steps=steps)
    return iter_scenario(self, steps=steps, seed=seed)

reset

reset() -> None

Restore the TOML initial state and clear controller/estimator state.

Deterministic reproduction is scenario.reset(); scenario.run(seed=0).

Source code in src/shinro/factories/scenario_factory.py

def reset(self) -> None:
    """Restore the TOML initial state and clear controller/estimator state.

    Deterministic reproduction is ``scenario.reset(); scenario.run(seed=0)``.
    """
    if self.sim is not None:
        self.sim.reset()
    else:
        init = self.config.get("plant", {}).get("initial_state")
        if init is not None:
            self.plant.state = self.plant.bk.array(init)
        else:
            self.plant.state = self.plant.bk.zeros_like(self.plant.get_state())
    if self.controller is not None:
        self.controller.reset()
    if self.estimator is not None:
        self.estimator.reset()

---

ScenarioFactory

Build a Scenario from a single TOML config.

Config sections

[scenario] name, description, duration, dt, tolerance, input_limits [physics] free_joint, model_path [plant] name (sim-backed) OR type + config + initial_state (plant-only) [controller] type, config (optional) [estimator] type, config (optional) [trajectory] type, config [sim] config (path to the RobotSim TOML; optional) [noise] measurement (optional) [adversarial] inject_at, value (optional)

Two modes:

1. Sim-backed (default): the plant is looked up by [plant].name on the RobotSim built from [sim]. When [physics].free_joint is set, the LeKiwi MJCF is rewritten so the arm hangs off a mobile free-jointed base.

2. Plant-only: when [sim] is absent, the plant is built directly from the registry via [plant].type (registered name) and [plant].config (path to a plant TOML). [plant].initial_state optionally seeds the state. The plant self-integrates its analytical dynamics — no MuJoCo engine. [scenario].dt must equal the plant's own dt (the plant integrates at its own time step).

The plant TOML is the single source of physics truth. Controller/estimator configs never need to restate it: dt is filled from plant.dt when omitted (a declared dt that disagrees is a loud error), and in plant-only mode an omitted A_dynamics/B_dynamics model is derived from the plant via shinro.utils.linearization.linearize_plant (upright equilibrium) and first-order Euler discretization.

Source code in src/shinro/factories/scenario_factory.py

def __init__(self, config_path: str):
    self.config_path = config_path
    with open(resolve_config_path(config_path), "rb") as f:
        self.config = tomllib.load(f)

build

build(backend: ArrayBackend | None = None) -> Scenario

Build and validate the full scenario.

Parameters:

Name	Type	Description	Default
backend	ArrayBackend | None	Array backend for controller/estimator/trajectory. Defaults to numpy (the physics engine is always numpy-backed).	None

Returns:

Type	Description
Scenario	A composed Scenario.

Raises:

Type	Description
KeyError	If a required section or registry type is missing.
ValueError	If component dimensions disagree with the plant.

Source code in src/shinro/factories/scenario_factory.py

def build(self, backend: ArrayBackend | None = None) -> Scenario:
    """Build and validate the full scenario.

    Args:
        backend: Array backend for controller/estimator/trajectory.
            Defaults to numpy (the physics engine is always numpy-backed).

    Returns:
        A composed :class:`Scenario`.

    Raises:
        KeyError: If a required section or registry type is missing.
        ValueError: If component dimensions disagree with the plant.
    """
    plant_cfg = self.config["plant"]
    sim_cfg = self.config.get("sim", {"config": "robot_config.toml"})
    physics_cfg = self.config.get("physics", {})

    if "sim" in self.config:
        sim, plant = self._build_sim_plant(plant_cfg, sim_cfg, physics_cfg)
        derive_model = False
        scenario_dt = self.config.get("scenario", {}).get("dt")
        if scenario_dt is not None:
            engine_dt = float(sim.engine.dt)
            ratio = float(scenario_dt) / engine_dt
            if abs(ratio - round(ratio)) > 1e-9 or ratio < 1:
                raise ValueError(
                    f"[scenario].dt ({scenario_dt}) must be an integer multiple of "
                    f"[engine].dt ({engine_dt}): the plant integrates every "
                    f"{round(ratio) if ratio >= 1 else '<1'} engine step(s)."
                )
    else:
        sim, plant = self._build_plant_only(plant_cfg)
        derive_model = True
        scenario_dt = self.config.get("scenario", {}).get("dt")
        if scenario_dt is not None and abs(float(scenario_dt) - float(plant.dt)) > 1e-12:
            raise ValueError(
                f"[scenario].dt ({scenario_dt}) must equal the plant's dt ({plant.dt}) "
                "in plant-only scenarios: the plant self-integrates at its own time step."
            )

    def _create(factory_cls, path: str):
        if backend is not None:
            return factory_cls(path).create(backend=backend)
        return factory_cls(path).create()

    trajectory = _create(TrajectoryFactory, self.config["trajectory"]["config"])

    controller = None
    estimator = None
    if "controller" in self.config:
        controller = self._create_loop_role(
            ControllerFactory, self.config["controller"]["config"], plant, backend, derive_model
        )
    if "estimator" in self.config:
        estimator = self._create_loop_role(
            EstimatorFactory, self.config["estimator"]["config"], plant, backend, derive_model
        )

    # MPPI is function-based: its dynamics/cost are produced from the
    # plant's model via a batched adapter, so it must be wired after the
    # plant is built (this also sets its D_u for validation). Q/R come
    # from the controller's own config.
    if isinstance(controller, MPPIController):
        controller.attach_plant(plant)

    if controller is not None and estimator is not None:
        n_x = plant.get_state().shape[0]
        _, B = plant.get_model()
        n_u = B.shape[1]
        self._validate_dimensions(n_x, n_u, controller, estimator)

    return Scenario(
        sim=sim,
        plant=plant,
        controller=controller,
        estimator=estimator,
        trajectory=trajectory,
        config=self.config,
    )

---

register_controller

Source code in src/shinro/factories/registry.py

def register_controller(name):
    def decorator(cls):
        # TODO: strict=True once onnx_rl / lerobot_diffusion adapters are rewritten.
        _check_config(cls, "Controller", name, strict=False)
        cls._registry_name = name
        _CONTROLLER_REGISTRY[name] = cls
        return cls

    return decorator

---

register_estimator

Source code in src/shinro/factories/registry.py

def register_estimator(name):
    def decorator(cls):
        _check_config(cls, "Estimator", name, strict=True)
        cls._registry_name = name
        _ESTIMATOR_REGISTRY[name] = cls
        return cls

    return decorator

---

register_trajectory

Source code in src/shinro/factories/registry.py

def register_trajectory(name):
    def decorator(cls):
        _check_config(cls, "Trajectory", name, strict=True)
        cls._registry_name = name
        _TRAJECTORY_REGISTRY[name] = cls
        return cls

    return decorator

---

register_plant

Source code in src/shinro/factories/registry.py

def register_plant(name):
    def decorator(cls):
        _check_config(cls, "Plant", name, strict=True)
        cls._registry_name = name
        _PLANT_REGISTRY[name] = cls
        return cls

    return decorator

---

_CONTROLLER_REGISTRY

---

_ESTIMATOR_REGISTRY

---

_TRAJECTORY_REGISTRY

---

_PLANT_REGISTRY