# Genesis Forge

> A modular robotics RL training framework built on the Genesis physics simulator.

Genesis Forge is a modular robotics reinforcement learning framework built on the
Genesis physics simulator. It provides a manager-based architecture for constructing
Gymnasium-compatible parallel RL environments.

Install: pip install genesis-forge
Source: https://github.com/jgillick/genesis-forge


# Project

# Genesis Forge

Genesis Forge is a powerful robotics reinforcement learning framework using the [Genesis](https://genesis-world.readthedocs.io/en/latest/) physics simulator. It provides a flexible and modular architecture to get your robot up and running quickly with less boilerplate work.

## RL Robotics, What?

Today, modern robots learn to balance, walk, manipulate objects, and more, using AI/[Reinforcement Learning](https://huggingface.co/learn/deep-rl-course/en/unit1/what-is-rl) algorithms. You simply create a program that defines a task and provides feedback on the robot's performance — much like training a dog with treats and commands. Genesis Forge is a framework that makes this very easy to, with [documentation](https://genesis-forge.readthedocs.io/en/latest/guide/index.html) and [examples](https://github.com/jgillick/genesis-forge/tree/main/examples) to get you started.

## Key Features

- ⚙️ Modular composable environment design
- 💥 Comprehensive contact/collision manager
- 🎬 Automatically record video snippets during training
- 🕹️ Connect a game controller for hands-on policy evaluation and play-mode.
- 🤖 Seamless integration with popular RL libraries: [RSL-RL](https://github.com/leggedrobotics/rsl_rl/tree/main) and [SKRL](https://skrl.readthedocs.io/en/latest/)

[](_static/cmd_locomotion.webm)

## Citation

If you used Genesis Forge in your research, we would appreciate it if you could cite it.

```
@misc{Genesis-Forge,
  author = {Jeremy Gillick},
  title = {Genesis Forge: A modular framework for RL robot environments},
  month = {September},
  year = {2025},
  url = {https://github.com/jgillick/genesis-forge}
}
```
# Environments

# Environments Overview

Your robot environment starts by extending one of these classes. Most often you'll use `ManagedEnvironment`, which gives you built-in access to all the managers.

# GenesisEnv

Base environment class for your simulated robot environment.

Parameters:

| Name                         | Type    | Description                                                                                                                    | Default                            |
| ---------------------------- | ------- | ------------------------------------------------------------------------------------------------------------------------------ | ---------------------------------- |
| `num_envs`                   | `int`   | Number of parallel environments.                                                                                               | `1`                                |
| `dt`                         | `float` | Simulation time step.                                                                                                          | `1 / 100`                          |
| `max_episode_length_sec`     | \`int   | None\`                                                                                                                         | Maximum episode length in seconds. |
| `max_episode_random_scaling` | `float` | Scale the maximum episode length by this amount (+/-) so that not all environments reset at the same time.                     | `0.0`                              |
| `extras_logging_key`         | `str`   | The key used, in info/extras dict, which is returned by step and reset functions, to send data to tensorboard by the RL agent. | `'episode'`                        |

Example::

```
class MyEnv(GenesisEnv):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        # ...Define scene here...
        self.scene = gs.Scene()
        self.terrain = self.scene.add_entity(gs.morphs.Plane())
        self.robot = self.scene.add_entity( ... )

    def step(self, actions: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor, dict[str, Any]]:
        # ...step logic here...
        return obs, rewards, terminations, truncations, info

    def reset(self, envs_idx: list[int] = None) -> tuple[torch.Tensor, dict[str, Any]]:
        # ...reset logic here...
        return obs, info

    def get_observations(self) -> torch.Tensor:
        # ...define current observations here...
        return obs
```

Source code in `genesis_forge/genesis_env.py`

```
def __init__(
    self,
    num_envs: int = 1,
    dt: float = 1 / 100,
    max_episode_length_sec: int | None = 10,
    max_episode_random_scaling: float = 0.0,
    extras_logging_key: str = "episode",
):
    self.dt = dt
    self.device = gs.device
    self.num_envs = num_envs
    self.scene: gs.Scene = None
    self.robot: RigidEntity = None
    self.terrain: RigidEntity = None

    self.extras_logging_key = extras_logging_key
    self._extras = {}
    self._extras[extras_logging_key] = {}

    self._actions: torch.Tensor = None
    self._last_actions: torch.Tensor = None

    self.step_count: int = 0
    self.episode_length = torch.zeros(
        (self.num_envs,), device=gs.device, dtype=torch.int32
    )
    self.max_episode_length: torch.Tensor = None

    self._max_episode_length_sec = 0.0
    self._base_max_episode_length = None
    self._max_episode_random_scaling = max_episode_random_scaling
    if max_episode_length_sec and max_episode_length_sec > 0:
        self.max_episode_length = torch.zeros(
            (self.num_envs,), device=gs.device, dtype=gs.tc_int
        )
        self.max_episode_length[:] = self.set_max_episode_length(
            max_episode_length_sec
        )
```

## `action_space = None`

## `actions`

The actions for each environment for this step. If you're using an action manager, these are the actions prior to being handled by the action manager.

## `can_be_wrapped = True`

## `device = gs.device`

## `dt = dt`

## `episode_length = torch.zeros((self.num_envs,), device=(gs.device), dtype=(torch.int32))`

## `extras`

The extras/infos dictionary reset at the start of every step, and contains additional data about the environment during that step.

## `extras_logging_key = extras_logging_key`

## `last_actions`

The actions for for the previous step.

## `max_episode_length = None`

## `max_episode_length_sec`

The max episode length, in seconds, for each environment.

## `max_episode_length_steps`

The max episode length, in steps, for each environment. If episode randomization scaling is enabled, this will be the base max episode length before scaling.

## `num_actions`

The number of actions for each environment.

## `num_envs = num_envs`

## `num_observations`

The number of observations for each environment.

## `observation_space = None`

## `robot = None`

## `scene = None`

## `step_count = 0`

## `step_dt`

The time step of the environment. This is an alias of the :attr:`dt` property.

## `terrain = None`

## `unwrapped`

Returns this environment, not a wrapped version of it.

## `build()`

Builds the environment before the first step. The Genesis scene and all the scene entities must be added before calling this method.

Source code in `genesis_forge/genesis_env.py`

```
def build(self) -> None:
    """
    Builds the environment before the first step.
    The Genesis scene and all the scene entities must be added before calling this method.
    """
    assert (
        self.scene is not None
    ), "The scene must be constructed and assigned to the <env>.scene attribute before building."
    self.scene.build(n_envs=self.num_envs)
```

## `close()`

Close the environment.

Source code in `genesis_forge/genesis_env.py`

```
def close(self):
    """Close the environment."""
    pass
```

## `get_observations()`

Returns the current observations for each environment. Override this method to return the observations for your environment.

Example::

```
def get_observations(self) -> torch.Tensor:
    return torch.cat(
    [
        self.base_ang_vel * self.obs_scales["ang_vel"],  # 3
        self.projected_gravity,  # 3
        self.commands * self.commands_scale,  # 3
        (self.dof_pos - self.default_dof_pos) * self.obs_scales["dof_pos"],  # 12
        self.dof_vel * self.obs_scales["dof_vel"],  # 12
        self.actions,  # 12
    ],
    axis=-1,
)
```

Source code in `genesis_forge/genesis_env.py`

```
def get_observations(self) -> torch.Tensor:
    """
    Returns the current observations for each environment.
    Override this method to return the observations for your environment.

    Example::

        def get_observations(self) -> torch.Tensor:
            return torch.cat(
            [
                self.base_ang_vel * self.obs_scales["ang_vel"],  # 3
                self.projected_gravity,  # 3
                self.commands * self.commands_scale,  # 3
                (self.dof_pos - self.default_dof_pos) * self.obs_scales["dof_pos"],  # 12
                self.dof_vel * self.obs_scales["dof_vel"],  # 12
                self.actions,  # 12
            ],
            axis=-1,
        )
    """
    if self.observation_space is not None:
        return torch.zeros(
            (self.num_envs, self.observation_space.shape[0]),
            device=gs.device,
            dtype=gs.tc_float,
        )
    return None
```

## `reset(env_ids=None)`

Reset one or more environments. Each of the registered managers will also be reset for those environments.

Parameters:

| Name      | Type        | Description                                                        | Default |
| --------- | ----------- | ------------------------------------------------------------------ | ------- |
| `env_ids` | `list[int]` | The environment ids to reset. If None, all environments are reset. | `None`  |

Returns:

| Type                            | Description                                                       |
| ------------------------------- | ----------------------------------------------------------------- |
| `tuple[Tensor, dict[str, Any]]` | A batch of observations and info from the vectorized environment. |

Source code in `genesis_forge/genesis_env.py`

```
def reset(
    self,
    env_ids: list[int] = None,
) -> tuple[torch.Tensor, dict[str, Any]]:
    """
    Reset one or more environments.
    Each of the registered managers will also be reset for those environments.

    Args:
        env_ids: The environment ids to reset. If None, all environments are reset.

    Returns:
        A batch of observations and info from the vectorized environment.
    """
    if env_ids is None:
        env_ids = torch.arange(self.num_envs, device=gs.device)

    # Initial reset, set buffers
    if self.step_count == 0 and self.action_space is not None:
        self._actions = torch.zeros(
            (self.num_envs, self.action_space.shape[0]),
            device=gs.device,
            dtype=gs.tc_float,
        )
        self._last_actions = torch.zeros_like(self._actions, device=gs.device)

    # Actions
    if env_ids.numel() > 0:
        if self.actions is not None:
            self.actions[env_ids] = 0.0
            self._last_actions[env_ids] = 0.0

        # Episode length
        self.episode_length[env_ids] = 0

    # Randomize max episode length for env_ids
    if (
        len(env_ids) > 0
        and self._max_episode_random_scaling > 0.0
        and self._base_max_episode_length is not None
    ):
        max_random_scaling = (
            self._base_max_episode_length * self._max_episode_random_scaling
        )
        randomization = (
            torch.empty((env_ids.numel(),)).uniform_(-1.0, 1.0)
            * max_random_scaling
        )
        self.max_episode_length[env_ids] = torch.round(
            self._base_max_episode_length + randomization
        ).to(gs.tc_int)

    return None, self.extras
```

## `set_max_episode_length(max_episode_length_sec)`

Set or change the maximum episode length.

Parameters:

| Name                     | Type  | Description                            | Default    |
| ------------------------ | ----- | -------------------------------------- | ---------- |
| `max_episode_length_sec` | `int` | The maximum episode length in seconds. | *required* |

Returns:

| Type  | Description                          |
| ----- | ------------------------------------ |
| `int` | The maximum episode length in steps. |

Source code in `genesis_forge/genesis_env.py`

```
def set_max_episode_length(self, max_episode_length_sec: int) -> int:
    """
    Set or change the maximum episode length.

    Args:
        max_episode_length_sec: The maximum episode length in seconds.

    Returns:
        The maximum episode length in steps.
    """
    self._max_episode_length_sec = max_episode_length_sec
    self._base_max_episode_length = math.ceil(max_episode_length_sec / self.dt)
    return self._base_max_episode_length
```

## `step(actions)`

Performs a step in all environments with the given actions.

Parameters:

| Name      | Type     | Description                                                              | Default    |
| --------- | -------- | ------------------------------------------------------------------------ | ---------- |
| `actions` | `Tensor` | Batch of actions for each environment with the :attr:action_space shape. | *required* |

Returns:

| Type                                                    | Description                                                              |
| ------------------------------------------------------- | ------------------------------------------------------------------------ |
| `tuple[Tensor, Tensor, Tensor, Tensor, dict[str, Any]]` | Batch of (observations, rewards, terminations, truncations, info/extras) |

Source code in `genesis_forge/genesis_env.py`

```
def step(
    self, actions: torch.Tensor
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor, dict[str, Any]]:
    """
    Performs a step in all environments with the given actions.

    Args:
        actions: Batch of actions for each environment with the :attr:`action_space` shape.

    Returns:
        Batch of (observations, rewards, terminations, truncations, info/extras)
    """
    self._extras = {}
    self._extras[self.extras_logging_key] = {}
    self.step_count += 1
    self.episode_length += 1

    if self._actions is None:
        self._actions = torch.zeros_like(actions, device=gs.device)
        self._last_actions = torch.zeros_like(actions, device=gs.device)

    self._last_actions[:] = self._actions[:]
    self._actions[:] = actions[:]

    return None, None, None, None, self._extras
```

# ManagedEnvironment

Bases: `GenesisEnv`

An environment which moves a lot of the logic of the environment to manager classes. This helps to keep the environment code clean and modular.

Parameters:

| Name                         | Type    | Description                                                                                                                    | Default                            |
| ---------------------------- | ------- | ------------------------------------------------------------------------------------------------------------------------------ | ---------------------------------- |
| `num_envs`                   | `int`   | Number of parallel environments.                                                                                               | `1`                                |
| `dt`                         | `float` | Simulation time step.                                                                                                          | `1 / 100`                          |
| `max_episode_length_sec`     | \`int   | None\`                                                                                                                         | Maximum episode length in seconds. |
| `max_episode_random_scaling` | `float` | Randomly scale the maximum episode length by this amount (+/-) so that not all environments reset at the same time.            | `0.0`                              |
| `extras_logging_key`         | `str`   | The key used, in info/extras dict, which is returned by step and reset functions, to send data to tensorboard by the RL agent. | `'episode'`                        |

Example::

```
class MyEnv(ManagedEnvironment):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        # ...Define scene here...

    def config(self):
        self.action_manager = PositionalActionManager(
            self,
            joint_names=".*",
            pd_kp=50,
            pd_kv=0.5,
            max_force=8.0,
            default_pos={
                # Hip joints
                "Leg[1-2]_Hip": -1.0,
                "Leg[3-4]_Hip": 1.0,
                # Femur joints
                "Leg[1-4]_Femur": 0.5,
                # Tibia joints
                "Leg[1-4]_Tibia": 0.6,
            },
        )
        self.reward_manager = RewardManager(
            self,
            term_cfg={
                "Default pose": {
                    "weight": -1.0,
                    "fn": rewards.dof_similar_to_default,
                    "params": {
                        "dof_action_manager": self.action_manager,
                    },
                },
                "Base height": {
                    "fn": mdp.rewards.base_height,
                    "params": { "target_height": 0.135 },
                    "weight": -100.0,
                },
            },
        )
        ObservationManager(
            self,
            cfg={
                "velocity_cmd": {"fn": self.velocity_command.observation},
                "robot_ang_vel": {
                    "fn": utils.entity_ang_vel,
                    "params": {"entity": self.robot},
                    "noise": 0.1,
                },
                "robot_lin_vel": {
                    "fn": utils.entity_lin_vel,
                    "params": {"entity": self.robot},
                    "noise": 0.1,
                },
                "robot_projected_gravity": {
                    "fn": utils.entity_projected_gravity,
                    "params": {"entity": self.robot},
                    "noise": 0.1,
                },
                "robot_dofs_position": {
                    "fn": self.action_manager.get_dofs_position,
                    "noise": 0.01,
                },
                "actions": {"fn": lambda: env.actions},
            },
        )
```

Source code in `genesis_forge/managed_env.py`

```
def __init__(
    self,
    num_envs: int = 1,
    dt: float = 1 / 100,
    max_episode_length_sec: int | None = 10,
    max_episode_random_scaling: float = 0.0,
    extras_logging_key: str = "episode",
):
    super().__init__(
        num_envs=num_envs,
        dt=dt,
        max_episode_length_sec=max_episode_length_sec,
        max_episode_random_scaling=max_episode_random_scaling,
        extras_logging_key=extras_logging_key,
    )

    self.managers: ManagersDict = {
        "contact": [],
        "entity": [],
        "command": [],
        "terrain": [],
        "actuator": [],
        "action": [],
        # there can only be one of each of these
        "observation": [],
        "reward": None,
        "termination": None,
    }

    self._action_space = None
    self._action_ranges: list[tuple[int, int]] = []
    self._observation_space = None
    self._reward_buf = torch.zeros(
        (self.num_envs,), device=gs.device, dtype=gs.tc_float
    )
    self._terminated_buf = torch.zeros(
        (self.num_envs,), device=gs.device, dtype=gs.tc_bool
    )
    self._truncated_buf = torch.zeros(
        (self.num_envs,), device=gs.device, dtype=gs.tc_bool
    )
    self._observations_buf = TensorDict({}, device=gs.device)
```

## `_action_ranges = []`

## `_action_space = None`

## `_actions = None`

## `_base_max_episode_length = None`

## `_extras = {}`

## `_last_actions = None`

## `_max_episode_length_sec = 0.0`

## `_max_episode_random_scaling = max_episode_random_scaling`

## `_observation_space = None`

## `_observations_buf = TensorDict({}, device=(gs.device))`

## `_reward_buf = torch.zeros((self.num_envs,), device=(gs.device), dtype=(gs.tc_float))`

## `_terminated_buf = torch.zeros((self.num_envs,), device=(gs.device), dtype=(gs.tc_bool))`

## `_truncated_buf = torch.zeros((self.num_envs,), device=(gs.device), dtype=(gs.tc_bool))`

## `action_space`

The action space, provided by the action manager(s), if any exist.

## `actions`

The actions for each environment for this step. If you're using an action manager, these are the actions prior to being handled by the action manager.

## `device = gs.device`

## `dt = dt`

## `episode_length = torch.zeros((self.num_envs,), device=(gs.device), dtype=(torch.int32))`

## `extras`

The extras/infos dictionary reset at the start of every step, and contains additional data about the environment during that step.

## `extras_logging_key = extras_logging_key`

## `last_actions`

The actions for for the previous step.

## `managers = {'contact': [], 'entity': [], 'command': [], 'terrain': [], 'actuator': [], 'action': [], 'observation': [], 'reward': None, 'termination': None}`

## `max_episode_length = None`

## `max_episode_length_sec`

The max episode length, in seconds, for each environment.

## `max_episode_length_steps`

The max episode length, in steps, for each environment. If episode randomization scaling is enabled, this will be the base max episode length before scaling.

## `num_actions`

The number of actions for each environment.

## `num_envs = num_envs`

## `num_observations`

The number of observations for each environment.

## `observation_space`

Observation space after :meth:`build`.

If a manager is named `"policy"`, that manager's space is used. Otherwise all observation managers are concatenated in registration order (same as :meth:`get_observations`).

## `robot = None`

## `scene = None`

## `step_count = 0`

## `step_dt`

The time step of the environment. This is an alias of the :attr:`dt` property.

## `terrain = None`

## `unwrapped`

Returns this environment, not a wrapped version of it.

## `_build_action_managers()`

Build the action managers and combine the action spaces.

Source code in `genesis_forge/managed_env.py`

```
def _build_action_managers(self):
    """
    Build the action managers and combine the action spaces.
    """
    if len(self.managers["action"]) == 0:
        return

    low = []
    high = []
    size = 0
    self._action_ranges = []
    for action_manager in self.managers["action"]:
        action_manager.build()

        start = size
        size += action_manager.action_space.shape[0]
        end = size
        self._action_ranges.append((start, end))

        low.append(action_manager.action_space.low)
        high.append(action_manager.action_space.high)

    self._action_space = spaces.Box(
        low=np.concatenate(low),
        high=np.concatenate(high),
        shape=(size,),
        dtype=np.float32,
    )
```

## `_build_observation_managers()`

Build observation managers and set :attr:`observation_space`.

If any manager is named `"policy"`, use that manager's space only. Otherwise concatenate all managers in registration order (same layout as :meth:`get_observations`).

Source code in `genesis_forge/managed_env.py`

```
def _build_observation_managers(self) -> None:
    """
    Build observation managers and set :attr:`observation_space`.

    If any manager is named ``"policy"``, use that manager's space only. Otherwise
    concatenate all managers in registration order (same layout as
    :meth:`get_observations`).
    """
    obs_managers = self.managers["observation"]
    if len(obs_managers) == 0:
        self._observation_space = None
        return

    # Build observation managers
    policy_obs_mgr = None
    for obs_manager in obs_managers:
        obs_manager.build()
        if obs_manager.name == "policy":
            policy_obs_mgr = obs_manager

    # If ther is an observation manager named "policy", that is the primary observation
    # manager and what will be returned to the main policy
    if policy_obs_mgr is not None:
        self._observation_space = obs_manager.observation_space
        return

    # Merge the observation manager spaces
    low = []
    high = []
    size = 0
    for obs_manager in obs_managers:
        size += obs_manager.observation_space.shape[0]
        low.append(obs_manager.observation_space.low)
        high.append(obs_manager.observation_space.high)
    self._observation_space = spaces.Box(
        low=np.concatenate(low),
        high=np.concatenate(high),
        shape=(size,),
        dtype=np.float32,
    )
```

## `add_manager(manager_type, manager)`

Adds a manager to the environment. This will automatically be called by the manager class.

Parameters:

| Name           | Type          | Description                 | Default    |
| -------------- | ------------- | --------------------------- | ---------- |
| `manager_type` | `ManagerType` | The type of manager to add. | *required* |
| `manager`      | `BaseManager` | The manager to add.         | *required* |

Source code in `genesis_forge/managed_env.py`

```
def add_manager(self, manager_type: ManagerType, manager: BaseManager):
    """
    Adds a manager to the environment.
    This will automatically be called by the manager class.

    Args:
        manager_type: The type of manager to add.
        manager: The manager to add.
    """
    if manager_type not in self.managers:
        raise ValueError(f"'{manager_type}' is not a valid manager type.")

    # Append manager if the dict item is a list
    if isinstance(self.managers[manager_type], list):
        self.managers[manager_type].append(manager)
    elif self.managers[manager_type] is None:
        self.managers[manager_type] = manager
    else:
        raise ValueError(
            f"Manager type '{manager_type}' already has a manager, and an environment cannot have multiple {manager_type} managers."
        )
```

## `build()`

Builds the environment before the first step. The Genesis scene and all the scene entities must be added before calling this method.

Source code in `genesis_forge/managed_env.py`

```
def build(self):
    """
    Builds the environment before the first step.
    The Genesis scene and all the scene entities must be added before calling this method.
    """
    super().build()
    self.config()

    for terrain_manager in self.managers["terrain"]:
        terrain_manager.build()

    for actuator_manager in self.managers["actuator"]:
        actuator_manager.build()
    self._build_action_managers()

    for contact_manager in self.managers["contact"]:
        contact_manager.build()
    if self.managers["termination"] is not None:
        self.managers["termination"].build()
    if self.managers["reward"] is not None:
        self.managers["reward"].build()
    for command_manager in self.managers["command"]:
        command_manager.build()
    for entity_manager in self.managers["entity"]:
        entity_manager.build()
    self._build_observation_managers()
```

## `close()`

Close the environment.

Source code in `genesis_forge/genesis_env.py`

```
def close(self):
    """Close the environment."""
    pass
```

## `config()`

Override this method and initialize all your managers here.

Example::

```
def config(self):
    EntityManager(
        self,
        entity_attr="robot",
        on_reset={
            "position": {
                "fn": reset.position,
                "params": {
                    "position": INITIAL_BODY_POSITION,
                    "quat": INITIAL_QUAT,
                },
            },
        },
    )
```

Source code in `genesis_forge/managed_env.py`

```
def config(self):
    """
    Override this method and initialize all your managers here.

    Example::

        def config(self):
            EntityManager(
                self,
                entity_attr="robot",
                on_reset={
                    "position": {
                        "fn": reset.position,
                        "params": {
                            "position": INITIAL_BODY_POSITION,
                            "quat": INITIAL_QUAT,
                        },
                    },
                },
            )
    """
    pass
```

## `get_observations()`

Returns the current observations for this step.

Named observations are stored in `extras["observations"]`. If a manager named `"policy"` exists, only its tensor is returned; otherwise all managers are concatenated in registration order.

Source code in `genesis_forge/managed_env.py`

```
def get_observations(self) -> torch.Tensor:
    """
    Returns the current observations for this step.

    Named observations are stored in `extras["observations"]`. 
    If a manager named `"policy"` exists, only its tensor is returned; 
    otherwise all managers are concatenated in registration order.
    """
    self.extras["observations"] = TensorDict({}, device=gs.device)

    if len(self.managers["observation"]) == 0:
        return super().get_observations()

    # Make observations
    parts: list[torch.Tensor] = []
    policy_obs = None
    for obs_manager in self.managers["observation"]:
        obs = obs_manager.get_observations()
        self.extras["observations"][obs_manager.name] = obs
        parts.append(obs)
        if obs_manager.name == "policy":
            policy_obs = obs

    # If there is a "policy" observation manager, this is the one returned to the policy
    if policy_obs is not None:
        return policy_obs
    # Otherwise, concatenate the observation manager spaces
    return torch.cat(parts, dim=-1)
```

## `reset(env_ids=None)`

Reset one or more environments. Each of the registered managers will also be reset for those environments.

Parameters:

| Name      | Type        | Description | Default                                                            |
| --------- | ----------- | ----------- | ------------------------------------------------------------------ |
| `env_ids` | \`list[int] | None\`      | The environment ids to reset. If None, all environments are reset. |

Returns:

| Type                            | Description                                                                                          |
| ------------------------------- | ---------------------------------------------------------------------------------------------------- |
| `tuple[Tensor, dict[str, Any]]` | A batch of observations (if env_ids is None) and an info dictionary from the vectorized environment. |

Source code in `genesis_forge/managed_env.py`

```
def reset(
    self, env_ids: list[int] | None = None
) -> tuple[torch.Tensor, dict[str, Any]]:
    """
    Reset one or more environments.
    Each of the registered managers will also be reset for those environments.

    Args:
        env_ids: The environment ids to reset. If None, all environments are reset.

    Returns:
        A batch of observations (if env_ids is None) and an info dictionary from the vectorized environment.
    """
    (obs, _) = super().reset(env_ids)

    for actuator_manager in self.managers["actuator"]:
        actuator_manager.reset(env_ids)
    for action_manager in self.managers["action"]:
        action_manager.reset(env_ids)
    for entity_manager in self.managers["entity"]:
        entity_manager.reset(env_ids)
    for contact_manager in self.managers["contact"]:
        contact_manager.reset(env_ids)
    if self.managers["termination"] is not None:
        self.managers["termination"].reset(env_ids)
    if self.managers["reward"] is not None:
        self.managers["reward"].reset(env_ids)
    for command_manager in self.managers["command"]:
        command_manager.reset(env_ids)
    for obs_manager in self.managers["observation"]:
        obs_manager.reset(env_ids)

    # Only get observations when env_ids is None because this will be the initial reset called before the first step
    # Otherwise, the observations are ignored
    if env_ids is None:
        obs = self.get_observations()

    return obs, self.extras
```

## `set_max_episode_length(max_episode_length_sec)`

Set or change the maximum episode length.

Parameters:

| Name                     | Type  | Description                            | Default    |
| ------------------------ | ----- | -------------------------------------- | ---------- |
| `max_episode_length_sec` | `int` | The maximum episode length in seconds. | *required* |

Returns:

| Type  | Description                          |
| ----- | ------------------------------------ |
| `int` | The maximum episode length in steps. |

Source code in `genesis_forge/genesis_env.py`

```
def set_max_episode_length(self, max_episode_length_sec: int) -> int:
    """
    Set or change the maximum episode length.

    Args:
        max_episode_length_sec: The maximum episode length in seconds.

    Returns:
        The maximum episode length in steps.
    """
    self._max_episode_length_sec = max_episode_length_sec
    self._base_max_episode_length = math.ceil(max_episode_length_sec / self.dt)
    return self._base_max_episode_length
```

## `step(actions)`

Performs a step in all environments with the given actions.

Parameters:

| Name      | Type     | Description                                                              | Default    |
| --------- | -------- | ------------------------------------------------------------------------ | ---------- |
| `actions` | `Tensor` | Batch of actions for each environment with the :attr:action_space shape. | *required* |

Returns:

| Type                                                    | Description                                                         |
| ------------------------------------------------------- | ------------------------------------------------------------------- |
| `tuple[Tensor, Tensor, Tensor, Tensor, dict[str, Any]]` | Batch of (observations, rewards, terminations, truncations, extras) |

Source code in `genesis_forge/managed_env.py`

```
def step(
    self, actions: torch.Tensor
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor, dict[str, Any]]:
    """
    Performs a step in all environments with the given actions.

    Args:
        actions: Batch of actions for each environment with the :attr:`action_space` shape.

    Returns:
        Batch of (observations, rewards, terminations, truncations, extras)
    """
    super().step(actions)

    # Execute the actions and a simulation step
    for i, action_manager in enumerate[PositionActionManager](
        self.managers["action"]
    ):
        (start, end) = self._action_ranges[i]
        processed_actions = action_manager.step(actions[:, start:end])
        action_manager.send_actions_to_simulation(processed_actions)
    self.scene.step()

    # Update entity managers
    for entity_manager in self.managers["entity"]:
        entity_manager.step()

    # Calculate contact forces
    for contact_manager in self.managers["contact"]:
        contact_manager.step()

    # Calculate termination and truncation
    reset_env_idx = None
    truncated = self._truncated_buf
    terminated = self._terminated_buf
    if self.managers["termination"] is not None:
        terminated, truncated = self.managers["termination"].step()
        reset_env_idx = (
            (terminated | truncated).nonzero(as_tuple=False).reshape((-1,)).detach()
        )

    # Calculate rewards
    rewards = self._reward_buf
    if self.managers["reward"] is not None:
        rewards = self.managers["reward"].step()

    # Command managers
    for command_manager in self.managers["command"]:
        command_manager.step()

    # Reset environments
    if reset_env_idx is not None and reset_env_idx.numel() > 0:
        self.reset(reset_env_idx)

    # Get observations
    obs = self.get_observations()

    return (
        obs,
        rewards,
        terminated,
        truncated,
        self.extras,
    )
```
# Managers

# Manager Overview

These modular classes are used to handle dedicated parts of your environment.

# Actuator

Bases: `BaseManager`

Configures and manages the actuators of your robot. You can define values for all actuators, or target specific joints by name or name pattern (see example). To add some domain randomization, you can define the values as `NoisyValue` objects, which will apply random noise at each reset.

Parameters:

| Name                  | Type         | Description                                                                                   | Default                    |
| --------------------- | ------------ | --------------------------------------------------------------------------------------------- | -------------------------- |
| `env`                 | `GenesisEnv` | The environment to manage the DOF actuators for.                                              | *required*                 |
| `joint_names`         | \`list[str]  | str\`                                                                                         | The joint names to manage. |
| `default_pos`         | \`float      | NoisyValue                                                                                    | dict\`                     |
| `kp`                  | \`float      | NoisyValue                                                                                    | dict\`                     |
| `kv`                  | \`float      | NoisyValue                                                                                    | dict\`                     |
| `max_force`           | \`float      | NoisyValue                                                                                    | tuple[Any, Any]            |
| `damping`             | \`float      | NoisyValue                                                                                    | dict\`                     |
| `stiffness`           | \`float      | NoisyValue                                                                                    | dict\`                     |
| `frictionloss`        | \`float      | NoisyValue                                                                                    | dict\`                     |
| `armature`            | \`float      | NoisyValue                                                                                    | dict\`                     |
| `entity_attr`         | `str`        | The attribute of the environment to get the robot from.                                       | `'robot'`                  |
| `default_noise_scale` | `float`      | (deprecated) This noise scale will be applied to all actuator values. Use NoisyValue instead. | `0.0`                      |

Example::

```
class MyEnv(ManagedEnvironment):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.actuator_manager = ActuatorManager(
            self,
            joint_names=".*",
            default_pos={
                "Leg[1-2]_Hip": -1.0,
                "Leg[3-4]_Hip": 1.0,
                "Leg[1-4]_Femur": 0.5,
                "Leg[1-4]_Tibia": 0.6,
            },
            kp={
                ".*_Hip": NoisyValue(50, 0.02),
                "*__Femur": NoisyValue(30, 0.01),
                "*__Tibia": NoisyValue(30, 0.01),
            },
            kv=0.5,
            max_force=8.0,
        )
```

Source code in `genesis_forge/managers/actuator/actuator_manager.py`

```
def __init__(
    self,
    env: GenesisEnv,
    joint_names: list[str] | str = ".*",
    default_pos: float | NoisyValue | dict = {".*": 0.0},
    kp: float | NoisyValue | dict = None,
    kv: float | NoisyValue | dict = None,
    max_force: float | NoisyValue | tuple[Any, Any] | dict = None,
    damping: float | NoisyValue | dict = None,
    stiffness: float | NoisyValue | dict = None,
    frictionloss: float | NoisyValue | dict = None,
    armature: float | NoisyValue | dict = None,
    default_noise_scale: float = 0.0,
    entity_attr: str = "robot",
):
    super().__init__(env, type="actuator")
    self._dofs: dict[str, int] = {}
    self._robot: RigidEntity = getattr(env, entity_attr)
    self._default_pos_cfg = ensure_dof_pattern(default_pos)
    self._kp_cfg = ensure_dof_pattern(kp)
    self._kv_cfg = ensure_dof_pattern(kv)
    self._max_force_cfg = ensure_dof_pattern(max_force)
    self._damping_cfg = ensure_dof_pattern(damping)
    self._stiffness_cfg = ensure_dof_pattern(stiffness)
    self._frictionloss_cfg = ensure_dof_pattern(frictionloss)
    self._armature_cfg = ensure_dof_pattern(armature)
    self._default_noise_scale = default_noise_scale

    self._batch_dofs_enabled = (
        env.scene.rigid_options.batch_dofs_info
        and env.scene.rigid_options.batch_links_info
    )

    self._values: ValueBuffers = {
        "default_pos": None,
        "force_min": None,
        "force_max": None,
        "kp": None,
        "kv": None,
        "damping": None,
        "stiffness": None,
        "frictionloss": None,
        "armature": None,
    }

    if isinstance(joint_names, str):
        self._joint_name_cfg = [joint_names]
    elif isinstance(joint_names, list):
        self._joint_name_cfg = joint_names
```

## `default_dofs_pos`

Return the default DOF positions.

## `dofs`

Get the names of the joints that are enabled, in the order of the DOF indices.

## `dofs_idx`

Get the indices of the DOF that are enabled (via joint_names).

## `dofs_names`

Get the names of the configured DOFs.

## `enabled = True`

## `env = env`

## `num_dofs`

Get the number of configured DOFs.

## `type = type`

## `build()`

Builds the manager and initialized all the buffers.

Source code in `genesis_forge/managers/actuator/actuator_manager.py`

```
def build(self):
    """
    Builds the manager and initialized all the buffers.
    """
    # Find all configured joints by names/patterns
    for joint in self._robot.joints:
        if joint.type != gs.JOINT_TYPE.REVOLUTE:
            continue
        name = joint.name
        for pattern in self._joint_name_cfg:
            if pattern == name or re.match(f"^{pattern}$", name):
                self._dofs[name] = joint.dof_start
                break

    # Default DOF positions
    # If no configuration is provided, use zero positions for all DOFs.
    if self._default_pos_cfg is not None:
        self._fill_value_buffer("default_pos", self._default_pos_cfg)
    else:
        self._fill_value_buffer("default_pos", {".*": 0.0})

    # Max force
    # The value can either be a single float or a tuple range
    # First normalize them into two dicts: min and max
    if self._max_force_cfg is not None:

        # Normalize the max_force values into a min & max dict
        force_min = {}
        force_max = {}
        for pattern, value in self._max_force_cfg.items():
            if isinstance(value, list):
                force_min[pattern] = value[0]
                force_max[pattern] = value[1]
            elif isinstance(value, NoisyValue):
                force_min[pattern] = NoisyValue(-value.value, value.noise)
                force_max[pattern] = value
            else:
                force_min[pattern] = -value
                force_max[pattern] = value

        self._fill_value_buffer("force_min", force_min)
        self._fill_value_buffer("force_max", force_max)

    # Armature
    # If DOF batching is not enabled, print a warning and set the armature values for all environments without noise.
    if self._armature_cfg is not None:
        self._fill_value_buffer("armature", self._armature_cfg)
        if not self._batch_dofs_enabled:
            armature = self._values["armature"]
            if torch.any(armature["noise"] != 0.0):
                print(
                    "WARNING: Armature randomization settings are only supported when 'batch_dofs_info' and 'batch_links_info' are True in RigidOptions."
                )
            self._robot.set_dofs_armature(armature["buffer"], self.dofs_idx)

    # Other actuator values
    self._fill_value_buffer("kp", self._kp_cfg)
    self._fill_value_buffer("kv", self._kv_cfg)
    self._fill_value_buffer("damping", self._damping_cfg)
    self._fill_value_buffer("stiffness", self._stiffness_cfg)
    self._fill_value_buffer("frictionloss", self._frictionloss_cfg)
```

## `control_dofs_position(position, dofs_idx=None)`

Control the position of the configured DOFs. This is a wrapper for `RigidEntity.control_dofs_position`.

Parameters:

| Name       | Type        | Description                                                                                                                             | Default                                                                                      |
| ---------- | ----------- | --------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------- |
| `position` | `Tensor`    | The position to set the DOFs to. The indices of this tensor should match the configured DOFs (see: dofs_names and dofs_idx properties). | *required*                                                                                   |
| `dofs_idx` | \`list[int] | None\`                                                                                                                                  | The indices of the DOFs to control. If None, all the DOFs of this actuator manager are used. |

Source code in `genesis_forge/managers/actuator/actuator_manager.py`

```
def control_dofs_position(
    self, position: torch.Tensor, dofs_idx: list[int] | None = None
):
    """
    Control the position of the configured DOFs.
    This is a wrapper for `RigidEntity.control_dofs_position`.

    Args:
        position: The position to set the DOFs to. The indices of this tensor should match the configured DOFs
                  (see: `dofs_names` and `dofs_idx` properties).
        dofs_idx: The indices of the DOFs to control. If None, all the DOFs of this actuator manager are used.
    """
    self._robot.control_dofs_position(position, dofs_idx or self.dofs_idx)
```

## `get_dofs_control_force(noise=0.0, clip_to_max_force=False, dofs_idx=None)`

Return the force output by the configured DOFs. This is a wrapper for `RigidEntity.get_dofs_control_force`.

Parameters:

| Name                | Type        | Description                                                             | Default                                                                                                |
| ------------------- | ----------- | ----------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------ |
| `noise`             | `float`     | The maximum amount of random noise to add to the force values returned. | `0.0`                                                                                                  |
| `clip_to_max_force` | `bool`      | Clip the force returned to the maximum force of the actuators.          | `False`                                                                                                |
| `dofs_idx`          | \`list[int] | None\`                                                                  | The indices of the DOFs to get the force for. If None, all the DOFs of this actuator manager are used. |

Returns:

| Name    | Type     | Description                                |
| ------- | -------- | ------------------------------------------ |
| `force` | `Tensor` | torch.Tensor, shape (n_envs, n_dofs)       |
|         | `Tensor` | The force experienced by the enabled DOFs. |

Source code in `genesis_forge/managers/actuator/actuator_manager.py`

```
def get_dofs_control_force(
    self,
    noise: float = 0.0,
    clip_to_max_force: bool = False,
    dofs_idx: list[int] | None = None,
) -> torch.Tensor:
    """
    Return the force output by the configured DOFs.
    This is a wrapper for `RigidEntity.get_dofs_control_force`.

    Args:
        noise: The maximum amount of random noise to add to the force values returned.
        clip_to_max_force: Clip the force returned to the maximum force of the actuators.
        dofs_idx: The indices of the DOFs to get the force for. If None, all the DOFs of this actuator manager are used.

    Returns:
        force: torch.Tensor, shape (n_envs, n_dofs)
        The force experienced by the enabled DOFs.
    """
    dofs_idx = dofs_idx if dofs_idx is not None else self.dofs_idx
    force = self._robot.get_dofs_control_force(dofs_idx)
    if noise > 0.0:
        force = self._add_random_noise(force, noise)
    if clip_to_max_force:
        [lower, upper] = self._robot.get_dofs_force_range(dofs_idx or self.dofs_idx)
        force = force.clamp(lower, upper)
    return force
```

## `get_dofs_force(noise=0.0, clip_to_max_force=False, dofs_idx=None)`

Return the force experienced by the configured DOFs. This is a wrapper for `RigidEntity.get_dofs_force`.

Parameters:

| Name                | Type        | Description                                                             | Default                                                                                                |
| ------------------- | ----------- | ----------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------ |
| `noise`             | `float`     | The maximum amount of random noise to add to the force values returned. | `0.0`                                                                                                  |
| `clip_to_max_force` | `bool`      | Clip the force returned to the maximum force of the actuators.          | `False`                                                                                                |
| `dofs_idx`          | \`list[int] | None\`                                                                  | The indices of the DOFs to get the force for. If None, all the DOFs of this actuator manager are used. |

Returns:

| Name    | Type     | Description                                |
| ------- | -------- | ------------------------------------------ |
| `force` | `Tensor` | torch.Tensor, shape (n_envs, n_dofs)       |
|         | `Tensor` | The force experienced by the enabled DOFs. |

Source code in `genesis_forge/managers/actuator/actuator_manager.py`

```
def get_dofs_force(
    self,
    noise: float = 0.0,
    clip_to_max_force: bool = False,
    dofs_idx: list[int] | None = None,
) -> torch.Tensor:
    """
    Return the force experienced by the configured DOFs.
    This is a wrapper for `RigidEntity.get_dofs_force`.

    Args:
        noise: The maximum amount of random noise to add to the force values returned.
        clip_to_max_force: Clip the force returned to the maximum force of the actuators.
        dofs_idx: The indices of the DOFs to get the force for. If None, all the DOFs of this actuator manager are used.

    Returns:
        force: torch.Tensor, shape (n_envs, n_dofs)
        The force experienced by the enabled DOFs.
    """
    dofs_idx = dofs_idx if dofs_idx is not None else self.dofs_idx
    force = self._robot.get_dofs_force(dofs_idx)
    if noise > 0.0:
        force = self._add_random_noise(force, noise)
    if clip_to_max_force:
        [lower, upper] = self._robot.get_dofs_force_range(dofs_idx or self.dofs_idx)
        force = force.clamp(lower, upper)
    return force
```

## `get_dofs_limits(dofs_idx=None)`

Return the position limits of the configured DOFs. This is a wrapper for `RigidEntity.get_dofs_limit`.

Parameters:

| Name       | Type        | Description | Default                                                                                                 |
| ---------- | ----------- | ----------- | ------------------------------------------------------------------------------------------------------- |
| `dofs_idx` | \`list[int] | None\`      | The indices of the DOFs to get the limits for. If None, all the DOFs of this actuator manager are used. |

Returns:

| Name          | Type     | Description                                                                                                       |
| ------------- | -------- | ----------------------------------------------------------------------------------------------------------------- |
| `lower_limit` | `Tensor` | torch.Tensor, shape (n_dofs,) or (n_envs, n_dofs) The lower limit of the positional limits for the entity's dofs. |
| `upper_limit` | `Tensor` | torch.Tensor, shape (n_dofs,) or (n_envs, n_dofs) The upper limit of the positional limits for the entity's dofs. |

Source code in `genesis_forge/managers/actuator/actuator_manager.py`

```
def get_dofs_limits(
    self, dofs_idx: list[int] | None = None
) -> tuple[torch.Tensor, torch.Tensor]:
    """
    Return the position limits of the configured DOFs.
    This is a wrapper for `RigidEntity.get_dofs_limit`.

    Args:
        dofs_idx: The indices of the DOFs to get the limits for. If None, all the DOFs of this actuator manager are used.

    Returns:
        lower_limit: torch.Tensor, shape (n_dofs,) or (n_envs, n_dofs)
                     The lower limit of the positional limits for the entity's dofs.
        upper_limit: torch.Tensor, shape (n_dofs,) or (n_envs, n_dofs)
                     The upper limit of the positional limits for the entity's dofs.
    """
    dofs_idx = dofs_idx if dofs_idx is not None else self.dofs_idx
    return self._robot.get_dofs_limit(dofs_idx)
```

## `get_dofs_max_force()`

Get the positive force/torque limit per configured DOF from `max_force` config.

Returns:

| Name     | Type     | Description                                                                                                            |
| -------- | -------- | ---------------------------------------------------------------------------------------------------------------------- |
| `limits` | `Tensor` | torch.Tensor, shape (n_envs, n_dofs) or (n_dofs,) Upper force limit per DOF (symmetric limits use the same magnitude). |

Raises:

| Type         | Description                                               |
| ------------ | --------------------------------------------------------- |
| `ValueError` | If max_force was not configured on this actuator manager. |

Source code in `genesis_forge/managers/actuator/actuator_manager.py`

```
def get_dofs_max_force(self) -> torch.Tensor:
    """
    Get the positive force/torque limit per configured DOF from ``max_force`` config.

    Returns:
        limits: torch.Tensor, shape (n_envs, n_dofs) or (n_dofs,)
            Upper force limit per DOF (symmetric limits use the same magnitude).

    Raises:
        ValueError: If ``max_force`` was not configured on this actuator manager.
    """
    if self._max_force_cfg is None or self._values.get("force_max") is None:
        raise ValueError(
            "ActuatorManager max_force is not configured. "
            "Set max_force on the actuator manager or pass an explicit threshold."
        )
    limits = torch.abs(self._get_value_buffer("force_max"))
    if limits.ndim == 1:
        limits = limits.unsqueeze(0).expand(self.env.num_envs, -1)
    return limits
```

## `get_dofs_position(noise=0.0, dofs_idx=None)`

Return the current position of the configured DOFs. This is a wrapper for `RigidEntity.get_dofs_position`.

Parameters:

| Name       | Type        | Description                                                                | Default                                                                                                   |
| ---------- | ----------- | -------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------- |
| `noise`    | `float`     | The maximum amount of random noise to add to the position values returned. | `0.0`                                                                                                     |
| `dofs_idx` | \`list[int] | None\`                                                                     | The indices of the DOFs to get the position for. If None, all the DOFs of this actuator manager are used. |

Returns:

| Name       | Type     | Description                          |
| ---------- | -------- | ------------------------------------ |
| `position` | `Tensor` | torch.Tensor, shape (n_envs, n_dofs) |
|            | `Tensor` | The position of the enabled DOFs.    |

Source code in `genesis_forge/managers/actuator/actuator_manager.py`

```
def get_dofs_position(
    self, noise: float = 0.0, dofs_idx: list[int] | None = None
) -> torch.Tensor:
    """
    Return the current position of the configured DOFs.
    This is a wrapper for `RigidEntity.get_dofs_position`.

    Args:
        noise: The maximum amount of random noise to add to the position values returned.
        dofs_idx: The indices of the DOFs to get the position for. If None, all the DOFs of this actuator manager are used.

    Returns:
        position: torch.Tensor, shape (n_envs, n_dofs)
        The position of the enabled DOFs.
    """
    dofs_idx = dofs_idx if dofs_idx is not None else self.dofs_idx
    pos = self._robot.get_dofs_position(dofs_idx)
    if noise > 0.0:
        pos = self._add_random_noise(pos, noise)
    return pos
```

## `get_dofs_velocity(noise=0.0, clip=None, dofs_idx=None)`

Return the current velocity of the configured DOFs. This is a wrapper for `RigidEntity.get_dofs_velocity`.

Parameters:

| Name       | Type                  | Description                                                                | Default                                                                                                   |
| ---------- | --------------------- | -------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------- |
| `noise`    | `float`               | The maximum amount of random noise to add to the velocity values returned. | `0.0`                                                                                                     |
| `clip`     | `tuple[float, float]` | Clip the velocity returned.                                                | `None`                                                                                                    |
| `dofs_idx` | \`list[int]           | None\`                                                                     | The indices of the DOFs to get the velocity for. If None, all the DOFs of this actuator manager are used. |

Returns:

| Name       | Type     | Description                          |
| ---------- | -------- | ------------------------------------ |
| `velocity` | `Tensor` | torch.Tensor, shape (n_envs, n_dofs) |
|            | `Tensor` | The velocity of the enabled DOFs.    |

Source code in `genesis_forge/managers/actuator/actuator_manager.py`

```
def get_dofs_velocity(
    self,
    noise: float = 0.0,
    clip: tuple[float, float] = None,
    dofs_idx: list[int] | None = None,
) -> torch.Tensor:
    """
    Return the current velocity of the configured DOFs.
    This is a wrapper for `RigidEntity.get_dofs_velocity`.

    Args:
        noise: The maximum amount of random noise to add to the velocity values returned.
        clip: Clip the velocity returned.
        dofs_idx: The indices of the DOFs to get the velocity for. If None, all the DOFs of this actuator manager are used.

    Returns:
        velocity:torch.Tensor, shape (n_envs, n_dofs)
        The velocity of the enabled DOFs.
    """
    dofs_idx = dofs_idx if dofs_idx is not None else self.dofs_idx
    vel = self._robot.get_dofs_velocity(dofs_idx)
    if noise > 0.0:
        vel = self._add_random_noise(vel, noise)
    if clip is not None:
        vel = vel.clamp(**clip)
    return vel
```

## `reset(envs_idx=None)`

Reset the DOF positions.

Source code in `genesis_forge/managers/actuator/actuator_manager.py`

```
def reset(
    self,
    envs_idx: list[int] = None,
):
    """Reset the DOF positions."""
    if not self.enabled:
        return
    dofs_idx = self.dofs_idx

    # Set actuator controller values
    if self._should_set_value("kp"):
        kp = self._get_value_buffer("kp", envs_idx)
        self._robot.set_dofs_kp(kp, dofs_idx, envs_idx)
        self._values["kp"]["has_been_set"] = True

    if self._should_set_value("kv"):
        kv = self._get_value_buffer("kv", envs_idx)
        self._robot.set_dofs_kv(kv, dofs_idx, envs_idx)
        self._values["kv"]["has_been_set"] = True

    if self._should_set_value("damping"):
        damping = self._get_value_buffer("damping", envs_idx)
        self._robot.set_dofs_damping(damping, dofs_idx, envs_idx)
        self._values["damping"]["has_been_set"] = True

    if self._should_set_value("stiffness"):
        stiffness = self._get_value_buffer("stiffness", envs_idx)
        self._robot.set_dofs_stiffness(stiffness, dofs_idx, envs_idx)
        self._values["stiffness"]["has_been_set"] = True

    if self._should_set_value("frictionloss"):
        frictionloss = self._get_value_buffer("frictionloss", envs_idx)
        self._robot.set_dofs_frictionloss(frictionloss, dofs_idx, envs_idx)
        self._values["frictionloss"]["has_been_set"] = True

    if self._should_set_value("armature") and self._batch_dofs_enabled:
        armature = self._get_value_buffer("armature", envs_idx)
        self._robot.set_dofs_armature(armature, dofs_idx, envs_idx)
        self._values["armature"]["has_been_set"] = True

    if self._should_set_value("force_min") or self._should_set_value("force_max"):
        force_min = self._get_value_buffer("force_min", envs_idx)
        force_max = self._get_value_buffer("force_max", envs_idx)
        self._robot.set_dofs_force_range(force_min, force_max, dofs_idx, envs_idx)
        self._values["force_min"]["has_been_set"] = True
        self._values["force_max"]["has_been_set"] = True

    # Reset DOF positions
    default_position = self._get_value_buffer("default_pos", envs_idx)
    self._robot.set_dofs_position(
        position=default_position,
        dofs_idx_local=dofs_idx,
        envs_idx=envs_idx,
    )
```

## `set_dofs_position(position, dofs_idx=None)`

Set the position of the configured DOFs. This is a wrapper for `RigidEntity.set_dofs_position`.

Parameters:

| Name       | Type        | Description                                                                                                                             | Default                                                                                      |
| ---------- | ----------- | --------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------- |
| `position` | `Tensor`    | The position to set the DOFs to. The indices of this tensor should match the configured DOFs (see: dofs_names and dofs_idx properties). | *required*                                                                                   |
| `dofs_idx` | \`list[int] | None\`                                                                                                                                  | The indices of the DOFs to control. If None, all the DOFs of this actuator manager are used. |

Source code in `genesis_forge/managers/actuator/actuator_manager.py`

```
def set_dofs_position(
    self, position: torch.Tensor, dofs_idx: list[int] | None = None
):
    """
    Set the position of the configured DOFs.
    This is a wrapper for `RigidEntity.set_dofs_position`.

    Args:
        position: The position to set the DOFs to. The indices of this tensor should match the configured DOFs
                  (see: `dofs_names` and `dofs_idx` properties).
        dofs_idx: The indices of the DOFs to control. If None, all the DOFs of this actuator manager are used.
    """
    self._robot.set_dofs_position(position, dofs_idx or self.dofs_idx)
```

## `step()`

Called when the environment is stepped

Source code in `genesis_forge/managers/base.py`

```
def step(self):
    """Called when the environment is stepped"""
    pass
```

# Contact

Bases: `BaseManager`

Tracks the contact forces between entity links in the environment.

Parameters:

| Name                         | Type                           | Description                                                                                                | Default                     |
| ---------------------------- | ------------------------------ | ---------------------------------------------------------------------------------------------------------- | --------------------------- |
| `env`                        | `GenesisEnv`                   | The environment to track the contact forces for.                                                           | *required*                  |
| `link_names`                 | `list[str]`                    | The names, or name regex patterns, of the entity links to track the contact forces for.                    | *required*                  |
| `entity_attr`                | `str`                          | The environment attribute which contains the entity with the links we're tracking. Defaults to robot.      | `'robot'`                   |
| `with_entity_attr`           | `str`                          | Filter the contact forces to only include contacts with the entity assigned to this environment attribute. | `None`                      |
| `with_links_names`           | `list[str]`                    | Filter the contact forces to only include contacts with these links.                                       | `None`                      |
| `track_air_time`             | `bool`                         | Whether to track the air time of the entity link contacts.                                                 | `False`                     |
| `air_time_contact_threshold` | `float`                        | When track_air_time is True, this is the threshold for the contact forces to be considered.                | `1.0`                       |
| `debug_visualizer`           | `bool`                         | Whether to visualize the contact points.                                                                   | `False`                     |
| `debug_visualizer_cfg`       | `ContactDebugVisualizerConfig` | The configuration for the contact debug visualizer.                                                        | `DEFAULT_VISUALIZER_CONFIG` |

Example with ManagedEnvironment::

```
class MyEnv(ManagedEnvironment):

    # ... Construct scene and other env setup ...

    def config(self):
        # Define contact manager
        self.foot_contact_manager = ContactManager(
            self,
            link_names=[".*_Foot"],
        )

        # Use contact manager in rewards
        self.reward_manager = RewardManager(
            self,
            term_cfg={
                "Foot contact": {
                    "weight": 5.0,
                    "fn": rewards.has_contact,
                    "params": {
                        "contact_manager": self.foot_contact_manager,
                        "min_contacts": 4,
                    },
                },
            },
        )

        # ... other managers here ...
```

Example using the contact manager directly::

```
class MyEnv(GenesisEnv):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.contact_manager = ContactManager(
            self,
            link_names=[".*_Foot"],
        )

    def build(self):
        super().build()
        self.contact_manager.build()

    def step(self, actions: torch.Tensor):
        super().step(actions)
        self.contact_manager.step()
        return obs, rewards, terminations, timeouts, info

    def reset(self, envs_idx: list[int] | None = None):
        super().reset(envs_idx)
        self.contact_manager.reset(envs_idx)
        return obs, info

    def calculate_rewards():
        # Reward for each foot in contact with something with at least 1.0N force
        CONTACT_THRESHOLD = 1.0
        CONTACT_WEIGHT = 0.005
        has_contact = self.contact_manager.contacts[:,:].norm(dim=-1) > CONTACT_THRESHOLD
        contact_reward = has_contact.sum(dim=1).float() * CONTACT_WEIGHT

        # Access contact positions for debugging or additional analysis
        contact_positions = self.contact_manager.contact_positions
        # contact_positions shape: (n_envs, n_target_links, 3)
        # Positions are automatically averaged when multiple contacts occur

        # ...additional reward calculations here...
```

Filtering::

```
class MyEnv(ManagedEnvironment):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.scene = gs.Scene()

        # Add terrain
        self.terrain = self.scene.add_entity(gs.morphs.Plane())

        # add robot
        self.robot = self.scene.add_entity(
            gs.morphs.URDF(file="urdf/go2/urdf/go2.urdf"),
        )

    def config(self):
        # Track all contacts between the robot's feet and the terrain
        self.contact_manager = ContactManager(
            self,
            entity_attr="robot",
            link_names=[".*_foot"],
            with_entity_attr="terrain",
        )

        # ...other managers here...

    # ...other operations here...
```

Source code in `genesis_forge/managers/contact/contact_manager.py`

```
def __init__(
    self,
    env: GenesisEnv,
    link_names: list[str],
    entity_attr: str = "robot",
    with_entity_attr: str = None,
    with_links_names: list[str] = None,
    track_air_time: bool = False,
    air_time_contact_threshold: float = 1.0,
    debug_visualizer: bool = False,
    debug_visualizer_cfg: ContactDebugVisualizerConfig = DEFAULT_VISUALIZER_CONFIG,
):
    super().__init__(env, "contact")

    self._link_names = link_names
    self._air_time_contact_threshold = air_time_contact_threshold
    self._track_air_time = track_air_time
    self._entity_attr = entity_attr
    self._link_ids = None
    self._local_link_ids = None
    self._with_entity_attr = with_entity_attr
    self._with_links_names = with_links_names
    self._with_link_ids = torch.empty(0, device=gs.device)
    self._with_local_link_ids = None
    self._has_with_filter = (
        with_entity_attr is not None or with_links_names is not None
    )

    self.debug_visualizer = debug_visualizer
    self.debug_envs_idx = None
    self.visualizer_cfg = {**DEFAULT_VISUALIZER_CONFIG, **debug_visualizer_cfg}
    self._debug_nodes = []
    self._contact_position_counts = None

    self.contacts: torch.Tensor | None = None
    """Contact forces experienced by the entity links."""

    self.contact_positions: torch.Tensor | None = None
    """Contact positions for each target link."""

    self.last_air_time: torch.Tensor | None = None
    """Time spent (in s) in the air before the last contact."""

    self.current_air_time: torch.Tensor | None = None
    """Time spent (in s) in the air since the last detach."""

    self.last_contact_time: torch.Tensor | None = None
    """Time spent (in s) in contact before the last detach."""

    self.current_contact_time: torch.Tensor | None = None
    """Time spent (in s) in contact since the last contact."""
```

## `contact_positions = None`

Contact positions for each target link.

## `contacts = None`

Contact forces experienced by the entity links.

## `current_air_time = None`

Time spent (in s) in the air since the last detach.

## `current_contact_time = None`

Time spent (in s) in contact since the last contact.

## `debug_envs_idx = None`

## `debug_visualizer = debug_visualizer`

## `enabled = True`

## `env = env`

## `last_air_time = None`

Time spent (in s) in the air before the last contact.

## `last_contact_time = None`

Time spent (in s) in contact before the last detach.

## `link_ids`

The global link indices for the target links.

## `local_link_ids`

The local link indices for the target links.

## `type = type`

## `visualizer_cfg = {None: DEFAULT_VISUALIZER_CONFIG, None: debug_visualizer_cfg}`

## `build()`

Initialize link indices and buffers.

Source code in `genesis_forge/managers/contact/contact_manager.py`

```
def build(self):
    """Initialize link indices and buffers."""
    super().build()

    # If debug envs_idx is not set, attempt to use the vis_options rendered_envs_idx
    self.debug_envs_idx = self.visualizer_cfg.get("envs_idx", None)
    if self.debug_envs_idx is None and self.env.scene.vis_options is not None:
        self.debug_envs_idx = self.env.scene.vis_options.rendered_envs_idx
    if self.debug_envs_idx is None:
        self.debug_envs_idx = list[int](range(self.env.num_envs))

    # Calculate the number of steps per debug render
    fps = self.visualizer_cfg.get("fps", 30)
    self._steps_per_debug_render = math.ceil(1.0 / fps / self.env.dt)

    # Get the link indices
    (self._link_ids, self._local_link_ids) = self._get_links_idx(
        self._entity_attr, self._link_names
    )
    if not self._link_ids.is_contiguous():
        self._link_ids = self._link_ids.contiguous()
    if self._with_entity_attr or self._with_links_names:
        with_entity_attr = (
            self._with_entity_attr
            if self._with_entity_attr is not None
            else "robot"
        )
        (self._with_link_ids, self._with_local_link_ids) = self._get_links_idx(
            with_entity_attr, self._with_links_names
        )
        if not self._with_link_ids.is_contiguous():
            self._with_link_ids = self._with_link_ids.contiguous()

    # Initialize buffers
    link_count = self._link_ids.shape[0]
    self.contacts = torch.zeros(
        (self.env.num_envs, link_count, 3), device=gs.device
    )
    self.contact_positions = torch.zeros(
        (self.env.num_envs, link_count, 3), device=gs.device
    )
    self._contact_position_counts = torch.zeros(
        (self.env.num_envs, link_count), device=gs.device
    )
    if self._track_air_time:
        self.last_air_time = torch.zeros(
            (self.env.num_envs, link_count), device=gs.device
        )
        self.current_air_time = torch.zeros_like(self.last_air_time)
        self.last_contact_time = torch.zeros_like(self.last_air_time)
        self.current_contact_time = torch.zeros_like(self.last_air_time)
```

## `get_contact_forces(link_idx)`

Get the contact forces for one or more links

Parameters:

| Name       | Type  | Description | Default                                                               |
| ---------- | ----- | ----------- | --------------------------------------------------------------------- |
| `link_idx` | \`int | list[int]\` | The link index or list of link indices to get the contact forces for. |

Returns:

| Type     | Description                                                                   |
| -------- | ----------------------------------------------------------------------------- |
| `Tensor` | The contact forces for the target links. Shape is (n_envs, n_target_links, 3) |

Source code in `genesis_forge/managers/contact/contact_manager.py`

```
def get_contact_forces(self, link_idx: int | list[int]) -> torch.Tensor:
    """
    Get the contact forces for one or more links

    Args:
        link_idx: The link index or list of link indices to get the contact forces for.

    Returns:
        The contact forces for the target links. Shape is (n_envs, n_target_links, 3)
    """
    idx = []
    if isinstance(link_idx, int):
        idx = torch.nonzero(self._link_ids == link_idx)[0]
    elif isinstance(link_idx, list):
        idx = [torch.nonzero(self._link_ids == i)[0].item() for i in link_idx]
    return self.contacts[:, idx, :]
```

## `has_broken_contact(dt, time_margin=1e-08)`

Checks links that have broken contact within the last :attr:`dt` seconds.

This function checks if the links have broken contact within the last :attr:`dt` seconds by comparing the current air time with the given time period. If the air time is less than the given time period, then the links are considered to not be in contact.

Parameters:

| Name          | Type    | Description                                       | Default    |
| ------------- | ------- | ------------------------------------------------- | ---------- |
| `dt`          | `float` | The time period since the contact was broken.     | *required* |
| `time_margin` | `float` | Adds a little error margin to the dt time period. | `1e-08`    |

Returns:

| Type     | Description                                                                    |
| -------- | ------------------------------------------------------------------------------ |
| `Tensor` | A boolean tensor indicating the links that have broken contact within the last |
| `Tensor` | attr:dt seconds. Shape is (n_envs, n_target_links)                             |

Raises:

| Type           | Description                                         |
| -------------- | --------------------------------------------------- |
| `RuntimeError` | If the manager is not configured to track air time. |

Source code in `genesis_forge/managers/contact/contact_manager.py`

```
def has_broken_contact(
    self, dt: float, time_margin: float = 1.0e-8
) -> torch.Tensor:
    """Checks links that have broken contact within the last :attr:`dt` seconds.

    This function checks if the links have broken contact within the last :attr:`dt` seconds
    by comparing the current air time with the given time period. If the air time is less
    than the given time period, then the links are considered to not be in contact.

    Args:
        dt: The time period since the contact was broken.
        time_margin: Adds a little error margin to the dt time period.

    Returns:
        A boolean tensor indicating the links that have broken contact within the last
        :attr:`dt` seconds. Shape is (n_envs, n_target_links)

    Raises:
        RuntimeError: If the manager is not configured to track air time.
    """
    # check if the sensor is configured to track contact time
    if not self._track_air_time:
        raise RuntimeError(
            "The contact manager is not configured to track air time."
            "Please enable the 'track_air_time' in the manager configuration."
        )
    currently_detached = self.current_air_time > 0.0
    less_than_dt_detached = self.current_air_time < (dt + time_margin)
    return currently_detached * less_than_dt_detached
```

## `has_made_contact(dt, time_margin=1e-08)`

Checks if links that have established contact within the last :attr:`dt` seconds.

This function checks if the links have established contact within the last :attr:`dt` seconds by comparing the current contact time with the given time period. If the contact time is less than the given time period, then the links are considered to be in contact.

Parameters:

| Name          | Type    | Description                                        | Default    |
| ------------- | ------- | -------------------------------------------------- | ---------- |
| `dt`          | `float` | The time period since the contact was established. | *required* |
| `time_margin` | `float` | Adds a little error margin to the dt time period.  | `1e-08`    |

Returns:

| Type     | Description                                                                         |
| -------- | ----------------------------------------------------------------------------------- |
| `Tensor` | A boolean tensor indicating the links that have established contact within the last |
| `Tensor` | attr:dt seconds. Shape is (n_envs, n_target_links)                                  |

Raises:

| Type           | Description                                         |
| -------------- | --------------------------------------------------- |
| `RuntimeError` | If the manager is not configured to track air time. |

Source code in `genesis_forge/managers/contact/contact_manager.py`

```
def has_made_contact(self, dt: float, time_margin: float = 1.0e-8) -> torch.Tensor:
    """
    Checks if links that have established contact within the last :attr:`dt` seconds.

    This function checks if the links have established contact within the last :attr:`dt` seconds
    by comparing the current contact time with the given time period. If the contact time is less
    than the given time period, then the links are considered to be in contact.

    Args:
        dt: The time period since the contact was established.
        time_margin: Adds a little error margin to the dt time period.

    Returns:
        A boolean tensor indicating the links that have established contact within the last
        :attr:`dt` seconds. Shape is (n_envs, n_target_links)

    Raises:
        RuntimeError: If the manager is not configured to track air time.
    """
    # check if the sensor is configured to track contact time
    if not self._track_air_time:
        raise RuntimeError(
            "The contact sensor is not configured to track air time."
            "Please enable the 'track_air_time' in the manager configuration."
        )
    # check if the bodies are in contact
    currently_in_contact = self.current_contact_time > 0.0
    less_than_dt_in_contact = self.current_contact_time < (dt + time_margin)
    return currently_in_contact * less_than_dt_in_contact
```

## `reset(envs_idx=None)`

Source code in `genesis_forge/managers/contact/contact_manager.py`

```
def reset(self, envs_idx: list[int] | None = None):
    super().reset(envs_idx)
    if envs_idx is None:
        envs_idx = torch.arange(self.env.num_envs, device=gs.device)

    if not self.enabled:
        return

    # reset the current air time
    if self._track_air_time:
        self.current_air_time[envs_idx] = 0.0
        self.current_contact_time[envs_idx] = 0.0
        self.last_air_time[envs_idx] = 0.0
        self.last_contact_time[envs_idx] = 0.0
```

## `step()`

Source code in `genesis_forge/managers/contact/contact_manager.py`

```
def step(self):
    super().step()
    if not self.enabled:
        return
    self._calculate_contact_forces()
    self._calculate_air_time()
```

# Entity

Tracks entity state (position, velocity, quaternion) and manages per-episode reset operations such as randomizing spawn positions and orientations.

Bases: `BaseManager`

Provides options for resetting an entity and adding noise and randomization to its state.

Parameters:

| Name          | Type                           | Description                                                         | Default    |
| ------------- | ------------------------------ | ------------------------------------------------------------------- | ---------- |
| `env`         | `GenesisEnv`                   | The environment instance.                                           | *required* |
| `entity_attr` | `str`                          | The attribute name of the environment that the entity is stored in. | *required* |
| `on_reset`    | `dict[str, EntityResetConfig]` | The reset configuration for the entity.                             | `{}`       |

Example::

```
class MyEnv(ManagedEnvironment):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def config(self):
        self.entity_manager = EntityManager(
            self,
            entity_attr="robot",
            on_reset={
                "position": {
                    "fn": reset.randomize_terrain_position,
                    "params": {
                        "terrain_manager": self.terrain_manager,
                        "subterrain": self._target_terrain,
                        "height_offset": 0.15,
                    },
                },
            },
        )
```

Source code in `genesis_forge/managers/entity_manager.py`

```
def __init__(
    self,
    env: GenesisEnv,
    entity_attr: str,
    on_reset: dict[str, EntityResetConfig] = {},
):
    super().__init__(env, type="entity")
    if hasattr(env, "add_entity_manager"):
        env.add_entity_manager(self)

    self.entity: RigidEntity | None = None
    self.on_reset = on_reset
    self._entity_attr = entity_attr

    # Wrap config items
    self.on_reset: dict[str, ConfigItem] = {}
    for name, cfg in on_reset.items():
        self.on_reset[name] = ConfigItem(cfg, env)

    # Buffers
    self._global_gravity = torch.tensor(
        [0.0, 0.0, -1.0], device=gs.device, dtype=gs.tc_float
    ).repeat(env.num_envs, 1)
    self._base_pos = torch.zeros(
        (env.num_envs, 3), device=gs.device, dtype=gs.tc_float
    )
    self._base_quat = torch.zeros(
        (env.num_envs, 4), device=gs.device, dtype=gs.tc_float
    )
    self._inv_base_quat = torch.zeros_like(self._base_quat)
```

## `base_pos`

The position of the entities base link.

## `base_quat`

The quaternion of the entity's base link.

## `enabled = True`

## `entity = None`

## `env = env`

## `inv_base_quat`

The inverse of the entity's base link quaternion.

## `on_reset = {}`

## `type = type`

## `build()`

Build the entity manager.

Source code in `genesis_forge/managers/entity_manager.py`

```
def build(self):
    """
    Build the entity manager.
    """
    self.entity = getattr(self.env, self._entity_attr)
    self._cached_calcs()

    # Build reset function classes
    for cfg in self.on_reset.values():
        cfg.build(entity=self.entity)
```

## `get_angular_velocity()`

The angular velocity of the entity's base link, in the entity's local frame.

Source code in `genesis_forge/managers/entity_manager.py`

```
def get_angular_velocity(self) -> torch.Tensor:
    """
    The angular velocity of the entity's base link, in the entity's local frame.
    """
    return transform_by_quat(self.entity.get_ang(), self._inv_base_quat)
```

## `get_linear_velocity()`

The linear velocity of the entity's base link, in the entity's local frame.

Source code in `genesis_forge/managers/entity_manager.py`

```
def get_linear_velocity(self) -> torch.Tensor:
    """
    The linear velocity of the entity's base link, in the entity's local frame.
    """
    return transform_by_quat(self.entity.get_vel(), self._inv_base_quat)
```

## `get_projected_gravity()`

The projected gravity of the entity's base link, in the entity's local frame.

Source code in `genesis_forge/managers/entity_manager.py`

```
def get_projected_gravity(self) -> torch.Tensor:
    """
    The projected gravity of the entity's base link, in the entity's local frame.
    """
    return transform_by_quat(self._global_gravity, self._inv_base_quat)
```

## `reset(envs_idx=None)`

Call all reset functions

Source code in `genesis_forge/managers/entity_manager.py`

```
def reset(self, envs_idx: list[int] | None = None):
    """
    Call all reset functions
    """
    if not self.enabled:
        return
    if envs_idx is None:
        envs_idx = torch.arange(self.env.num_envs, device=gs.device)

    for name, cfg in self.on_reset.items():
        try:
            cfg.execute(envs_idx)
        except Exception as e:
            print(f"Error resetting entity with config: '{name}'")
            raise e
```

## `step()`

Run some common shared calculations at each step.

Source code in `genesis_forge/managers/entity_manager.py`

```
def step(self):
    """
    Run some common shared calculations at each step.
    """
    self._cached_calcs()
```

# NoisyValue

Defines a base value and the noise which will be added to it. This class is merely a configuration container, and does not apply the noise directly.

Parameters:

| Name    | Type    | Description                                     | Default    |
| ------- | ------- | ----------------------------------------------- | ---------- |
| `value` | `float` | The value to configure the manager with.        | *required* |
| `noise` | `float` | The noise (+/-) to apply to the value as noise. | `0.0`      |

Example::

```
value = NoisyValue(10.0, noise=2.0)
# The base value is 10.0, and the noise will be +/- 2.0
# So the final value will be between 8.0 and 12.0
```

Source code in `genesis_forge/managers/actuator/noisy_value.py`

```
def __init__(
    self,
    value: float,
    noise: float = 0.0,
):
    self.value = value
    self.noise = noise
```

## `noise = noise`

## `value = value`

# Observation

Bases: `BaseManager`

Defines the observations and observation space for the environment.

Parameters:

| Name          | Type                           | Description                                                                                                                                      | Default                                                            |
| ------------- | ------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------ |
| `env`         | `GenesisEnv`                   | The environment.                                                                                                                                 | *required*                                                         |
| `cfg`         | `dict[str, ObservationConfig]` | The configuration for the observation manager.                                                                                                   | *required*                                                         |
| `name`        | `str`                          | The name to categorize the observations under, generally used for asymmetrical RL. It's required to have one observation manager named "policy". | `'policy'`                                                         |
| `noise`       | \`float                        | None\`                                                                                                                                           | The range of random noise to add to all observations.              |
| `history_len` | \`int                          | None\`                                                                                                                                           | The number of previous observations to include in the observation. |

Example with ManagedEnvironment::

```
class MyEnv(ManagedEnvironment):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    config(self):
        ObservationManager(
            self,
            cfg={
                "velocity_cmd": {"fn": self.velocity_command.observation},
                "robot_ang_vel": {
                    "fn": utils.entity_ang_vel,
                    "params": {"entity": self.robot},
                    "noise": 0.1,
                },
                "robot_lin_vel": {
                    "fn": utils.entity_lin_vel,
                    "params": {"entity": self.robot},
                    "noise": 0.1,
                },
                "robot_projected_gravity": {
                    "fn": utils.entity_projected_gravity,
                    "params": {"entity": self.robot},
                    "noise": 0.1,
                },
                "robot_dofs_position": {
                    "fn": self.action_manager.get_dofs_position,
                    "noise": 0.01,
                },
                "actions": {"fn": lambda: env.actions},
            },
        )
```

Example using the observation manager directly::

```
class MyEnv(GenesisEnv):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.observation_manager = ObservationManager(
            self,
            cfg={
                "velocity_cmd": {"fn": self.velocity_command.observation},
                "robot_ang_vel": {
                    "fn": utils.entity_ang_vel,
                    "params": {"entity": self.robot},
                    "noise": 0.1,
                },
                "robot_lin_vel": {
                    "fn": utils.entity_lin_vel,
                    "params": {"entity": self.robot},
                    "noise": 0.1,
                },
                "robot_projected_gravity": {
                    "fn": utils.entity_projected_gravity,
                    "params": {"entity": self.robot},
                    "noise": 0.1,
                },
                "robot_dofs_position": {
                    "fn": self.action_manager.get_dofs_position,
                    "noise": 0.01,
                },
                "actions": {"fn": lambda: env.actions},
            },
        )

    @property
    observation_space(self):
        return self.obs_manager.observation_space

    def build(self):
        super().build()
        self.obs_manager.build()

    def step(self, actions: torch.Tensor):
        super().step(actions)

        # ... step logic ...

        obs = self.observation_manager.observation()
        return obs, rewards, terminations, timeouts, info

    def reset(self, envs_idx: list[int] | None = None):
        super().reset(envs_idx)

        # ... reset logic ...

        obs = self.observation_manager.observation()
        return obs, info
```

Source code in `genesis_forge/managers/observation_manager.py`

```
def __init__(
    self,
    env: GenesisEnv,
    cfg: dict[str, ObservationConfig],
    name: str = "policy",
    history_len: int | None = None,
    noise: float | None = None,
):
    super().__init__(env, "observation")
    self._name = name
    self.cfg = cfg
    self.noise = noise
    self._observation_size = 1
    self._observation_space = None

    if history_len is not None and history_len < 1:
        raise ValueError("history_len must be greater than 0")
    self._history_len = history_len if history_len is not None else 1
    self._history = []

    # Wrap config items
    self.cfg: dict[str, ObservationConfigItem] = {}
    for name, cfg in cfg.items():
        self.cfg[name] = ObservationConfigItem(cfg, env)
```

## `cfg = {}`

## `enabled = True`

## `env = env`

## `name`

The name to categorize the observations under This is generally used for asymmetrical RL and it's required to have one observation manager named "policy".

## `noise = noise`

## `observation_space`

The observation space.

## `type = type`

## `build()`

Determine the observation space and setup the buffers.

Source code in `genesis_forge/managers/observation_manager.py`

```
def build(self):
    """
    Determine the observation space and setup the buffers.
    """
    if not self.enabled:
        self._observation_size = 1
        self._observation_space = spaces.Box(
            low=-np.inf,
            high=np.inf,
            shape=(1,),
            dtype=np.float32,
        )
        return

    # Setup observation functions and the observation space
    single_obs_size = self._setup_observation_functions()
    self._observation_size = single_obs_size * self._history_len
    self._observation_space = spaces.Box(
        low=-np.inf,
        high=np.inf,
        shape=(self._observation_size,),
        dtype=np.float32,
    )

    # Fill history buffer
    shape = (self.env.num_envs, single_obs_size)
    self._history = [
        torch.zeros(shape, device=gs.device) for _ in range(self._history_len)
    ]
    self._history_output = torch.zeros(
        (self.env.num_envs, self._observation_size),
        device=gs.device,
    )
```

## `get_observations(values=None)`

Generate current observations for all environments.

Optionally, you can provide the observation values directly as a dictionary of values, and this method will return the formatted/scaled (without noise) tensor for the policy. This is useful for manual deployments or troubleshooting.

Parameters:

| Name     | Type               | Description | Default |
| -------- | ------------------ | ----------- | ------- |
| `values` | \`dict\[str, float | Tensor\]    | None\`  |

Returns:

| Type     | Description                            |
| -------- | -------------------------------------- |
| `Tensor` | The observations for all environments. |

Source code in `genesis_forge/managers/observation_manager.py`

```
def get_observations(
    self, values: dict[str, float | torch.Tensor] | None = None
) -> torch.Tensor:
    """
    Generate current observations for all environments.

    Optionally, you can provide the observation values directly as a dictionary of values, and 
    this method will return the formatted/scaled (without noise) tensor for the policy.
    This is useful for manual deployments or troubleshooting.

    Args:
        values: (optional) If provided, these values will be used instead of fetching observations from the config functions.
                It's expected that this dict contains a key for every observation configuration.
                These values will be scaled, based on the configuration, but not receive any noise.
                This is useful for providing observations for deployment.

    Returns:
        The observations for all environments.
    """
    if not self.enabled:
        return torch.zeros((self.env.num_envs, self._observation_size))

    buffer = self._history.pop()
    self._perform_observation(buffer, values)
    self._history.insert(0, buffer)

    # Concatenate the history buffers into the pre-allocated output buffer
    # This is more performant than torch.cat()
    offset = 0
    for obs in self._history:
        size = obs.shape[1]
        self._history_output[:, offset : offset + size] = obs
        offset += size
    return self._history_output.clone()
```

## `reset(envs_idx=None)`

One or more environments have been reset

Source code in `genesis_forge/managers/base.py`

```
def reset(self, envs_idx: list[int] | None = None):
    """One or more environments have been reset"""
    pass
```

## `step()`

Called when the environment is stepped

Source code in `genesis_forge/managers/base.py`

```
def step(self):
    """Called when the environment is stepped"""
    pass
```

# Reward

Bases: `BaseManager`

Handles calculating and logging the rewards for the environment.

This works with a dictionary configuration of reward handlers. For each dictionary item, a function will be called to calculate a reward value for the environment.

Parameters:

| Name              | Type                      | Description                                              | Default     |
| ----------------- | ------------------------- | -------------------------------------------------------- | ----------- |
| `env`             | `GenesisEnv`              | The environment to manage the rewards for.               | *required*  |
| `cfg`             | `dict[str, RewardConfig]` | A dictionary of reward conditions.                       | *required*  |
| `logging_enabled` | `bool`                    | Whether to log the rewards to tensorboard.               | `True`      |
| `logging_tag`     | `str`                     | The section name used to log the rewards to tensorboard. | `'Rewards'` |

Example with ManagedEnvironment::

```
class MyEnv(ManagedEnvironment):
    def config(self):
        self.reward_manager = RewardManager(
            self,
            cfg={
                "Default pose": {
                    "fn": mdp.rewards.dof_similar_to_default,
                    "weight": -0.1,
                },
                "Base height": {
                    "fn": mdp.rewards.base_height,
                    "params": { "target_height": 0.135 },
                    "weight": -100.0,
                },
            },
        )
```

Example using the reward manager directly::

```
class MyEnv(GenesisEnv):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.reward_manager = RewardManager(
            self,
            cfg={
                "Base height": {
                    "fn": mdp.rewards.base_height,
                    "params": { "target_height": 0.135 },
                    "weight": -100.0,
                },
            },
        )

    def build(self):
        super().build()
        self.reward_manager.build()

    def step(self, actions: torch.Tensor):
        super().step(actions)
        rewards = self.reward_manager.step()
        # ... other step logic ...
        return obs, rewards, terminations, timeouts, info

    def reset(self, envs_idx: list[int] | None = None):
        super().reset(envs_idx)
        # ... other reset logic ...
        return obs, info
```

Source code in `genesis_forge/managers/reward_manager.py`

```
def __init__(
    self,
    env: GenesisEnv,
    cfg: dict[str, RewardConfig],
    logging_enabled: bool = True,
    logging_tag: str = "Rewards",
):
    super().__init__(env, type="reward")

    self.logging_enabled = logging_enabled
    self.logging_tag = logging_tag

    # Wrap config items
    self.cfg: dict[str, RewardConfigItem] = {}
    for name, cfg in cfg.items():
        self.cfg[name] = RewardConfigItem(cfg, env)

    # Initialize buffers
    self._reward_buf = torch.zeros(
        (env.num_envs,), device=gs.device, dtype=gs.tc_float
    )
    self._episode_seconds = torch.zeros(
        (self.env.num_envs,), device=gs.device, dtype=gs.tc_float
    )
    self._episode_mean: dict[str, torch.Tensor] = dict()
    self._episode_data: dict[str, torch.Tensor] = dict()
    for name in self.cfg.keys():
        self._episode_data[name] = torch.zeros(
            (env.num_envs,), device=gs.device, dtype=gs.tc_float
        )
```

## `cfg = {}`

## `enabled = True`

## `env = env`

## `episode_data`

Get the accumulated reward data for the current episode of all environments.

## `logging_enabled = logging_enabled`

## `logging_tag = logging_tag`

## `rewards`

The rewards calculated for the most recent step. Shape is (num_envs,).

## `type = type`

## `build()`

Build any config item function classes.

Source code in `genesis_forge/managers/reward_manager.py`

```
def build(self):
    """
    Build any config item function classes.
    """
    for cfg in self.cfg.values():
        cfg.build()
```

## `last_episode_mean_reward(name, before_weight=True)`

Get the last mean reward for an episode for a given reward name. The mean reward is only calculated when episodes end/reset.

Parameters:

| Name            | Type   | Description                                                                | Default    |
| --------------- | ------ | -------------------------------------------------------------------------- | ---------- |
| `name`          | `str`  | The name of the reward to get the mean for.                                | *required* |
| `before_weight` | `bool` | If True, this will be the base reward value before the weight was applied. | `True`     |

Returns:

| Type    | Description                                                  |
| ------- | ------------------------------------------------------------ |
| `float` | The last mean reward for an episode for a given reward name. |

Source code in `genesis_forge/managers/reward_manager.py`

```
def last_episode_mean_reward(self, name: str, before_weight: bool = True) -> float:
    """
    Get the last mean reward for an episode for a given reward name.
    The mean reward is only calculated when episodes end/reset.

    Args:
        name: The name of the reward to get the mean for.
        before_weight: If True, this will be the base reward value before the weight was applied.

    Returns:
        The last mean reward for an episode for a given reward name.
    """
    rew = self._episode_mean.get(name, 0.0)
    if before_weight:
        rew /= self.cfg[name].weight
    return rew
```

## `reset(envs_idx=None)`

Log the reward mean values at the end of the episode

Source code in `genesis_forge/managers/reward_manager.py`

```
def reset(self, envs_idx: list[int] | None = None):
    """Log the reward mean values at the end of the episode"""
    if envs_idx is None:
        envs_idx = torch.arange(self.env.num_envs, device=gs.device)

    # Reset function classes
    for cfg in self.cfg.values():
        cfg.reset(envs_idx)

    # Log the reward data
    if self.enabled and self.logging_enabled:
        logging_dict = self.env.extras[self.env.extras_logging_key]

        episode_seconds = self._episode_seconds[envs_idx]
        for name, value in self._episode_data.items():
            # Don't log items that have zero weight
            cfg = self.cfg[name]
            if cfg.weight != 0:
                # Calculate average for each environment
                value[envs_idx] /= episode_seconds

                # Take the mean across all episodes
                episode_mean = torch.mean(value[envs_idx])
                self._episode_mean[name] = episode_mean.item()
                logging_dict[f"{self.logging_tag} / {name}"] = episode_mean

            # Reset episodic data
            self._episode_data[name][envs_idx] = 0.0

    # Reset episode seconds to nearly zero, to prevent divide by zero errors
    self._episode_seconds[envs_idx] = 1e-10
```

## `step()`

Calculate the rewards for this step

Returns:

| Type     | Description                                             |
| -------- | ------------------------------------------------------- |
| `Tensor` | The rewards for the environments. Shape is (num_envs,). |

Source code in `genesis_forge/managers/reward_manager.py`

```
def step(self) -> torch.Tensor:
    """
    Calculate the rewards for this step

    Returns:
        The rewards for the environments. Shape is (num_envs,).
    """
    if not self.enabled:
        return self._reward_buf

    dt = self.env.dt
    self._reward_buf[:] = 0.0
    self._episode_seconds += dt
    for name, cfg in self.cfg.items():
        # Don't calculate reward if the weight is zero
        if cfg.weight == 0:
            continue

        # Get reward value from function
        weight = cfg.weight * dt
        value = cfg.fn(self.env, **cfg.params) * weight

        # Add to reward buffer
        self._reward_buf += value

        # Add to episode data for logging (if enabled)
        if self.logging_enabled:
            self._episode_data[name] += value

    return self._reward_buf
```

# Termination

Bases: `BaseManager`

Handles calculating and logging the "dones" (termination or truncation) for the environments.

This works with a dictionary configuration of termination conditions. For each dictionary item, a function will be called to calculate a termination signal for the environment.

Parameters:

| Name              | Type                           | Description                                                         | Default          |
| ----------------- | ------------------------------ | ------------------------------------------------------------------- | ---------------- |
| `env`             | `GenesisEnv`                   | The environment to manage the termination for.                      | *required*       |
| `term_cfg`        | `dict[str, TerminationConfig]` | A dictionary of termination conditions.                             | *required*       |
| `logging_enabled` | `bool`                         | Whether to log the termination signals to tensorboard.              | `True`           |
| `logging_tag`     | `str`                          | The section tag used to log the termination signals to tensorboard. | `'Terminations'` |

Example with ManagedEnvironment::

```
class MyEnv(ManagedEnvironment):
    def config(self):
        self.termination_manager = TerminationManager(
            self,
            term_cfg={
                "Min Height": {
                    "fn": mdp.terminations.min_height,
                    "params": {"min_height": 0.05},
                },
            },
        )
```

Example using the termination manager directly::

```
class MyEnv(GenesisEnv):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.termination_manager = TerminationManager(
            self,
            term_cfg={
                "Min Height": {
                    "fn": mdp.terminations.min_height,
                    "params": {"min_height": 0.5},
                },
                "Rolled over": {
                    "fn": mdp.terminations.max_angle,
                    "params": { "quat_threshold": 0.35 },
                },
            },
        )

    def build(self):
        super().build()
        self.termination_manager.build()

    def step(self, actions: torch.Tensor):
        super().step(actions)
        # ...handle actions...

        # Calculate dones (terminated or truncated)
        terminated, truncated = self.termination_manager.step()
        dones = terminated | truncated
        reset_env_idx = dones.nonzero(as_tuple=False).reshape((-1,))

        # Reset environments
        if reset_env_idx.numel() > 0:
            self.reset(reset_env_idx)

        return obs, rewards, terminated, truncated, info

    def reset(self, envs_idx: Sequence[int] = None):
        super().reset(envs_idx)
        # ...do reset logic here...x

        self.termination_manager.reset(envs_idx)
        return obs, info
```

Source code in `genesis_forge/managers/termination_manager.py`

```
def __init__(
    self,
    env: GenesisEnv,
    term_cfg: dict[str, TerminationConfig],
    logging_enabled: bool = True,
    logging_tag: str = "Terminations",
):
    super().__init__(env, type="termination")

    self.term_cfg = term_cfg
    self.logging_enabled = logging_enabled
    self.logging_tag = logging_tag

    # Wrap config items
    self.term_cfg: dict[str, TerminationConfigItem] = {}
    for name, cfg in term_cfg.items():
        self.term_cfg[name] = TerminationConfigItem(cfg, env)

    # Buffers
    self._terminated_buf = torch.zeros(
        env.num_envs, device=gs.device, dtype=torch.bool
    )
    self._truncated_buf = torch.zeros_like(self._terminated_buf)
```

## `dones`

The termination signals for the environments. Shape is (num_envs,).

## `enabled = True`

## `env = env`

## `logging_enabled = logging_enabled`

## `logging_tag = logging_tag`

## `term_cfg = {}`

## `terminated`

The termination signals for the environments. Shape is (num_envs,).

## `truncated`

The truncation signals for the environments. Shape is (num_envs,).

## `type = type`

## `build()`

Build any config item function classes.

Source code in `genesis_forge/managers/termination_manager.py`

```
def build(self):
    """
    Build any config item function classes.
    """
    for cfg in self.term_cfg.values():
        cfg.build()
```

## `reset(envs_idx=None)`

One or more environments have been reset

Source code in `genesis_forge/managers/base.py`

```
def reset(self, envs_idx: list[int] | None = None):
    """One or more environments have been reset"""
    pass
```

## `step()`

Calculate the termination/truncation signals for this step

Returns:

| Type     | Description                                                                      |
| -------- | -------------------------------------------------------------------------------- |
| `Tensor` | terminated - The termination signals for the environments. Shape is (num_envs,). |
| `Tensor` | truncated - The truncation signals for the environments. Shape is (num_envs,).   |

Source code in `genesis_forge/managers/termination_manager.py`

```
def step(self) -> tuple[torch.Tensor, torch.Tensor]:
    """
    Calculate the termination/truncation signals for this step

    Returns:
        terminated - The termination signals for the environments. Shape is (num_envs,).
        truncated - The truncation signals for the environments. Shape is (num_envs,).
    """
    if not self.enabled:
        return self._terminated_buf, self._truncated_buf

    self._terminated_buf[:] = False
    self._truncated_buf[:] = False
    logging_dict = self.env.extras[self.env.extras_logging_key]
    for name, term_item in self.term_cfg.items():
        try:
            # Get termination value
            params = term_item.params
            value = term_item.fn(self.env, **params)

            # Add to the correct buffer using in-place operations
            if term_item.time_out:
                self._truncated_buf |= value
            else:
                self._terminated_buf |= value

            # Logging, if there are some terminations and timeouts
            dones = value.nonzero(as_tuple=True)[0]
            if self.logging_enabled and dones.numel() > 0:
                logging_dict[f"{self.logging_tag} / {name}"] = (
                    value.float().mean().detach()
                )

        except Exception as e:
            print(f"Error calculating termination for '{name}'")
            raise e

    self.env.extras["terminations"] = self._terminated_buf
    self.env.extras["time_outs"] = self._truncated_buf
    return self._terminated_buf, self._truncated_buf
```

# Terrain

Bases: `BaseManager`

Provides useful functions for the environment terrain. The manager maps out the sizes and heights of the terrain and subterrain. This allows your environment to calculate the robot's height above rough terrain. You can also generate random positions on the terrain or subterrain to place your robots on reset.

Parameters:

| Name           | Type         | Description                                           | Default     |
| -------------- | ------------ | ----------------------------------------------------- | ----------- |
| `env`          | `GenesisEnv` | The environment instance.                             | *required*  |
| `terrain_attr` | `str`        | The attribute name of the terrain in the environment. | `'terrain'` |

Example::

```
class MyEnv(ManagedEnvironment):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.scene = gs.Scene()

        # Add terrain
        self.terrain = self.scene.add_entity(
            morph=gs.morphs.Terrain(
                n_subterrains=(2, 2),
                subterrain_size=(25, 25),
                subterrain_types=[
                    ["flat_terrain", "random_uniform_terrain"],
                    ["discrete_obstacles_terrain", "pyramid_stairs_terrain"],
                ],
            ),
        )

    def config(self):
        self.terrain_manager = TerrainManager(
            self,
            terrain_attr="terrain",
        )

     def reset(self, envs_idx: list[int] = None) -> tuple[torch.Tensor, dict[str, Any]]:
        # Randomize positions on the terrain
        pos = self.terrain_manager.generate_random_env_pos(
            envs_idx=envs_idx,
            subterrain="flat_terrain",
            height_offset=0.15,
        )
        self.robot.set_pos(pos, envs_idx=envs_idx)
```

Source code in `genesis_forge/managers/terrain_manager.py`

```
def __init__(
    self,
    env: GenesisEnv,
    terrain_attr: str = "terrain",
):
    super().__init__(env, type="terrain")

    self._origin = (0, 0, 0)
    self._bounds = (0, 0, 0, 0)  # x_min, x_max, y_min, y_max
    self._size = (0, 0)
    self._terrain: RigidEntity = None
    self._terrain_attr = terrain_attr
    self._subterrain_bounds = {}
    self._height_field: torch.Tensor | None = None
    self._env_pos_buffer = torch.zeros(
        (self.env.num_envs, 3), device=gs.device, dtype=gs.tc_float
    )

    # Pre-allocated buffers for terrain height calculation to avoid memory allocations
    self._norm_coords_buffer = torch.zeros(
        (self.env.num_envs, 2), device=gs.device, dtype=gs.tc_float
    )
    self._grid_buffer = torch.zeros(
        (self.env.num_envs, 1, 1, 2), device=gs.device, dtype=gs.tc_float
    )
    self._heights_buffer = torch.zeros(
        self.env.num_envs, device=gs.device, dtype=gs.tc_float
    )
```

## `enabled = True`

## `env = env`

## `type = type`

## `build()`

Cache the terrain height field

Source code in `genesis_forge/managers/terrain_manager.py`

```
def build(self):
    """Cache the terrain height field"""
    self._terrain = self.env.__getattribute__(self._terrain_attr)
    self._map_terrain()
```

## `generate_random_env_pos(envs_idx=None, usable_ratio=0.5, subterrain=None, height_offset=0.0001)`

Generate one X/Y/Z position on the terrain for each environment. The X & Y positions will be random points and the Z position will be at the approximate terrain height at that point.

Parameters:

| Name            | Type        | Description                                                                                                                                                                                                                                                          | Default                                                                                                |
| --------------- | ----------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------ |
| `envs_idx`      | \`list[int] | Tensor                                                                                                                                                                                                                                                               | None\`                                                                                                 |
| `usable_ratio`  | `float`     | How much of the terrain/subterrain area should be used for random positions. For example, 0.25 will only generate positions within the center 25% of the area of the terrain/subterrain. This helps avoid placing things right on th edge of the terrain/subterrain. | `0.5`                                                                                                  |
| `subterrain`    | \`str       | None\`                                                                                                                                                                                                                                                               | The subterrain to generate positions for. If None, positions will be generated for the entire terrain. |
| `height_offset` | `float`     | The offset to add to the terrain height.                                                                                                                                                                                                                             | `0.0001`                                                                                               |

Returns:

| Type     | Description                         |
| -------- | ----------------------------------- |
| `Tensor` | The position tensor of shape (1, 3) |

Source code in `genesis_forge/managers/terrain_manager.py`

```
def generate_random_env_pos(
    self,
    envs_idx: list[int] | torch.Tensor | None = None,
    usable_ratio: float = 0.5,
    subterrain: str | None = None,
    height_offset: float = 0.1e-3,
) -> torch.Tensor:
    """
    Generate one X/Y/Z position on the terrain for each environment.
    The X & Y positions will be random points and the Z position will be at the approximate terrain height at that point.

    Args:
        envs_idx: The indices of the environments to generate positions for.
                  If None, positions will be generated for all environments.
                  Can be a list of ints or a torch.Tensor for better performance.
        usable_ratio: How much of the terrain/subterrain area should be used for random positions.
                      For example, 0.25 will only generate positions within the center 25% of the area of the terrain/subterrain.
                      This helps avoid placing things right on th edge of the terrain/subterrain.
        subterrain: The subterrain to generate positions for.
                    If None, positions will be generated for the entire terrain.
        height_offset: The offset to add to the terrain height.

    Returns:
        The position tensor of shape (1, 3)
    """
    if envs_idx is None:
        envs_idx = torch.arange(self.env.num_envs, device=gs.device)
    elif isinstance(envs_idx, list):
        envs_idx = torch.tensor(envs_idx, device=gs.device, dtype=torch.long)

    # Update the position buffer in-place
    self.generate_random_positions(
        output=self._env_pos_buffer,
        out_idx=envs_idx,
        usable_ratio=usable_ratio,
        subterrain=subterrain,
        height_offset=height_offset,
    )
    return self._env_pos_buffer[envs_idx]
```

## `generate_random_positions(num=None, usable_ratio=0.5, subterrain=None, height_offset=0.0001, output=None, out_idx=None)`

Distribute X/Y/Z positions across the terrain or subterrain. The X & Y positions will be random points and the Z position will be at the approximate terrain height at that point.

Parameters:

| Name            | Type     | Description                                                                                                                                                                                                                                                          | Default                                                                                                |
| --------------- | -------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------ |
| `num`           | \`int    | None\`                                                                                                                                                                                                                                                               | The number of positions to generate. Not necessary if output is provided                               |
| `output`        | \`Tensor | None\`                                                                                                                                                                                                                                                               | The position tensor to update in-place.                                                                |
| `out_idx`       | \`Tensor | list[int]                                                                                                                                                                                                                                                            | None\`                                                                                                 |
| `usable_ratio`  | `float`  | How much of the terrain/subterrain area should be used for random positions. For example, 0.25 will only generate positions within the center 25% of the area of the terrain/subterrain. This helps avoid placing things right on th edge of the terrain/subterrain. | `0.5`                                                                                                  |
| `subterrain`    | \`str    | None\`                                                                                                                                                                                                                                                               | The subterrain to generate positions for. If None, positions will be generated for the entire terrain. |
| `height_offset` | `float`  | The offset to add to the terrain height. Since the height is approximate, this can prevent items being placed below the terrain.                                                                                                                                     | `0.0001`                                                                                               |

Returns:

| Type     | Description                            |
| -------- | -------------------------------------- |
| `Tensor` | The positions tensor of shape (num, 3) |

Source code in `genesis_forge/managers/terrain_manager.py`

```
def generate_random_positions(
    self,
    num: int | None = None,
    usable_ratio: float = 0.5,
    subterrain: str | None = None,
    height_offset: float = 0.1e-3,
    output: torch.Tensor | None = None,
    out_idx: torch.Tensor | list[int] | None = None,
) -> torch.Tensor:
    """
    Distribute X/Y/Z positions across the terrain or subterrain.
    The X & Y positions will be random points and the Z position will be at the approximate terrain height at that point.

    Args:
        num: The number of positions to generate. Not necessary if output is provided
        output: The position tensor to update in-place.
        out_idx: The indices of the output position tensor to update.
                 Can be a list of ints or a torch.Tensor for better performance.
        usable_ratio: How much of the terrain/subterrain area should be used for random positions.
                      For example, 0.25 will only generate positions within the center 25% of the area of the terrain/subterrain.
                      This helps avoid placing things right on th edge of the terrain/subterrain.
        subterrain: The subterrain to generate positions for.
                    If None, positions will be generated for the entire terrain.
        height_offset: The offset to add to the terrain height.
                       Since the height is approximate, this can prevent items being placed below the terrain.

    Returns:
        The positions tensor of shape (num, 3)
    """
    # Prep output buffer
    assert (
        output is not None or num is not None
    ), "Either output or num must be provided"
    if output is None:
        output = torch.zeros(num, 3, device=gs.device)
    if out_idx is None:
        out_idx = torch.arange(output.shape[0], device=gs.device)
    elif isinstance(out_idx, list):
        out_idx = torch.tensor(out_idx, device=gs.device, dtype=torch.long)
    if num is None:
        num = out_idx.shape[0]

    # Get total bounds
    bounds = self._bounds
    size = self._size
    if subterrain is not None and subterrain in self._subterrain_bounds:
        size = self._subterrain_size
        bounds = self._subterrain_bounds[subterrain]

    (x_origin, x_max, y_origin, y_max) = bounds
    (x_size, y_size) = size

    # Adjust size based on buffer ratio
    usable_x_size = x_size * usable_ratio
    usable_y_size = y_size * usable_ratio
    buffer_x_size = (x_size - usable_x_size) / 2
    buffer_y_size = (y_size - usable_y_size) / 2

    # Calculate the bounds of the usable area within the section
    x_min = x_origin + buffer_x_size
    x_max = x_origin + x_size - buffer_x_size
    y_min = y_origin + buffer_y_size
    y_max = y_origin + y_size - buffer_y_size

    # Generate random positions
    x_rand = torch.empty(num, device=gs.device, dtype=gs.tc_float)
    y_rand = torch.empty(num, device=gs.device, dtype=gs.tc_float)
    x_rand.uniform_(x_min, x_max)
    y_rand.uniform_(y_min, y_max)

    # Get terrain heights at these positions
    terrain_heights = self.get_terrain_height(x_rand, y_rand)
    output[out_idx, 0] = x_rand
    output[out_idx, 1] = y_rand
    output[out_idx, 2] = terrain_heights + height_offset

    return output
```

## `get_bounds(subterrain=None)`

Get the bounds of the terrain, or subterrain

Source code in `genesis_forge/managers/terrain_manager.py`

```
def get_bounds(
    self, subterrain: str | None = None
) -> tuple[float, float, float, float]:
    """
    Get the bounds of the terrain, or subterrain
    """
    if subterrain is not None and subterrain in self._subterrain_bounds:
        return self._subterrain_bounds[subterrain]
    return self._bounds
```

## `get_terrain_height(x, y)`

Get interpolated terrain height at world coordinates (x, y).

Parameters:

| Name | Type     | Description               | Default    |
| ---- | -------- | ------------------------- | ---------- |
| `x`  | `Tensor` | Tensor of shape (n_envs,) | *required* |
| `y`  | `Tensor` | Tensor of shape (n_envs,) | *required* |

Returns:

| Type     | Description                                    |
| -------- | ---------------------------------------------- |
| `Tensor` | Heights in the torch.Tensor of shape (n_envs,) |

Source code in `genesis_forge/managers/terrain_manager.py`

```
def get_terrain_height(self, x: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
    """
    Get interpolated terrain height at world coordinates (x, y).

    Args:
        x: Tensor of shape (n_envs,)
        y: Tensor of shape (n_envs,)

    Returns:
        Heights in the torch.Tensor of shape (n_envs,)
    """
    n_envs = x.shape[0]

    # No height field, so we can assume the height is consistent
    if self._height_field is None:
        self._heights_buffer[:n_envs] = self._origin[2]
        return self._heights_buffer[:n_envs]

    # Normalize coordinates to [-1, 1] range expected by grid_sample
    (x_min, x_max, y_min, y_max) = self._bounds

    # Use pre-allocated buffer and in-place operations to avoid memory allocation
    norm_x = self._norm_coords_buffer[:n_envs, 0]
    norm_y = self._norm_coords_buffer[:n_envs, 1]

    # In-place normalization to avoid creating new tensors
    # norm_x = 2 * (norm_x - x_min) / (x_max - x_min) - 1
    norm_x.copy_(x)
    norm_x.sub_(x_min)
    norm_x.div_(x_max - x_min)
    norm_x.mul_(2)
    norm_x.sub_(1)
    # norm_y = 2 * (norm_y - y_min) / (y_max - y_min) - 1
    norm_y.copy_(y)
    norm_y.sub_(y_min)
    norm_y.div_(y_max - y_min)
    norm_y.mul_(2)
    norm_y.sub_(1)

    # Use pre-allocated grid buffer
    grid = self._grid_buffer[:n_envs]
    grid[:, 0, 0, 0] = norm_x
    grid[:, 0, 0, 1] = norm_y

    # Border padding mode isn't supported on Mac GPU (mps)
    # https://github.com/pytorch/pytorch/issues/125098
    if gs.device.type == "mps":
        padding_mode = "zeros"
        grid.clamp_(-1, 1)
    else:
        padding_mode = "border"

    # Use the height field directly without expansion to save memory
    # The height field is already in the correct format (1, height, width)
    interpolated = F.grid_sample(
        self._height_field.unsqueeze(0).expand(n_envs, -1, -1, -1),
        grid,
        mode="bilinear",
        padding_mode=padding_mode,
        align_corners=True,
    )

    # Extract the height values at the specific coordinates
    heights = self._heights_buffer[:n_envs]
    heights.copy_(interpolated[:, 0, 0, 0])

    return heights
```

## `reset(envs_idx=None)`

One or more environments have been reset

Source code in `genesis_forge/managers/base.py`

```
def reset(self, envs_idx: list[int] | None = None):
    """One or more environments have been reset"""
    pass
```

## `step()`

Called when the environment is stepped

Source code in `genesis_forge/managers/base.py`

```
def step(self):
    """Called when the environment is stepped"""
    pass
```

# Action

These managers take in actions and convert them to DOF actuator values.

# PositionActionManager

Bases: `BaseActionManager`

Converts actions to DOF positions, using affine transformations (scale and offset).

.. math::

position = offset + scaling * action

If `use_default_offset` is `True`, the `offset` will be set to the `default_pos` value for each DOF/joint.

Parameters:

| Name                      | Type                  | Description                                                                                                                                                    | Default                                                                                                                     |
| ------------------------- | --------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------- |
| `env`                     | `GenesisEnv`          | The environment to manage the DOF actuators for.                                                                                                               | *required*                                                                                                                  |
| `actuator_manager`        | \`ActuatorManager     | None\`                                                                                                                                                         | The actuator manager which is used to setup and control the DOF joints.                                                     |
| `actuator_joints`         | \`list[str]           | str\`                                                                                                                                                          | Which joints of the actuator manager that this action manager will control. These can be full names or regular expressions. |
| `scale`                   | \`float               | dict[str, float]\`                                                                                                                                             | How much to scale the action.                                                                                               |
| `offset`                  | \`float               | dict[str, float]\`                                                                                                                                             | Offset factor for the action.                                                                                               |
| `use_default_offset`      | `bool`                | Whether to use default joint positions configured in the articulation asset as offset. Defaults to True.                                                       | `True`                                                                                                                      |
| `clip`                    | \`tuple[float, float] | dict\[str, tuple[float, float]\]\`                                                                                                                             | Clip the action values to the range. If omitted, the action values will automatically be clipped to the joint limits.       |
| `soft_limit_scale_factor` | `float`               | Scales the clip range of all limits by this factor around the midpoint of each joint's limits to establish a safety region within the limits. Defaults to 1.0. | `1.0`                                                                                                                       |
| `quiet_action_errors`     | `bool`                | Whether to quiet action errors.                                                                                                                                | `False`                                                                                                                     |
| `delay_step`              | `int`                 | The number of steps to delay the actions for. This is an easy way to emulate the latency in the system.                                                        | `0`                                                                                                                         |

Example::

```
class MyEnv(ManagedEnvironment):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # ...define scene and robot...

    def config(self):
        self.actuator_manager = ActuatorManager(
            self,
            joint_names=".*",
            default_pos={".*": 0.0},
            kp=50,
            kv=0.5,
            max_force=8.0,
        )
        self.action_manager = PositionActionManager(
            self,
            scale=0.5,
            use_default_offset=True,
            actuator_manager=self.actuator_manager,
            actuator_joints=[".*"], # optional joint filter
        )
```

Example using the manager directly::

```
class MyEnv(GenesisEnv):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # ...define scene and robot...

        self.actuator_manager = ActuatorManager(
            self,
            joint_names=".*",
            default_pos={".*": 0.0},
            kp=50,
            kv=0.5,
            max_force=8.0,
        )
        self.action_manager = PositionActionManager(
            self,
            scale=0.5,
            offset=0.0,
            use_default_offset=True,
        )

    def build(self):
        super().build()
        self.actuator_manager.build()
        self.action_manager.build()

    step(self, actions: torch.Tensor) -> None:
        super().step(actions)
        self.action_manager.step(actions)

        # ...do other step things...

    reset(self, envs_idx: list[int] = None) -> None:
        super().reset(envs_idx)
        self.actuator_manager.reset(envs_idx)
        self.action_manager.reset(envs_idx)

        # ...do other reset things...
```

Source code in `genesis_forge/managers/action/position_action_manager.py`

```
def __init__(
    self,
    env: GenesisEnv,
    actuator_manager: ActuatorManager | None = None,
    actuator_joints: list[str] | str = ".*",
    scale: float | dict[str, float] = 1.0,
    offset: float | dict[str, float] = 0.0,
    clip: tuple[float, float] | dict[str, tuple[float, float]] = None,
    soft_limit_scale_factor: float = 1.0,
    use_default_offset: bool = True,
    quiet_action_errors: bool = False,
    delay_step: int = 0,
    **kwargs,
):
    super().__init__(
        env,
        delay_step=delay_step,
        actuator_manager=actuator_manager,
        actuator_joints=actuator_joints,
        **kwargs,
    )
    self._offset_cfg = ensure_dof_pattern(offset)
    self._scale_cfg = ensure_dof_pattern(scale)
    self._clip_cfg = ensure_dof_pattern(clip)
    self._soft_limit_scale_factor = soft_limit_scale_factor
    self._quiet_action_errors = quiet_action_errors
    self._enabled_dof = None
    self._use_default_offset = use_default_offset

    self._dofs_pos_buffer: torch.Tensor = None

    if use_default_offset and offset != 0.0:
        raise ValueError("Cannot set both use_default_offset and offset")
```

## `action_space`

Returns the actions space for the environment, based on the number of DOFs defined in this action manager.

## `actions`

The processed actions for for the current step.

## `actuator_dof_filter`

An index filter for the actuator DOF buffer values.

## `actuator_manager`

Get the actuator manager.

## `actuators`

## `default_dofs_pos`

Return the default DOF positions.

## `dofs`

Get a dictionary of the DOF names and their indices

## `dofs_idx`

Get the indices of the DOFs that this action manager controls.

## `enabled = True`

## `env = env`

## `last_actions`

The processed actions for for the previous step.

## `num_actions`

Get the number of actions.

## `raw_actions`

The actions received from the policy, before being processed.

## `type = type`

## `build()`

Builds the manager and initialized all the buffers.

Source code in `genesis_forge/managers/action/position_action_manager.py`

```
def build(self):
    """
    Builds the manager and initialized all the buffers.
    """
    super().build()

    # Define the clip values
    lower_limit, upper_limit = self.actuator_manager.get_dofs_limits(self.dofs_idx)
    self._clip_values = torch.stack([lower_limit, upper_limit], dim=1)
    if self._clip_cfg is not None:
        self._get_dof_value_tensor(self._clip_cfg, output=self._clip_values)
    if self._soft_limit_scale_factor != 1.0:
        midpoint = (self._clip_values[:, 0] + self._clip_values[:, 1]) * 0.5
        half_range = (self._clip_values[:, 1] - self._clip_values[:, 0]) * 0.5 * self._soft_limit_scale_factor
        self._clip_values[:, 0] = midpoint - half_range
        self._clip_values[:, 1] = midpoint + half_range

    # Scale
    self._scale_values = None
    if self._scale_cfg is not None:
        self._scale_values = self._get_dof_value_tensor(self._scale_cfg)

    # Offset
    self._offset_values = None
    if self._use_default_offset:
        self._offset_values = self.default_dofs_pos
    else:
        offset = self._offset_cfg if self._offset_cfg is not None else 0.0
        self._offset_values = self._get_dof_value_tensor(offset)
```

## `get_actions()`

Get the current actions for the environments.

Source code in `genesis_forge/managers/action/base.py`

```
def get_actions(self) -> torch.Tensor:
    """
    Get the current actions for the environments.
    """
    if self._actions is None:
        return torch.zeros((self.env.num_envs, self.num_actions))
    return self._actions
```

## `get_actions_dict(env_idx=0)`

Get the latest actions for an environment as a dictionary of DOF names and values.

Source code in `genesis_forge/managers/action/base.py`

```
def get_actions_dict(self, env_idx: int = 0) -> dict[str, float]:
    """
    Get the latest actions for an environment as a dictionary of DOF names and values.
    """
    return {
        name: value.item()
        for name, value in zip(
            self.dofs.keys(), self._actions[env_idx, :]
        )
    }
```

## `get_dofs_force(clip_to_max_force=False)`

A wrapper for `RigidEntity.get_dofs_force` that returns the force experienced by the controlled DOFs.

Parameters:

| Name                | Type   | Description                                                                                                      | Default |
| ------------------- | ------ | ---------------------------------------------------------------------------------------------------------------- | ------- |
| `clip_to_max_force` | `bool` | Clip the force returned to the maximum force defined by the max_force parameter defined in the actuator manager. | `False` |

Returns:

| Name    | Type     | Description                                |
| ------- | -------- | ------------------------------------------ |
| `force` | `Tensor` | torch.Tensor, shape (n_envs, n_dofs)       |
|         | `Tensor` | The force experienced by the enabled DOFs. |

Source code in `genesis_forge/managers/action/base.py`

```
def get_dofs_force(self, clip_to_max_force: bool = False) -> torch.Tensor:
    """
    A wrapper for `RigidEntity.get_dofs_force` that returns the force experienced by the controlled DOFs.

    Args:
        clip_to_max_force: Clip the force returned to the maximum force defined by the `max_force` parameter
                           defined in the actuator manager.

    Returns:
        force: torch.Tensor, shape (n_envs, n_dofs)
        The force experienced by the enabled DOFs.
    """
    return self.actuator_manager.get_dofs_force(
        clip_to_max_force=clip_to_max_force, dofs_idx=self.dofs_idx
    )
```

## `get_dofs_limits()`

A wrapper for `RigidEntity.get_dofs_limit` that returns the limits of the controlled DOFs.

Returns:

| Name          | Type     | Description                                                                                                       |
| ------------- | -------- | ----------------------------------------------------------------------------------------------------------------- |
| `lower_limit` | `Tensor` | torch.Tensor, shape (n_dofs,) or (n_envs, n_dofs) The lower limit of the positional limits for the entity's dofs. |
| `upper_limit` | `Tensor` | torch.Tensor, shape (n_dofs,) or (n_envs, n_dofs) The upper limit of the positional limits for the entity's dofs. |

Source code in `genesis_forge/managers/action/base.py`

```
def get_dofs_limits(self) -> tuple[torch.Tensor, torch.Tensor]:
    """
    A wrapper for `RigidEntity.get_dofs_limit` that returns the limits of the controlled DOFs.

    Returns:
        lower_limit: torch.Tensor, shape (n_dofs,) or (n_envs, n_dofs)
                     The lower limit of the positional limits for the entity's dofs.
        upper_limit: torch.Tensor, shape (n_dofs,) or (n_envs, n_dofs)
                     The upper limit of the positional limits for the entity's dofs.
    """
    return self.actuator_manager.get_dofs_limits(dofs_idx=self.dofs_idx)
```

## `get_dofs_position()`

A wrapper for `RigidEntity.get_dofs_limits` that returns the position limits of the controlled DOFs.

Returns:

| Name       | Type     | Description                                                                                   |
| ---------- | -------- | --------------------------------------------------------------------------------------------- |
| `position` | `Tensor` | torch.Tensor, shape (n_envs, n_dofs) The position of the DOFs managed by this action manager. |

Source code in `genesis_forge/managers/action/base.py`

```
def get_dofs_position(self) ->  torch.Tensor:
    """
    A wrapper for `RigidEntity.get_dofs_limits` that returns the position limits of the controlled DOFs.

    Returns:
        position: torch.Tensor, shape (n_envs, n_dofs)
                  The position of the DOFs managed by this action manager.
    """
    return self.actuator_manager.get_dofs_position(dofs_idx=self.dofs_idx)
```

## `get_dofs_velocity(clip=None)`

A wrapper for `RigidEntity.get_dofs_velocity` that returns the current velocity of the controlled DOFs.

Parameters:

| Name   | Type                  | Description                    | Default |
| ------ | --------------------- | ------------------------------ | ------- |
| `clip` | `tuple[float, float]` | Range to clip the velocity to. | `None`  |

Returns:

| Name       | Type     | Description                                                      |
| ---------- | -------- | ---------------------------------------------------------------- |
| `velocity` | `Tensor` | torch.Tensor, shape (n_envs, n_dofs)                             |
|            | `Tensor` | The velocity of the enabled DOFs managed by this action manager. |

Source code in `genesis_forge/managers/action/base.py`

```
def get_dofs_velocity(self, clip: tuple[float, float] = None) -> torch.Tensor:
    """
    A wrapper for `RigidEntity.get_dofs_velocity` that returns the current velocity of the controlled DOFs.

    Args:
        clip: Range to clip the velocity to.

    Returns:
        velocity: torch.Tensor, shape (n_envs, n_dofs)
        The velocity of the enabled DOFs managed by this action manager.
    """
    return self.actuator_manager.get_dofs_velocity(
        clip=clip, dofs_idx=self.dofs_idx
    )
```

## `process_actions(actions)`

Convert the actions to position commands, and clamp them to the limits.

Parameters:

| Name      | Type     | Description                          | Default    |
| --------- | -------- | ------------------------------------ | ---------- |
| `actions` | `Tensor` | The incoming step actions to handle. | *required* |

Returns:

| Type     | Description                       |
| -------- | --------------------------------- |
| `Tensor` | The actions as position commands. |

Source code in `genesis_forge/managers/action/position_action_manager.py`

```
def process_actions(self, actions: torch.Tensor) -> torch.Tensor:
    """
    Convert the actions to position commands, and clamp them to the limits.

    Args:
        actions: The incoming step actions to handle.

    Returns:
        The actions as position commands.
    """
    # Validate actions
    if not self._quiet_action_errors:
        if torch.isnan(actions).any():
            print(f"ERROR: NaN actions received! Actions: {actions}")
        if torch.isinf(actions).any():
            print(f"ERROR: Infinite actions received! Actions: {actions}")

    # Process actions
    actions = actions * self._scale_values + self._offset_values
    actions = torch.clamp(
        actions,
        min=self._clip_values[:, 0],
        max=self._clip_values[:, 1],
    )
    return actions
```

## `reset(envs_idx)`

Reset environments.

Source code in `genesis_forge/managers/action/base.py`

```
def reset(self, envs_idx: list[int] | None):
    """Reset environments."""
    if (
        self._delay_step > 0
        and len(self._action_delay_buffer) < self._delay_step
        and self.num_actions > 0
    ):
        while len(self._action_delay_buffer) < self._delay_step:
            self._action_delay_buffer.append(
                torch.zeros((self.env.num_envs, self.num_actions), device=gs.device)
            )
```

## `send_actions_to_simulation(actions)`

Sends the actions as position commands to the actuators in the simulation.

Source code in `genesis_forge/managers/action/position_action_manager.py`

```
def send_actions_to_simulation(self, actions: torch.Tensor) -> torch.Tensor:
    """
    Sends the actions as position commands to the actuators in the simulation.
    """
    actions = self.get_actions()
    self.actuator_manager.control_dofs_position(actions, self.dofs_idx)
```

## `step(actions)`

Handle actions received in this step.

Source code in `genesis_forge/managers/action/base.py`

```
def step(self, actions: torch.Tensor) -> None:
    """
    Handle actions received in this step.
    """
    # Action delay buffer
    if self._delay_step > 0:
        self._action_delay_buffer.insert(0, actions)
        actions = self._action_delay_buffer.pop()

    # Copy the actions into the manager buffer
    self._raw_actions = actions
    if self._actions is None:
        self._actions = torch.empty_like(actions, device=gs.device)
        self._last_actions = torch.zeros_like(actions, device=gs.device)
    self._last_actions[:] = self._actions[:]

    # Process the actions
    self._actions[:] = self.process_actions(self._raw_actions[:])

    return self._actions
```

# PositionWithinLimitsActionManager

Bases: `PositionActionManager`

This is similar to `PositionActionManager` but converts actions from the range -1.0 - 1.0 to DOF positions within the limits of the actuators.

Parameters:

| Name                      | Type                  | Description                                                                                                    | Default                                                                                                                                              |
| ------------------------- | --------------------- | -------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------- |
| `env`                     | `GenesisEnv`          | The environment to manage the DOF actuators for.                                                               | *required*                                                                                                                                           |
| `actuator_manager`        | \`ActuatorManager     | None\`                                                                                                         | The actuator manager which is used to setup and control the DOF joints.                                                                              |
| `actuator_joints`         | \`list[str]           | str\`                                                                                                          | Which joints of the actuator manager that this action manager will control. These can be full names or regular expressions.                          |
| `limit`                   | \`tuple[float, float] | dict\[str, tuple[float, float]\]\`                                                                             | A dictionary of DOF name patterns and their position limits. If omitted, the limits will be set to the limits of the actuators defined in the model. |
| `soft_limit_scale_factor` | `float`               | Scales the range of all limits by this factor to establish a safety region within the limits. Defaults to 1.0. | `1.0`                                                                                                                                                |
| `quiet_action_errors`     | `bool`                | Whether to quiet action errors.                                                                                | `False`                                                                                                                                              |
| `delay_step`              | `int`                 | The number of steps to delay the actions for. This is an easy way to emulate the latency in the system.        | `0`                                                                                                                                                  |

Simple example using the limits defined in the model::

```
class MyEnv(ManagedEnvironment):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def config(self):
        self.actuator_manager = ActuatorManager(
            self,
            joint_names=".*",
            default_pos={".*": 0.0},
            kp={".*": 50},
            kv={".*": 0.5},
        )
        self.action_manager = PositionalActionManager(
            self,
            actuator_manager=self.actuator_manager,
            actuator_joints=[".*"], # optional joint filter
        )
```

Example defining custom limits::

```
class MyEnv(ManagedEnvironment):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def config(self):
        self.actuator_manager = ActuatorManager(
            self,
            joint_names=".*",
            default_pos={".*": 0.0},
            kp={".*": 50},
            kv={".*": 0.5},
        )
        self.action_manager = PositionalActionManager(
            self,
            actuator_manager=self.actuator_manager,
            limit = {
                ".*_Hip": (-1.0, 1.0),
                ".*_Femur": (-1.5, 1.2),
            },
        )
```

Source code in `genesis_forge/managers/action/position_within_limits.py`

```
def __init__(
    self,
    env: GenesisEnv,
    actuator_manager: ActuatorManager | None = None,
    actuator_joints: list[str] | str = ".*",
    quiet_action_errors: bool = False,
    limit: tuple[float, float] | dict[str, tuple[float, float]] = {},
    soft_limit_scale_factor: float = 1.0,
    delay_step: int = 0,
    **kwargs,
):
    super().__init__(
        env,
        actuator_manager=actuator_manager,
        actuator_joints=actuator_joints,
        quiet_action_errors=quiet_action_errors,
        delay_step=delay_step,
        **kwargs,
    )
    self._limit_cfg = ensure_dof_pattern(limit)
    self._soft_limit_scale_factor = soft_limit_scale_factor
```

## `action_space`

Returns the actions space for the environment, based on the number of DOFs defined in this action manager.

## `actions`

The processed actions for for the current step.

## `actuator_dof_filter`

An index filter for the actuator DOF buffer values.

## `actuator_manager`

Get the actuator manager.

## `actuators`

## `default_dofs_pos`

Return the default DOF positions.

## `dofs`

Get a dictionary of the DOF names and their indices

## `dofs_idx`

Get the indices of the DOFs that this action manager controls.

## `enabled = True`

## `env = env`

## `last_actions`

The processed actions for for the previous step.

## `num_actions`

Get the number of actions.

## `raw_actions`

The actions received from the policy, before being processed.

## `type = type`

## `build()`

Builds the manager and initialized all the buffers.

Source code in `genesis_forge/managers/action/position_within_limits.py`

```
def build(self):
    """
    Builds the manager and initialized all the buffers.
    """
    super().build()

    lower, upper = self._get_dof_limits()
    lower = lower.unsqueeze(0).expand(self.env.num_envs, -1)
    upper = upper.unsqueeze(0).expand(self.env.num_envs, -1)
    self._offset = (upper + lower) * 0.5
    self._scale = (upper - lower) * 0.5 * self._soft_limit_scale_factor
```

## `get_actions()`

Get the current actions for the environments.

Source code in `genesis_forge/managers/action/base.py`

```
def get_actions(self) -> torch.Tensor:
    """
    Get the current actions for the environments.
    """
    if self._actions is None:
        return torch.zeros((self.env.num_envs, self.num_actions))
    return self._actions
```

## `get_actions_dict(env_idx=0)`

Get the latest actions for an environment as a dictionary of DOF names and values.

Source code in `genesis_forge/managers/action/base.py`

```
def get_actions_dict(self, env_idx: int = 0) -> dict[str, float]:
    """
    Get the latest actions for an environment as a dictionary of DOF names and values.
    """
    return {
        name: value.item()
        for name, value in zip(
            self.dofs.keys(), self._actions[env_idx, :]
        )
    }
```

## `get_dofs_force(clip_to_max_force=False)`

A wrapper for `RigidEntity.get_dofs_force` that returns the force experienced by the controlled DOFs.

Parameters:

| Name                | Type   | Description                                                                                                      | Default |
| ------------------- | ------ | ---------------------------------------------------------------------------------------------------------------- | ------- |
| `clip_to_max_force` | `bool` | Clip the force returned to the maximum force defined by the max_force parameter defined in the actuator manager. | `False` |

Returns:

| Name    | Type     | Description                                |
| ------- | -------- | ------------------------------------------ |
| `force` | `Tensor` | torch.Tensor, shape (n_envs, n_dofs)       |
|         | `Tensor` | The force experienced by the enabled DOFs. |

Source code in `genesis_forge/managers/action/base.py`

```
def get_dofs_force(self, clip_to_max_force: bool = False) -> torch.Tensor:
    """
    A wrapper for `RigidEntity.get_dofs_force` that returns the force experienced by the controlled DOFs.

    Args:
        clip_to_max_force: Clip the force returned to the maximum force defined by the `max_force` parameter
                           defined in the actuator manager.

    Returns:
        force: torch.Tensor, shape (n_envs, n_dofs)
        The force experienced by the enabled DOFs.
    """
    return self.actuator_manager.get_dofs_force(
        clip_to_max_force=clip_to_max_force, dofs_idx=self.dofs_idx
    )
```

## `get_dofs_limits()`

A wrapper for `RigidEntity.get_dofs_limit` that returns the limits of the controlled DOFs.

Returns:

| Name          | Type     | Description                                                                                                       |
| ------------- | -------- | ----------------------------------------------------------------------------------------------------------------- |
| `lower_limit` | `Tensor` | torch.Tensor, shape (n_dofs,) or (n_envs, n_dofs) The lower limit of the positional limits for the entity's dofs. |
| `upper_limit` | `Tensor` | torch.Tensor, shape (n_dofs,) or (n_envs, n_dofs) The upper limit of the positional limits for the entity's dofs. |

Source code in `genesis_forge/managers/action/base.py`

```
def get_dofs_limits(self) -> tuple[torch.Tensor, torch.Tensor]:
    """
    A wrapper for `RigidEntity.get_dofs_limit` that returns the limits of the controlled DOFs.

    Returns:
        lower_limit: torch.Tensor, shape (n_dofs,) or (n_envs, n_dofs)
                     The lower limit of the positional limits for the entity's dofs.
        upper_limit: torch.Tensor, shape (n_dofs,) or (n_envs, n_dofs)
                     The upper limit of the positional limits for the entity's dofs.
    """
    return self.actuator_manager.get_dofs_limits(dofs_idx=self.dofs_idx)
```

## `get_dofs_position()`

A wrapper for `RigidEntity.get_dofs_limits` that returns the position limits of the controlled DOFs.

Returns:

| Name       | Type     | Description                                                                                   |
| ---------- | -------- | --------------------------------------------------------------------------------------------- |
| `position` | `Tensor` | torch.Tensor, shape (n_envs, n_dofs) The position of the DOFs managed by this action manager. |

Source code in `genesis_forge/managers/action/base.py`

```
def get_dofs_position(self) ->  torch.Tensor:
    """
    A wrapper for `RigidEntity.get_dofs_limits` that returns the position limits of the controlled DOFs.

    Returns:
        position: torch.Tensor, shape (n_envs, n_dofs)
                  The position of the DOFs managed by this action manager.
    """
    return self.actuator_manager.get_dofs_position(dofs_idx=self.dofs_idx)
```

## `get_dofs_velocity(clip=None)`

A wrapper for `RigidEntity.get_dofs_velocity` that returns the current velocity of the controlled DOFs.

Parameters:

| Name   | Type                  | Description                    | Default |
| ------ | --------------------- | ------------------------------ | ------- |
| `clip` | `tuple[float, float]` | Range to clip the velocity to. | `None`  |

Returns:

| Name       | Type     | Description                                                      |
| ---------- | -------- | ---------------------------------------------------------------- |
| `velocity` | `Tensor` | torch.Tensor, shape (n_envs, n_dofs)                             |
|            | `Tensor` | The velocity of the enabled DOFs managed by this action manager. |

Source code in `genesis_forge/managers/action/base.py`

```
def get_dofs_velocity(self, clip: tuple[float, float] = None) -> torch.Tensor:
    """
    A wrapper for `RigidEntity.get_dofs_velocity` that returns the current velocity of the controlled DOFs.

    Args:
        clip: Range to clip the velocity to.

    Returns:
        velocity: torch.Tensor, shape (n_envs, n_dofs)
        The velocity of the enabled DOFs managed by this action manager.
    """
    return self.actuator_manager.get_dofs_velocity(
        clip=clip, dofs_idx=self.dofs_idx
    )
```

## `process_actions(actions)`

Convert the actions to position commands within the limits.

Parameters:

| Name      | Type     | Description                          | Default    |
| --------- | -------- | ------------------------------------ | ---------- |
| `actions` | `Tensor` | The incoming step actions to handle. | *required* |

Returns:

| Type     | Description                       |
| -------- | --------------------------------- |
| `Tensor` | The actions as position commands. |

Source code in `genesis_forge/managers/action/position_within_limits.py`

```
def process_actions(self, actions: torch.Tensor) -> torch.Tensor:
    """
    Convert the actions to position commands within the limits.

    Args:
        actions: The incoming step actions to handle.

    Returns:
        The actions as position commands.
    """
    # Convert the action from -1 to 1, to absolute position within the actuator limits
    actions = actions.clamp(-1.0, 1.0)
    actions = actions * self._scale + self._offset
    return actions
```

## `reset(envs_idx)`

Reset environments.

Source code in `genesis_forge/managers/action/base.py`

```
def reset(self, envs_idx: list[int] | None):
    """Reset environments."""
    if (
        self._delay_step > 0
        and len(self._action_delay_buffer) < self._delay_step
        and self.num_actions > 0
    ):
        while len(self._action_delay_buffer) < self._delay_step:
            self._action_delay_buffer.append(
                torch.zeros((self.env.num_envs, self.num_actions), device=gs.device)
            )
```

## `send_actions_to_simulation(actions)`

Sends the actions as position commands to the actuators in the simulation.

Source code in `genesis_forge/managers/action/position_action_manager.py`

```
def send_actions_to_simulation(self, actions: torch.Tensor) -> torch.Tensor:
    """
    Sends the actions as position commands to the actuators in the simulation.
    """
    actions = self.get_actions()
    self.actuator_manager.control_dofs_position(actions, self.dofs_idx)
```

## `step(actions)`

Handle actions received in this step.

Source code in `genesis_forge/managers/action/base.py`

```
def step(self, actions: torch.Tensor) -> None:
    """
    Handle actions received in this step.
    """
    # Action delay buffer
    if self._delay_step > 0:
        self._action_delay_buffer.insert(0, actions)
        actions = self._action_delay_buffer.pop()

    # Copy the actions into the manager buffer
    self._raw_actions = actions
    if self._actions is None:
        self._actions = torch.empty_like(actions, device=gs.device)
        self._last_actions = torch.zeros_like(actions, device=gs.device)
    self._last_actions[:] = self._actions[:]

    # Process the actions
    self._actions[:] = self.process_actions(self._raw_actions[:])

    return self._actions
```

# Command

Sends regular commands, such as a velocity direction, to your robot

# CommandManager

Bases: `BaseManager`

Generates a command from uniform distribution of values. You can use this to regularly sample a commands for each environment, such as a height command, a target destination, or a specific pose.

Parameters:

| Name                | Type           | Description                                                            | Default    |
| ------------------- | -------------- | ---------------------------------------------------------------------- | ---------- |
| `env`               | `GenesisEnv`   | The environment to control                                             | *required* |
| `range`             | `CommandRange` | The number range, or dict of ranges, to generate target command(s) for | *required* |
| `resample_time_sec` | `float`        | The time interval between changing the command                         | `5.0`      |

Example::

```
class MyEnv(GenesisEnv):
    def config(self):
        # Create a height command
        self.height_command = CommandManager(self, range=(0.1, 0.2))

        # Rewards
        RewardManager(
            self,
            logging_enabled=True,
            cfg={
                "base_height_target": {
                    "weight": -50.0,
                    "fn": rewards.base_height,
                    "params": {
                        "height_command": self.height_command,
                    },
                },
                # ... other rewards ...
            },
        )

        # Observations
        ObservationManager(
            self,
            cfg={
                "height_cmd": {"fn": self.height_command.observation},
                # ... other observations ...
            },
        )
```

Source code in `genesis_forge/managers/command/command_manager.py`

```
def __init__(
    self,
    env: GenesisEnv,
    range: CommandRange,
    resample_time_sec: float = 5.0,
):
    super().__init__(env, type="command")

    self._range = range
    self.resample_time_sec = resample_time_sec
    self._external_controller = None
    self._gamepad_cfg = None
    self._gamepad_axis_command_buffer = None

    num_ranges = len(range) if isinstance(range, dict) else 1
    self._command = torch.zeros(env.num_envs, num_ranges, device=gs.device)
    self._range_idx = {}
    if isinstance(range, dict):
        self._range_idx = {key: i for i, key in enumerate(range.keys())}
```

## `command`

The desired command value. Shape is (num_envs, num_ranges).

## `enabled = True`

## `env = env`

## `range`

The range of values to generate target command(s) for.

## `resample_time_sec`

The time interval (in seconds) between changing the command for each environment.

## `type = type`

## `build()`

Called when the scene is built

Source code in `genesis_forge/managers/base.py`

```
def build(self):
    """Called when the scene is built"""
    pass
```

## `get_command(range_key)`

If the range is a dict, get the command values for the given key.

Source code in `genesis_forge/managers/command/command_manager.py`

```
def get_command(self, range_key: str) -> torch.Tensor:
    """
    If the range is a dict, get the command values for the given key.
    """
    if not isinstance(self._range, dict):
        raise ValueError("The range is not a dict")
    return self._command[:, self._range_idx[range_key]]
```

## `get_command_idx(key)`

If the range is a dict, get the command index for the given key.

Source code in `genesis_forge/managers/command/command_manager.py`

```
def get_command_idx(self, key: str) -> int:
    """
    If the range is a dict, get the command index for the given key.
    """
    if not isinstance(self._range, dict):
        raise ValueError("The range is not a dict")
    return self._range_idx[key]
```

## `increment_range(range_key, increment, limit=None)`

Increment a command range target values by the given amount, with an optional limit.

Both increment and limit can be passed as a single float or a tuple of two floats. When they are tuples, they represent the increment and limit for both min and max. When they are a single float, they will be applied to both min and max as (-value, +value).

Example::

# Increment the range by 0.5 (+/-) for both min and max

command_manager.increment_range("height", 0.5)

```
# Increment the range min by -0.25 and max by 1.0
command_manager.increment_range("height", (-0.25, 1.0))

# Increment the range by (-0.25, 1.0) and keep min above -0.5 and max below 2.0
command_manager.increment_range("height", (-0.25, 1.0), (-0.5, 2.0))
```

Parameters:

| Name        | Type    | Description                                | Default                                                                                                                                                                                                                     |
| ----------- | ------- | ------------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `range_key` | `str`   | The key of the command range to increment. | *required*                                                                                                                                                                                                                  |
| `increment` | \`float | tuple[float, float]\`                      | The amount to increment the min & max range values by (+/-). If this is a tuple, it will be: (min_increment, max_increment). If this is a single float, it will be applied to both min and max as (-increment, +increment). |
| `limit`     | \`float | tuple[float, float]\`                      | Do not set the values beyond this limit range. This is a tuple of (min_limit, max_limit).                                                                                                                                   |

Source code in `genesis_forge/managers/command/command_manager.py`

```
def increment_range(
    self,
    range_key: str,
    increment: float | tuple[float, float],
    limit: float | tuple[float, float] = None,
):
    """
    Increment a command range target values by the given amount, with an optional limit.

    Both increment and limit can be passed as a single float or a tuple of two floats.
    When they are tuples, they represent the increment and limit for both min and max.
    When they are a single float, they will be applied to both min and max as (-value, +value).

    Example::
        # Increment the range by 0.5 (+/-) for both min and max
        command_manager.increment_range("height", 0.5)

        # Increment the range min by -0.25 and max by 1.0
        command_manager.increment_range("height", (-0.25, 1.0))

        # Increment the range by (-0.25, 1.0) and keep min above -0.5 and max below 2.0
        command_manager.increment_range("height", (-0.25, 1.0), (-0.5, 2.0))

    Args:
        range_key: The key of the command range to increment.
        increment: The amount to increment the min & max range values by (+/-).
                   If this is a tuple, it will be: `(min_increment, max_increment)`.
                   If this is a single float, it will be applied to both min and max as (-increment, +increment).
        limit: Do not set the values beyond this limit range.
               This is a tuple of `(min_limit, max_limit)`.
    """
    # Get the range to increment, and ensure it is a list, not a tuple (tuples are immutable)
    if not isinstance(self.range, dict):
        raise ValueError("Cannot increment a non-dict range item")
    range_item = self._range.get(range_key, None)
    if range_item is None:
        raise ValueError(f"Range item {range_key} not found")
    range_item = list[float](range_item)  # Ensure it's a list, not a tuple

    # Expand single increment/limit values to tuples
    if not isinstance(increment, (list, tuple)):
        increment = (-increment, increment)
    if limit is not None and not isinstance(limit, (list, tuple)):
        limit = (-limit, limit)

    # Increment the range values
    for i, value in enumerate(range_item):
        value += increment[i]
        if limit is not None:
            if increment[i] > 0:
                value = min(value, limit[i])
            else:
                value = max(value, limit[i])
        range_item[i] = value
    self._range[range_key] = range_item
```

## `observation(env)`

Function that returns the current command for each environment.

Source code in `genesis_forge/managers/command/command_manager.py`

```
def observation(self, env: GenesisEnv) -> torch.Tensor:
    """Function that returns the current command for each environment."""
    return self.command
```

## `resample_command(env_ids)`

Create a new command for the given environment ids.

Source code in `genesis_forge/managers/command/command_manager.py`

```
def resample_command(self, env_ids: list[int]):
    """Create a new command for the given environment ids."""

    # Get range values (this might have changed since init due to curriculum training)
    ranges = None
    if isinstance(self._range, dict):
        ranges = list(self._range.values())
    else:
        ranges = [self._range]

    # Resample the command
    for i in range(self._command.shape[1]):
        buffer = torch.empty(len(env_ids), device=gs.device).uniform_(*ranges[i])
        self._command[env_ids, i] = buffer
```

## `reset(env_ids=None)`

One or more environments have been reset

Source code in `genesis_forge/managers/command/command_manager.py`

```
def reset(self, env_ids: list[int] | None = None):
    """One or more environments have been reset"""
    if not self.enabled:
        return
    if env_ids is None:
        env_ids = torch.arange(self.env.num_envs, device=gs.device)
    self.resample_command(env_ids)
```

## `set_command(range_key, value, envs_idx=None)`

Update a command value for selected environments.

Source code in `genesis_forge/managers/command/command_manager.py`

```
def set_command(
    self,
    range_key: str,
    value: torch.Tensor,
    envs_idx: list[int] | None = None,
):
    """
    Update a command value for selected environments.
    """
    if not isinstance(self._range, dict):
        raise ValueError("The range is not a dict")
    if envs_idx is None:
        self._command[:, self._range_idx[range_key]] = value
    else:
        self._command[envs_idx, self._range_idx[range_key]] = value
```

## `step()`

Resample the command if necessary

Source code in `genesis_forge/managers/command/command_manager.py`

```
def step(self):
    """Resample the command if necessary"""
    if not self.enabled or self._external_controller is not None:
        return

    resample_command_envs = (
        (self.env.episode_length % self._resample_steps == 0)
        .nonzero(as_tuple=False)
        .reshape((-1,))
    )
    self.resample_command(resample_command_envs)
```

## `use_external_controller(controller)`

Bypass the internal command controller, and generate the command values with an external control function. This can be used to connect a gamepad, joystick, or other external controller to the command manager.

Parameters:

| Name         | Type                            | Description                                                                                                        | Default    |
| ------------ | ------------------------------- | ------------------------------------------------------------------------------------------------------------------ | ---------- |
| `controller` | `Callable[[int], CommandRange]` | A function that takes the step index and returns a tensor of command values with the shape (num_envs, num_ranges). | *required* |

Example::

```
N_ENVS = 1
MIN_HEIGHT = 0.1
MAX_HEIGHT = 0.2

# Create environment
class MyEnv(GenesisEnv):
    def config(self):
        self.height_command = CommandManager(self, range=(MIN_HEIGHT, MAX_HEIGHT))
    # ...

# Setup gamepad
gamepad = Gamepad(GAMEPAD_PRODUCT)
cmd_buffer = torch.zeros((N_ENVS, 1), device=gs.device)
def gamepad_controller(_step):
    a_pressed = "a" in gamepad.state.buttons
    cmd_buffer[:, 0] = MAX_HEIGHT if a_pressed else MIN_HEIGHT
    return cmd_buffer

# Create environment & connect gamepad
env = MyEnv(num_envs=N_ENVS)
env.build()
env.command_manager.use_external_controller(gamepad_controller)
```

Source code in `genesis_forge/managers/command/command_manager.py`

```
def use_external_controller(self, controller: Callable[[int], CommandRange]):
    """
    Bypass the internal command controller, and generate the command values with an external control function.
    This can be used to connect a gamepad, joystick, or other external controller to the command manager.

    Args:
        controller: A function that takes the step index and returns a tensor of command values with the shape (num_envs, num_ranges).

    Example::

        N_ENVS = 1
        MIN_HEIGHT = 0.1
        MAX_HEIGHT = 0.2

        # Create environment
        class MyEnv(GenesisEnv):
            def config(self):
                self.height_command = CommandManager(self, range=(MIN_HEIGHT, MAX_HEIGHT))
            # ...

        # Setup gamepad
        gamepad = Gamepad(GAMEPAD_PRODUCT)
        cmd_buffer = torch.zeros((N_ENVS, 1), device=gs.device)
        def gamepad_controller(_step):
            a_pressed = "a" in gamepad.state.buttons
            cmd_buffer[:, 0] = MAX_HEIGHT if a_pressed else MIN_HEIGHT
            return cmd_buffer

        # Create environment & connect gamepad
        env = MyEnv(num_envs=N_ENVS)
        env.build()
        env.command_manager.use_external_controller(gamepad_controller)
    """
    self._external_controller = controller
```

## `use_gamepad(gamepad, range_axis, invert_axis=False)`

A wrapper around use_external_controller that converts a gamepad joystick axis to a command value.

Parameters:

| Name          | Type      | Description                 | Default                                                                                        |
| ------------- | --------- | --------------------------- | ---------------------------------------------------------------------------------------------- |
| `gamepad`     | `Gamepad` | The gamepad wrapper to use. | *required*                                                                                     |
| `range_axis`  | \`int     | dict[str, int]\`            | The axis or dict of axes to use for the command value. This should match the range init param. |
| `invert_axis` | \`bool    | dict[str, bool]\`           | Whether to invert each mapped axis before converting it into the command range.                |

Example::

```
# Create environment
class MyEnv(GenesisEnv):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.height_command = CommandManager(self, range=(0.1, 0.2))
    # ...

# Connect gamepad
from genesis_forge.gamepads import Gamepad
gamepad = Gamepad()

# Create environment & connect gamepad
env = MyEnv(num_envs=1)
env.build()

# Connect joystick axis 3 to the height command
env.height_command.use_gamepad(gamepad, range_axis=3)
```

Example with multiple ranges::

```
# Create environment
class MyEnv(GenesisEnv):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.height_command = CommandManager(
            self,
            range={
                "cmd1": (-2.0, 2.0),
                "cmd2": (1.0, 5.0),
            }
        )
    # ...

# Connect gamepad
from genesis_forge.gamepads import Gamepad
gamepad = Gamepad()

# Create environment & connect gamepad
env = MyEnv(num_envs=1)
env.build()

# Connect joystick axis 2 and 3 to to the indvidual ranges
env.height_command.use_gamepad(
    gamepad,
    range_axis={
        "cmd1": 2,
        "cmd2": 3,
    })
```

Source code in `genesis_forge/managers/command/command_manager.py`

```
def use_gamepad(
    self,
    gamepad: Gamepad,
    range_axis: int | dict[str, int],
    invert_axis: bool | dict[str, bool] = False,
):
    """
    A wrapper around use_external_controller that converts a gamepad joystick axis to a command value.

    Args:
        gamepad: The gamepad wrapper to use.
        range_axis: The axis or dict of axes to use for the command value. This should match the range init param.
        invert_axis: Whether to invert each mapped axis before converting it into the command range.

    Example::

        # Create environment
        class MyEnv(GenesisEnv):
            def __init__(self, *args, **kwargs):
                super().__init__(*args, **kwargs)
                self.height_command = CommandManager(self, range=(0.1, 0.2))
            # ...

        # Connect gamepad
        from genesis_forge.gamepads import Gamepad
        gamepad = Gamepad()

        # Create environment & connect gamepad
        env = MyEnv(num_envs=1)
        env.build()

        # Connect joystick axis 3 to the height command
        env.height_command.use_gamepad(gamepad, range_axis=3)

    Example with multiple ranges::

        # Create environment
        class MyEnv(GenesisEnv):
            def __init__(self, *args, **kwargs):
                super().__init__(*args, **kwargs)
                self.height_command = CommandManager(
                    self,
                    range={
                        "cmd1": (-2.0, 2.0),
                        "cmd2": (1.0, 5.0),
                    }
                )
            # ...

        # Connect gamepad
        from genesis_forge.gamepads import Gamepad
        gamepad = Gamepad()

        # Create environment & connect gamepad
        env = MyEnv(num_envs=1)
        env.build()

        # Connect joystick axis 2 and 3 to to the indvidual ranges
        env.height_command.use_gamepad(
            gamepad,
            range_axis={
                "cmd1": 2,
                "cmd2": 3,
            })
    """
    self._external_controller = self._gamepad_axis_command

    # Map axis to range keys
    axis_map = []
    axis_invert_map = []
    if isinstance(range_axis, int):
        axis_map.append(range_axis)
        axis_invert_map.append(invert_axis)
    elif isinstance(range_axis, dict):
        for key in self._range.keys():
            axis_map.append(range_axis[key])
            axis_invert_map.append(
                invert_axis[key] if isinstance(invert_axis, dict) else invert_axis
            )

    self._gamepad_cfg = {
        "gamepad": gamepad,
        "axis_map": axis_map,
        "axis_invert_map": axis_invert_map,
    }
    self._gamepad_axis_command_buffer = torch.zeros_like(
        self._command, device=gs.device
    )
```

# VelocityCommandManager

Bases: `CommandManager`

Generates a velocity command from uniform distribution. The command comprises of a linear velocity in x and y direction and an angular velocity around the z-axis.

IMPORTANT: The velocity commands are interpreted as robot-relative coordinates:

- X-axis: Forward/backward relative to robot's current orientation
- Y-axis: Left/right relative to robot's current orientation
- Z-axis: Yaw rotation around robot's vertical axis

Debug Visualization

If you set `debug_visualizer` to True, arrows will be rendered above your robot showing the commanded velocity vs the actual velocity.

Arrow meanings:

- GREEN: Commanded velocity (robot-relative, transformed to world coordinates for visualization) When joystick is "forward", this arrow points in the robot's forward direction
- BLUE: Actual robot velocity in world coordinates

Parameters:

| Name                   | Type                            | Description                                                                                 | Default                     |
| ---------------------- | ------------------------------- | ------------------------------------------------------------------------------------------- | --------------------------- |
| `env`                  | `GenesisEnv`                    | The environment to control                                                                  | *required*                  |
| `range`                | `VelocityCommandRange`          | The ranges of linear & angular velocities                                                   | *required*                  |
| `standing_probability` | `float`                         | The probability of all velocities being zero for an environment (0.0 = never, 1.0 = always) | `0.0`                       |
| `resample_time_sec`    | `float`                         | The time interval between changing the command                                              | `5.0`                       |
| `debug_visualizer`     | `bool`                          | Enable the debug arrow visualization                                                        | `False`                     |
| `debug_visualizer_cfg` | `VelocityDebugVisualizerConfig` | The configuration for the debug visualizer                                                  | `DEFAULT_VISUALIZER_CONFIG` |

Example::

```
class MyEnv(GenesisEnv):
    def config(self):
        # Create a velocity command manager
        self.command_manager = VelocityCommandManager(
            self,
            visualize=True,
            range = {
                "lin_vel_x_range": (-1.0, 1.0),
                "lin_vel_y_range": (-1.0, 1.0),
                "ang_vel_z_range": (-0.5, 0.5),
            }
        )

        RewardManager(
            self,
            logging_enabled=True,
            cfg={
                "tracking_lin_vel": {
                    "weight": 1.0,
                    "fn": rewards.command_tracking_lin_vel,
                    "params": {
                        "vel_cmd_manager": self.velocity_command,
                    },
                },
                "tracking_ang_vel": {
                    "weight": 1.0,
                    "fn": rewards.command_tracking_ang_vel,
                    "params": {
                        "vel_cmd_manager": self.velocity_command,
                    },
                },
                # ... other rewards ...
            },
        )

        # Observations
        ObservationManager(
            self,
            cfg={
                "velocity_cmd": {"fn": self.velocity_command.observation},
                # ... other observations ...
            },
        )
```

Source code in `genesis_forge/managers/command/velocity_command.py`

```
def __init__(
    self,
    env: GenesisEnv,
    range: VelocityCommandRange,
    resample_time_sec: float = 5.0,
    standing_probability: float = 0.0,
    debug_visualizer: bool = False,
    debug_visualizer_cfg: VelocityDebugVisualizerConfig = DEFAULT_VISUALIZER_CONFIG,
):
    super().__init__(env, range=range, resample_time_sec=resample_time_sec)
    self._arrow_nodes: list = []
    self.standing_probability = standing_probability
    self.debug_visualizer = debug_visualizer
    self.visualizer_cfg = {**DEFAULT_VISUALIZER_CONFIG, **debug_visualizer_cfg}
    self.debug_envs_idx = None

    self._is_standing_env = torch.zeros(
        env.num_envs, dtype=torch.bool, device=gs.device
    )
```

## `command`

The desired command value. Shape is (num_envs, num_ranges).

## `debug_envs_idx = None`

## `debug_visualizer = debug_visualizer`

## `enabled = True`

## `env = env`

## `range`

The velocity range dict.

## `resample_time_sec`

The time interval (in seconds) between changing the command for each environment.

## `standing_probability = standing_probability`

## `type = type`

## `visualizer_cfg = {None: DEFAULT_VISUALIZER_CONFIG, None: debug_visualizer_cfg}`

## `build()`

Build the velocity command manager

Source code in `genesis_forge/managers/command/velocity_command.py`

```
def build(self):
    """Build the velocity command manager"""
    super().build()
    self.build_debug()
```

## `build_debug()`

Build the debug components of the velocity command manager

Source code in `genesis_forge/managers/command/velocity_command.py`

```
def build_debug(self):
    """Build the debug components of the velocity command manager"""
    if not self.debug_visualizer or self.visualizer_cfg is None:
        return

    # Pre-allocate buffers
    self._arrow_pos_buffer = torch.zeros(self.env.num_envs, 3, device=gs.device)
    self._actual_vec_buffer = torch.zeros(self.env.num_envs, 3, device=gs.device)
    self._vec_3d_buffer = torch.zeros(self.env.num_envs, 3, device=gs.device)
    self._scene_env_offset = torch.from_numpy(self.env.scene.envs_offset).to(
        gs.device
    )

    # If debug envs_idx is not set, attempt to use the vis_options rendered_envs_idx
    self.debug_envs_idx = self.visualizer_cfg.get("envs_idx", None)
    if self.debug_envs_idx is None and self.env.scene.vis_options is not None:
        self.debug_envs_idx = self.env.scene.vis_options.rendered_envs_idx
    if self.debug_envs_idx is None:
        self.debug_envs_idx = list[int](range(self.env.num_envs))

    # Calculate the number of steps per debug render
    fps = self.visualizer_cfg.get("fps", 30)
    self._steps_per_debug_render = math.ceil(1.0 / fps / self.env.dt)

    # Arrow scale factor
    # Scales the arrow size based on the maximum target velocity range
    self._arrow_scale_factor = self.visualizer_cfg["arrow_max_length"] / max(
        *self._range["lin_vel_x"],
        *self._range["lin_vel_y"],
        *self._range["ang_vel_z"],
    )
```

## `get_command(range_key)`

If the range is a dict, get the command values for the given key.

Source code in `genesis_forge/managers/command/command_manager.py`

```
def get_command(self, range_key: str) -> torch.Tensor:
    """
    If the range is a dict, get the command values for the given key.
    """
    if not isinstance(self._range, dict):
        raise ValueError("The range is not a dict")
    return self._command[:, self._range_idx[range_key]]
```

## `get_command_idx(key)`

If the range is a dict, get the command index for the given key.

Source code in `genesis_forge/managers/command/command_manager.py`

```
def get_command_idx(self, key: str) -> int:
    """
    If the range is a dict, get the command index for the given key.
    """
    if not isinstance(self._range, dict):
        raise ValueError("The range is not a dict")
    return self._range_idx[key]
```

## `increment_range(range_key, increment, limit=None)`

Increment a command range target values by the given amount, with an optional limit.

Both increment and limit can be passed as a single float or a tuple of two floats. When they are tuples, they represent the increment and limit for both min and max. When they are a single float, they will be applied to both min and max as (-value, +value).

Example::

# Increment the range by 0.5 (+/-) for both min and max

command_manager.increment_range("height", 0.5)

```
# Increment the range min by -0.25 and max by 1.0
command_manager.increment_range("height", (-0.25, 1.0))

# Increment the range by (-0.25, 1.0) and keep min above -0.5 and max below 2.0
command_manager.increment_range("height", (-0.25, 1.0), (-0.5, 2.0))
```

Parameters:

| Name        | Type    | Description                                | Default                                                                                                                                                                                                                     |
| ----------- | ------- | ------------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `range_key` | `str`   | The key of the command range to increment. | *required*                                                                                                                                                                                                                  |
| `increment` | \`float | tuple[float, float]\`                      | The amount to increment the min & max range values by (+/-). If this is a tuple, it will be: (min_increment, max_increment). If this is a single float, it will be applied to both min and max as (-increment, +increment). |
| `limit`     | \`float | tuple[float, float]\`                      | Do not set the values beyond this limit range. This is a tuple of (min_limit, max_limit).                                                                                                                                   |

Source code in `genesis_forge/managers/command/command_manager.py`

```
def increment_range(
    self,
    range_key: str,
    increment: float | tuple[float, float],
    limit: float | tuple[float, float] = None,
):
    """
    Increment a command range target values by the given amount, with an optional limit.

    Both increment and limit can be passed as a single float or a tuple of two floats.
    When they are tuples, they represent the increment and limit for both min and max.
    When they are a single float, they will be applied to both min and max as (-value, +value).

    Example::
        # Increment the range by 0.5 (+/-) for both min and max
        command_manager.increment_range("height", 0.5)

        # Increment the range min by -0.25 and max by 1.0
        command_manager.increment_range("height", (-0.25, 1.0))

        # Increment the range by (-0.25, 1.0) and keep min above -0.5 and max below 2.0
        command_manager.increment_range("height", (-0.25, 1.0), (-0.5, 2.0))

    Args:
        range_key: The key of the command range to increment.
        increment: The amount to increment the min & max range values by (+/-).
                   If this is a tuple, it will be: `(min_increment, max_increment)`.
                   If this is a single float, it will be applied to both min and max as (-increment, +increment).
        limit: Do not set the values beyond this limit range.
               This is a tuple of `(min_limit, max_limit)`.
    """
    # Get the range to increment, and ensure it is a list, not a tuple (tuples are immutable)
    if not isinstance(self.range, dict):
        raise ValueError("Cannot increment a non-dict range item")
    range_item = self._range.get(range_key, None)
    if range_item is None:
        raise ValueError(f"Range item {range_key} not found")
    range_item = list[float](range_item)  # Ensure it's a list, not a tuple

    # Expand single increment/limit values to tuples
    if not isinstance(increment, (list, tuple)):
        increment = (-increment, increment)
    if limit is not None and not isinstance(limit, (list, tuple)):
        limit = (-limit, limit)

    # Increment the range values
    for i, value in enumerate(range_item):
        value += increment[i]
        if limit is not None:
            if increment[i] > 0:
                value = min(value, limit[i])
            else:
                value = max(value, limit[i])
        range_item[i] = value
    self._range[range_key] = range_item
```

## `observation(env)`

Function that returns the current command for each environment.

Source code in `genesis_forge/managers/command/command_manager.py`

```
def observation(self, env: GenesisEnv) -> torch.Tensor:
    """Function that returns the current command for each environment."""
    return self.command
```

## `resample_command(env_ids)`

Overwrites commands for environments that should be standing still.

Source code in `genesis_forge/managers/command/velocity_command.py`

```
def resample_command(self, env_ids: list[int]):
    """
    Overwrites commands for environments that should be standing still.
    """
    super().resample_command(env_ids)
    if not self.enabled:
        return

    # Set standing environments
    rand_buffer = torch.empty(len(env_ids), device=gs.device).uniform_(0.0, 1.0)
    self._is_standing_env[env_ids] = rand_buffer <= self.standing_probability
    standing_envs_idx = self._is_standing_env.nonzero(as_tuple=False).flatten()
    self._command[standing_envs_idx, :] = 0.0
```

## `reset(env_ids=None)`

One or more environments have been reset

Source code in `genesis_forge/managers/command/command_manager.py`

```
def reset(self, env_ids: list[int] | None = None):
    """One or more environments have been reset"""
    if not self.enabled:
        return
    if env_ids is None:
        env_ids = torch.arange(self.env.num_envs, device=gs.device)
    self.resample_command(env_ids)
```

## `set_command(range_key, value, envs_idx=None)`

Update a command value for selected environments.

Source code in `genesis_forge/managers/command/command_manager.py`

```
def set_command(
    self,
    range_key: str,
    value: torch.Tensor,
    envs_idx: list[int] | None = None,
):
    """
    Update a command value for selected environments.
    """
    if not isinstance(self._range, dict):
        raise ValueError("The range is not a dict")
    if envs_idx is None:
        self._command[:, self._range_idx[range_key]] = value
    else:
        self._command[envs_idx, self._range_idx[range_key]] = value
```

## `step()`

Render the command arrows

Source code in `genesis_forge/managers/command/velocity_command.py`

```
def step(self):
    """Render the command arrows"""
    if not self.enabled:
        return
    super().step()
    self._render_arrows()
```

## `use_external_controller(controller)`

Bypass the internal command controller, and generate the command values with an external control function. This can be used to connect a gamepad, joystick, or other external controller to the command manager.

Parameters:

| Name         | Type                            | Description                                                                                                        | Default    |
| ------------ | ------------------------------- | ------------------------------------------------------------------------------------------------------------------ | ---------- |
| `controller` | `Callable[[int], CommandRange]` | A function that takes the step index and returns a tensor of command values with the shape (num_envs, num_ranges). | *required* |

Example::

```
N_ENVS = 1
MIN_HEIGHT = 0.1
MAX_HEIGHT = 0.2

# Create environment
class MyEnv(GenesisEnv):
    def config(self):
        self.height_command = CommandManager(self, range=(MIN_HEIGHT, MAX_HEIGHT))
    # ...

# Setup gamepad
gamepad = Gamepad(GAMEPAD_PRODUCT)
cmd_buffer = torch.zeros((N_ENVS, 1), device=gs.device)
def gamepad_controller(_step):
    a_pressed = "a" in gamepad.state.buttons
    cmd_buffer[:, 0] = MAX_HEIGHT if a_pressed else MIN_HEIGHT
    return cmd_buffer

# Create environment & connect gamepad
env = MyEnv(num_envs=N_ENVS)
env.build()
env.command_manager.use_external_controller(gamepad_controller)
```

Source code in `genesis_forge/managers/command/command_manager.py`

```
def use_external_controller(self, controller: Callable[[int], CommandRange]):
    """
    Bypass the internal command controller, and generate the command values with an external control function.
    This can be used to connect a gamepad, joystick, or other external controller to the command manager.

    Args:
        controller: A function that takes the step index and returns a tensor of command values with the shape (num_envs, num_ranges).

    Example::

        N_ENVS = 1
        MIN_HEIGHT = 0.1
        MAX_HEIGHT = 0.2

        # Create environment
        class MyEnv(GenesisEnv):
            def config(self):
                self.height_command = CommandManager(self, range=(MIN_HEIGHT, MAX_HEIGHT))
            # ...

        # Setup gamepad
        gamepad = Gamepad(GAMEPAD_PRODUCT)
        cmd_buffer = torch.zeros((N_ENVS, 1), device=gs.device)
        def gamepad_controller(_step):
            a_pressed = "a" in gamepad.state.buttons
            cmd_buffer[:, 0] = MAX_HEIGHT if a_pressed else MIN_HEIGHT
            return cmd_buffer

        # Create environment & connect gamepad
        env = MyEnv(num_envs=N_ENVS)
        env.build()
        env.command_manager.use_external_controller(gamepad_controller)
    """
    self._external_controller = controller
```

## `use_gamepad(gamepad, lin_vel_y_axis=0, lin_vel_x_axis=1, ang_vel_z_axis=2)`

Use a connected gamepad to control the command.

Parameters:

| Name             | Type      | Description                                                             | Default    |
| ---------------- | --------- | ----------------------------------------------------------------------- | ---------- |
| `gamepad`        | `Gamepad` | The gamepad to use.                                                     | *required* |
| `lin_vel_x_axis` | `int`     | Map this gamepad axis index to the linear velocity in the x-direction.  | `1`        |
| `lin_vel_y_axis` | `int`     | Map this gamepad axis index to the linear velocity in the y-direction.  | `0`        |
| `ang_vel_z_axis` | `int`     | Map this gamepad axis index to the angular velocity in the z-direction. | `2`        |

Source code in `genesis_forge/managers/command/velocity_command.py`

```
def use_gamepad(
    self,
    gamepad: Gamepad,
    lin_vel_y_axis: int = 0,
    lin_vel_x_axis: int = 1,
    ang_vel_z_axis: int = 2,
):
    """
    Use a connected gamepad to control the command.

    Args:
        gamepad: The gamepad to use.
        lin_vel_x_axis: Map this gamepad axis index to the linear velocity in the x-direction.
        lin_vel_y_axis: Map this gamepad axis index to the linear velocity in the y-direction.
        ang_vel_z_axis: Map this gamepad axis index to the angular velocity in the z-direction.
    """
    super().use_gamepad(
        gamepad,
        range_axis={
            "lin_vel_x": lin_vel_x_axis,
            "lin_vel_y": lin_vel_y_axis,
            "ang_vel_z": ang_vel_z_axis,
        },
        invert_axis={
            "lin_vel_x": True,
            "lin_vel_y": True,
            "ang_vel_z": True,
        },
    )
```
# MDP Functions

# MDP Function Overview

MDP stands for [Markov Decision Process](https://en.wikipedia.org/wiki/Markov_decision_process), and is the process at the heart of reinforcement learning.

This is a collection of simple functions used to generate rewards, terminations, observations, or handle robot resets.

# Observations

## `contact_force(env, contact_manager)`

Returns the vector norm contact force at each contact point.

Parameters:

| Name              | Type             | Description                              | Default    |
| ----------------- | ---------------- | ---------------------------------------- | ---------- |
| `env`             | `GenesisEnv`     | The Genesis Forge environment            | *required* |
| `contact_manager` | `ContactManager` | The contact manager to check for contact | *required* |

Returns:

| Type     | Description                                   |
| -------- | --------------------------------------------- |
| `Tensor` | torch.Tensor: Shape (num_envs, num_contacts). |

Source code in `genesis_forge/mdp/observations.py`

```
def contact_force(env: GenesisEnv, contact_manager: ContactManager) -> torch.Tensor:
    """
    Returns the vector norm contact force at each contact point.

    Args:
        env: The Genesis Forge environment
        contact_manager: The contact manager to check for contact

    Returns:
        torch.Tensor: Shape `(num_envs, num_contacts)`.
    """
    return torch.norm(contact_manager.contacts[:, :, :], dim=-1)
```

## `current_actions(env, action_manager=None)`

The most current step actions.

Source code in `genesis_forge/mdp/observations.py`

```
def current_actions(
    env: GenesisEnv,
    action_manager: PositionActionManager = None,
) -> torch.Tensor:
    """
    The most current step actions.
    """
    if action_manager is not None:
        return action_manager.get_actions()
    return env.actions
```

## `entity_angular_velocity(env, entity_manager=None, entity_attr='robot')`

The angular velocity of the entity's base link, in the entity's local frame.

Parameters:

| Name             | Type            | Description                                                                                                                                           | Default    |
| ---------------- | --------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- | ---------- |
| `env`            | `GenesisEnv`    | The Genesis environment containing the entity                                                                                                         | *required* |
| `entity_manager` | `EntityManager` | The entity manager for the robot/entity the observation is being computed for. This is slightly more performant than using the entity_attr parameter. | `None`     |
| `entity_attr`    | `str`           | The attribute name of the entity in the environment. This isn't necessary if entity_manager is provided.                                              | `'robot'`  |

Returns:

| Type     | Description                                                                                |
| -------- | ------------------------------------------------------------------------------------------ |
| `Tensor` | torch.Tensor: The angular velocity of the entity's base link, in the entity's local frame. |

Source code in `genesis_forge/mdp/observations.py`

```
def entity_angular_velocity(
    env: GenesisEnv, entity_manager: EntityManager = None, entity_attr: str = "robot"
) -> torch.Tensor:
    """
    The angular velocity of the entity's base link, in the entity's local frame.

    Args:
        env: The Genesis environment containing the entity
        entity_manager: The entity manager for the robot/entity the observation is being computed for.
                        This is slightly more performant than using the `entity_attr` parameter.
        entity_attr: The attribute name of the entity in the environment. This isn't necessary if `entity_manager` is provided.

    Returns:
        torch.Tensor: The angular velocity of the entity's base link, in the entity's local frame.
    """
    if entity_manager is not None:
        return entity_manager.get_angular_velocity()
    entity = getattr(env, entity_attr)
    return entity_ang_vel(entity)
```

## `entity_dofs_force(env, actuator_manager=None, entity_attr='robot', dofs_idx=None, clip_to_max_force=False, action_manager=None)`

The DOF's force being experienced.

Parameters:

| Name                | Type                    | Description                                                                                                | Default    |
| ------------------- | ----------------------- | ---------------------------------------------------------------------------------------------------------- | ---------- |
| `env`               | `GenesisEnv`            | The Genesis environment containing the entity                                                              | *required* |
| `actuator_manager`  | `ActuatorManager`       | The actuator manager for the robot/entity. This bypasses the need for dofs_idx and entity_attr parameters. | `None`     |
| `entity_attr`       | `str`                   | The attribute name of the entity in the environment. This isn't necessary if action_manager is provided.   | `'robot'`  |
| `dofs_idx`          | `list[int]`             | The indices of the DOFs to get the force of. This isn't necessary if action_manager is provided.           | `None`     |
| `clip_to_max_force` | `bool`                  | Clip the force to the maximum force defined in the action_manager.                                         | `False`    |
| `action_manager`    | `PositionActionManager` | (deprecated) The action manager for the robot/entity.                                                      | `None`     |

Returns:

| Type     | Description                                   |
| -------- | --------------------------------------------- |
| `Tensor` | torch.Tensor: The force of the entity's DOFs. |

Source code in `genesis_forge/mdp/observations.py`

```
def entity_dofs_force(
    env: GenesisEnv,
    actuator_manager: ActuatorManager = None,
    entity_attr: str = "robot",
    dofs_idx: list[int] = None,
    clip_to_max_force: bool = False,
    action_manager: PositionActionManager = None,
) -> torch.Tensor:
    """
    The DOF's force being experienced.

    Args:
        env: The Genesis environment containing the entity
        actuator_manager: The actuator manager for the robot/entity.
                          This bypasses the need for dofs_idx and entity_attr parameters.
        entity_attr: The attribute name of the entity in the environment. This isn't necessary if `action_manager` is provided.
        dofs_idx: The indices of the DOFs to get the force of. This isn't necessary if `action_manager` is provided.
        clip_to_max_force: Clip the force to the maximum force defined in the `action_manager`.
        action_manager: (deprecated) The action manager for the robot/entity.

    Returns:
        torch.Tensor: The force of the entity's DOFs.
    """
    if actuator_manager is not None:
        return actuator_manager.get_dofs_force(clip_to_max_force=clip_to_max_force)
    elif action_manager is not None:
        return action_manager.get_dofs_force(clip_to_max_force=clip_to_max_force)
    entity: RigidEntity = getattr(env, entity_attr)
    return entity.get_dofs_force(dofs_idx)
```

## `entity_dofs_position(env, actuator_manager=None, entity_attr='robot', dofs_idx=None, action_manager=None)`

The position of the entity's DOFs.

Parameters:

| Name               | Type                    | Description                                                                                                           | Default    |
| ------------------ | ----------------------- | --------------------------------------------------------------------------------------------------------------------- | ---------- |
| `env`              | `GenesisEnv`            | The Genesis environment containing the entity                                                                         | *required* |
| `actuator_manager` | `ActuatorManager`       | The actuator manager for the robot/entity. This bypasses the need for dofs_idx and entity_attr parameters.            | `None`     |
| `entity_attr`      | `str`                   | The attribute name of the entity in the environment. This isn't necessary if action_manager is provided.              | `'robot'`  |
| `dofs_idx`         | `list[int]`             | The indices of the DOFs to get the position of. This isn't necessary if action_manager is provided.                   | `None`     |
| `action_manager`   | `PositionActionManager` | (deprecated) The action manager for the robot/entity. This bypasses the need for dofs_idx and entity_attr parameters. | `None`     |

Returns:

| Type     | Description                                      |
| -------- | ------------------------------------------------ |
| `Tensor` | torch.Tensor: The position of the entity's DOFs. |

Source code in `genesis_forge/mdp/observations.py`

```
def entity_dofs_position(
    env: GenesisEnv,
    actuator_manager: ActuatorManager = None,
    entity_attr: str = "robot",
    dofs_idx: list[int] = None,
    action_manager: PositionActionManager = None,
) -> torch.Tensor:
    """
    The position of the entity's DOFs.

    Args:
        env: The Genesis environment containing the entity
        actuator_manager: The actuator manager for the robot/entity.
                          This bypasses the need for dofs_idx and entity_attr parameters.
        entity_attr: The attribute name of the entity in the environment. This isn't necessary if `action_manager` is provided.
        dofs_idx: The indices of the DOFs to get the position of. This isn't necessary if `action_manager` is provided.
        action_manager: (deprecated) The action manager for the robot/entity.
                        This bypasses the need for dofs_idx and entity_attr parameters.

    Returns:
        torch.Tensor: The position of the entity's DOFs.
    """
    if actuator_manager is not None:
        return actuator_manager.get_dofs_position()
    if action_manager is not None:
        return action_manager.get_dofs_position()
    entity: RigidEntity = getattr(env, entity_attr)
    return entity.get_dofs_position(dofs_idx)
```

## `entity_dofs_velocity(env, action_manager=None, entity_attr='robot', dofs_idx=None)`

The velocity of the entity's DOFs.

Parameters:

| Name             | Type                    | Description                                                                                                     | Default    |
| ---------------- | ----------------------- | --------------------------------------------------------------------------------------------------------------- | ---------- |
| `env`            | `GenesisEnv`            | The Genesis environment containing the entity                                                                   | *required* |
| `action_manager` | `PositionActionManager` | The action manager for the robot/entity. This is slightly more performant than using the entity_attr parameter. | `None`     |
| `entity_attr`    | `str`                   | The attribute name of the entity in the environment. This isn't necessary if action_manager is provided.        | `'robot'`  |
| `dofs_idx`       | `list[int]`             | The indices of the DOFs to get the velocity of. This isn't necessary if action_manager is provided.             | `None`     |

Returns:

| Type     | Description                                      |
| -------- | ------------------------------------------------ |
| `Tensor` | torch.Tensor: The velocity of the entity's DOFs. |

Source code in `genesis_forge/mdp/observations.py`

```
def entity_dofs_velocity(
    env: GenesisEnv,
    action_manager: PositionActionManager = None,
    entity_attr: str = "robot",
    dofs_idx: list[int] = None,
) -> torch.Tensor:
    """
    The velocity of the entity's DOFs.

    Args:
        env: The Genesis environment containing the entity
        action_manager: The action manager for the robot/entity.
                        This is slightly more performant than using the `entity_attr` parameter.
        entity_attr: The attribute name of the entity in the environment. This isn't necessary if `action_manager` is provided.
        dofs_idx: The indices of the DOFs to get the velocity of. This isn't necessary if `action_manager` is provided.

    Returns:
        torch.Tensor: The velocity of the entity's DOFs.
    """
    if action_manager is not None:
        return action_manager.get_dofs_velocity()
    entity: RigidEntity = getattr(env, entity_attr)
    return entity.get_dofs_velocity(dofs_idx)
```

## `entity_linear_velocity(env, entity_manager=None, entity_attr='robot')`

The linear velocity of the entity's base link, in the entity's local frame.

Parameters:

| Name             | Type            | Description                                                                                                                                           | Default    |
| ---------------- | --------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- | ---------- |
| `env`            | `GenesisEnv`    | The Genesis environment containing the entity                                                                                                         | *required* |
| `entity_manager` | `EntityManager` | The entity manager for the robot/entity the observation is being computed for. This is slightly more performant than using the entity_attr parameter. | `None`     |
| `entity_attr`    | `str`           | The attribute name of the entity in the environment. This isn't necessary if entity_manager is provided.                                              | `'robot'`  |

Returns:

| Type     | Description                                                                               |
| -------- | ----------------------------------------------------------------------------------------- |
| `Tensor` | torch.Tensor: The linear velocity of the entity's base link, in the entity's local frame. |

Source code in `genesis_forge/mdp/observations.py`

```
def entity_linear_velocity(
    env: GenesisEnv, entity_manager: EntityManager = None, entity_attr: str = "robot"
) -> torch.Tensor:
    """
    The linear velocity of the entity's base link, in the entity's local frame.

    Args:
        env: The Genesis environment containing the entity
        entity_manager: The entity manager for the robot/entity the observation is being computed for.
                        This is slightly more performant than using the `entity_attr` parameter.
        entity_attr: The attribute name of the entity in the environment. This isn't necessary if `entity_manager` is provided.

    Returns:
        torch.Tensor: The linear velocity of the entity's base link, in the entity's local frame.
    """
    if entity_manager is not None:
        return entity_manager.get_linear_velocity()
    entity = getattr(env, entity_attr)
    return entity_lin_vel(entity)
```

## `has_contact(env, contact_manager, threshold=1.0)`

Return boolean (1/0) for each link in the contact manager that meets the contact threshold.

Parameters:

| Name              | Type             | Description                                                      | Default    |
| ----------------- | ---------------- | ---------------------------------------------------------------- | ---------- |
| `env`             | `GenesisEnv`     | The Genesis Forge environment                                    | *required* |
| `contact_manager` | `ContactManager` | The contact manager to check for contact                         | *required* |
| `threshold`       | `float`          | The minimum force necessary for contact detection (default: 1.0) | `1.0`      |

Returns:

| Type     | Description                                   |
| -------- | --------------------------------------------- |
| `Tensor` | 1 for each link meeting the contact threshold |

Source code in `genesis_forge/mdp/observations.py`

```
def has_contact(
    env: GenesisEnv, contact_manager: ContactManager, threshold: float=1.0
) -> torch.Tensor:
    """
    Return boolean (1/0) for each link in the contact manager that meets the contact threshold.

    Args:
        env: The Genesis Forge environment
        contact_manager: The contact manager to check for contact
        threshold: The minimum force necessary for contact detection (default: 1.0)

    Returns:
        1 for each link meeting the contact threshold
    """
    has_contact = contact_manager.contacts.norm(dim=-1) > threshold
    return has_contact.float()
```

## `read_imu(env, imu)`

Makes an IMU reading and returns the concatenated linear acceleration and angular velocity readings.

Example::

```
self.imu = gs.sensors.IMU(
    entity_idx=self.robot.idx,
    pos_offset=(0.24, 0.0, 0.0),
    euler_offset=(0.0, 0.0, 0.0),
)

...

ObservationManager(
    self,
    cfg={
        "imu_sensor": {
            "fn": self.imu_observation,
        },
    }
)
```

Returns:

| Type     | Description                                                           |
| -------- | --------------------------------------------------------------------- |
| `Tensor` | torch.Tensor: Shape (n_envs, 6) — [lin_acc_xyz, ang_vel_xyz] per env. |

Source code in `genesis_forge/mdp/observations.py`

```
def read_imu(env: GenesisEnv, imu: gs.sensors.IMU) -> torch.Tensor:
    """
    Makes an IMU reading and returns the concatenated linear acceleration and angular velocity readings.

    Example::

        self.imu = gs.sensors.IMU(
            entity_idx=self.robot.idx,
            pos_offset=(0.24, 0.0, 0.0),
            euler_offset=(0.0, 0.0, 0.0),
        )

        ...

        ObservationManager(
            self,
            cfg={
                "imu_sensor": {
                    "fn": self.imu_observation,
                },
            }
        )

    Returns:
        torch.Tensor: Shape `(n_envs, 6)` — `[lin_acc_xyz, ang_vel_xyz]` per env.
    """
    value = imu.read()
    return torch.cat([value.lin_acc, value.ang_vel], dim=-1)
```

# Resets

These are simple helper functions that can be used in the [EntityManager](https://genesis-forge.readthedocs.io/en/latest/api/managers/entity.html) `on_reset` configuration.

## `XYZRotation = dict[Literal['x', 'y', 'z'], float | tuple[float, float]]`

Define the rotation around the X/Y/Z axes. The value can either be a distinct value, or a tuple of (min, max) values to randomize within.

## `position(env, entity, position, quat=None, zero_velocity=True)`

Bases: `ResetMdpFnClass`

Reset the entity to a fixed position and (optional) rotation

Parameters:

| Name            | Type                                | Description                                                                                                                             | Default                              |
| --------------- | ----------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------ |
| `env`           | `GenesisEnv`                        | The environment                                                                                                                         | *required*                           |
| `entity`        | `RigidEntity`                       | The entity to set the position of.                                                                                                      | *required*                           |
| `position`      | `tuple[float, float, float]`        | The position to set the entity to.                                                                                                      | *required*                           |
| `quat`          | \`tuple[float, float, float, float] | None\`                                                                                                                                  | The quaternion to set the entity to. |
| `zero_velocity` | `bool`                              | Whether to zero the velocity of all the entity's dofs. Defaults to True. This is a safety measure after a sudden change in entity pose. | `True`                               |

Source code in `genesis_forge/mdp/reset.py`

```
def __init__(
    self,
    env: GenesisEnv,
    entity: RigidEntity,
    position: tuple[float, float, float],
    quat: tuple[float, float, float, float] | None = None,
    zero_velocity: bool = True,
):
    self.zero_velocity = zero_velocity
    self.reset_pos = torch.tensor(position, device=gs.device)
    self._pos_buffer = torch.zeros(
        (env.num_envs, 3), device=gs.device, dtype=gs.tc_float
    )

    self.reset_quat = None
    self._quat_buffer = None
    if quat is not None:
        self.reset_quat = torch.tensor(quat, device=gs.device)
        self._quat_buffer = torch.zeros(
            (env.num_envs, 4), device=gs.device, dtype=gs.tc_float
        )
```

### `env = env`

### `reset_pos = torch.tensor(position, device=(gs.device))`

### `reset_quat = None`

### `zero_velocity = zero_velocity`

### `build()`

Source code in `genesis_forge/managers/config/mdp_fn_class.py`

```
def build(self):
    pass
```

### `reset(envs_idx)`

Called when environments are reset. Override to clear per-env state.

Source code in `genesis_forge/managers/config/mdp_fn_class.py`

```
def reset(self, envs_idx):
    """Called when environments are reset. Override to clear per-env state."""
    pass
```

## `randomize_link_mass_shift(env, entity, link_name, mass_range)`

Bases: `ResetMdpFnClass`

Randomly add/subtract mass to one or more links of the entity. This picks a random value from `mass_range` and passes it to `set_mass_shift` for each environment.

See: https://genesis-world.readthedocs.io/en/latest/api_reference/entity/rigid_entity/rigid_entity.html#genesis.engine.entities.rigid_entity.rigid_entity.RigidEntity.set_mass_shift

Parameters:

| Name         | Type                  | Description                                                                            | Default    |
| ------------ | --------------------- | -------------------------------------------------------------------------------------- | ---------- |
| `env`        | `GenesisEnv`          | The environment                                                                        | *required* |
| `entity`     | `RigidEntity`         | The entity to set the rotation of.                                                     | *required* |
| `link_name`  | `str`                 | The name, or regex pattern, of the link(s) to set the mass for.                        | *required* |
| `mass_range` | `tuple[float, float]` | The range of the mass that will be added or subtracted from the link(s) on each reset. | *required* |

Source code in `genesis_forge/mdp/reset.py`

```
def __init__(
    self,
    env: GenesisEnv,
    entity: RigidEntity,
    link_name: str,
    mass_range: tuple[float, float],
):
    self.env = env
    self._entity = entity
    self._link_name = link_name
    self._links_idx_local = []
    self._mass_shift_buffer: torch.tensor | None = None
    self.build()
```

### `env = env`

### `build()`

Source code in `genesis_forge/mdp/reset.py`

```
def build(self):
    self._links_idx_local = []
    self._orig_mass = None
    if self._link_name is not None:
        links = links_by_name_pattern(self._entity, self._link_name)
        if len(links) > 0:
            self._links_idx_local = [link.idx_local for link in links]
            self._mass_shift_buffer = torch.zeros(
                (self.env.num_envs, len(self._links_idx_local)), device=gs.device
            )
        else:
            raise ValueError(
                f"No links found with name/pattern '{self._link_name}'"
            )
```

### `reset(envs_idx)`

Called when environments are reset. Override to clear per-env state.

Source code in `genesis_forge/managers/config/mdp_fn_class.py`

```
def reset(self, envs_idx):
    """Called when environments are reset. Override to clear per-env state."""
    pass
```

## `randomize_terrain_position(env, entity, envs_idx, terrain_manager, height_offset=0.0001, subterrain=None, rotation={'z': (0, 2 * math.pi)}, zero_velocity=True)`

Bases: `ResetMdpFnClass`

Place the entity in a random position on the terrain for each environment.

Parameters:

| Name              | Type             | Description                                                                                                                             | Default                                                                                                                      |
| ----------------- | ---------------- | --------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------- |
| `env`             | `GenesisEnv`     | The environment                                                                                                                         | *required*                                                                                                                   |
| `entity`          | `RigidEntity`    | The entity to set the position of.                                                                                                      | *required*                                                                                                                   |
| `envs_idx`        | `list[int]`      | The environment ids to set the position for.                                                                                            | *required*                                                                                                                   |
| `terrain_manager` | `TerrainManager` | The terrain manager to use to generate the random position.                                                                             | *required*                                                                                                                   |
| `height_offset`   | `float`          | The height offset to add to the random position.                                                                                        | `0.0001`                                                                                                                     |
| `subterrain`      | \`str            | Callable\[[], str\]                                                                                                                     | None\`                                                                                                                       |
| `rotation`        | \`XYZRotation    | None\`                                                                                                                                  | The X/Y/Z rotation to set the entity to. Defaults to a random rotation around the z-axis. Set to None to not set a rotation. |
| `zero_velocity`   | `bool`           | Whether to zero the velocity of all the entity's dofs. Defaults to True. This is a safety measure after a sudden change in entity pose. | `True`                                                                                                                       |

Source code in `genesis_forge/mdp/reset.py`

```
def __init__(
    self,
    env: GenesisEnv,
    entity: RigidEntity,
    envs_idx: list[int],
    terrain_manager: TerrainManager,
    height_offset: float = 0.1e-3,
    subterrain: str | Callable[[], str] | None = None,
    rotation: XYZRotation | None = {"z": (0, 2 * math.pi)},
    zero_velocity: bool = True,
):
    super().__init__(env, entity)
    self.env = env
    self.rotation = rotation
    self._rotation_buffer = None
    self._quat_buffer = None
```

### `env = env`

### `rotation = rotation`

### `build()`

Initialize the buffers

Source code in `genesis_forge/mdp/reset.py`

```
def build(self):
    """
    Initialize the buffers
    """
    self._rotation_buffer = torch.zeros(
        (self.env.num_envs, 3), device=gs.device, dtype=gs.tc_float
    )
    self._quat_buffer = torch.zeros(
        (self.env.num_envs, 4), device=gs.device, dtype=gs.tc_float
    )
```

### `define_quat(envs_idx, rotation)`

Set the rotation quaternion for the given environment ids.

Source code in `genesis_forge/mdp/reset.py`

```
def define_quat(self, envs_idx: list[int], rotation: XYZRotation):
    """
    Set the rotation quaternion for the given environment ids.
    """
    x = rotation["x"] if "x" in rotation else 0
    y = rotation["y"] if "y" in rotation else 0
    z = rotation["z"] if "z" in rotation else 0
    n_envs = len(envs_idx)

    if isinstance(x, tuple):
        self._rotation_buffer[envs_idx, 0] = torch.empty(
            n_envs, device=gs.device
        ).uniform_(*x)
    if isinstance(y, tuple):
        self._rotation_buffer[envs_idx, 1] = torch.empty(
            n_envs, device=gs.device
        ).uniform_(*y)
    if isinstance(z, tuple):
        self._rotation_buffer[envs_idx, 2] = torch.empty(
            n_envs, device=gs.device
        ).uniform_(*z)

    # Set angle as quat
    self._quat_buffer[envs_idx] = xyz_to_quat(self._rotation_buffer[envs_idx])
```

### `reset(envs_idx)`

Called when environments are reset. Override to clear per-env state.

Source code in `genesis_forge/managers/config/mdp_fn_class.py`

```
def reset(self, envs_idx):
    """Called when environments are reset. Override to clear per-env state."""
    pass
```

## `set_rotation(env, entity, envs_idx, x=0, y=0, z=0)`

Set the entity's rotation in either absolute or randomized euler angles. If the x/y/z value is a tuple (for example: `(0, 2 * math.pi)`), the rotation will be randomized within that radian range.

Parameters:

| Name       | Type          | Description                                  | Default                                      |
| ---------- | ------------- | -------------------------------------------- | -------------------------------------------- |
| `env`      | `GenesisEnv`  | The environment                              | *required*                                   |
| `entity`   | `RigidEntity` | The entity to set the rotation of.           | *required*                                   |
| `envs_idx` | `list[int]`   | The environment ids to set the rotation for. | *required*                                   |
| `x`        | \`float       | tuple[float, float]\`                        | The x angle or range to set the rotation to. |
| `y`        | \`float       | tuple[float, float]\`                        | The y angle or range to set the rotation to. |
| `z`        | \`float       | tuple[float, float]\`                        | The z angle or range to set the rotation to. |

Source code in `genesis_forge/mdp/reset.py`

```
def set_rotation(
    env: GenesisEnv,
    entity: RigidEntity,
    envs_idx: list[int],
    x: float | tuple[float, float] = 0,
    y: float | tuple[float, float] = 0,
    z: float | tuple[float, float] = 0,
):
    """
    Set the entity's rotation in either absolute or randomized euler angles.
    If the x/y/z value is a tuple (for example: `(0, 2 * math.pi)`), the rotation will be randomized within that radian range.

    Args:
        env: The environment
        entity: The entity to set the rotation of.
        envs_idx: The environment ids to set the rotation for.
        x: The x angle or range to set the rotation to.
        y: The y angle or range to set the rotation to.
        z: The z angle or range to set the rotation to.
    """

    angle_buffer = torch.zeros((len(envs_idx), 3), device=gs.device)
    if isinstance(x, tuple):
        angle_buffer[:, 0].uniform_(*x)
    if isinstance(y, tuple):
        angle_buffer[:, 1].uniform_(*y)
    if isinstance(z, tuple):
        angle_buffer[:, 2].uniform_(*z)

    # Set angle as quat
    quat = xyz_to_quat(angle_buffer)
    entity.set_quat(quat, envs_idx=envs_idx)
```

## `zero_all_dofs_velocity(env, entity, envs_idx)`

Zero the velocity of all dofs of the entity.

Source code in `genesis_forge/mdp/reset.py`

```
def zero_all_dofs_velocity(
    env: GenesisEnv,
    entity: RigidEntity,
    envs_idx: list[int],
):
    """
    Zero the velocity of all dofs of the entity.
    """
    entity.zero_all_dofs_velocity(envs_idx)
```

# Rewards

Reward functions for the Genesis Forge environment. Each of these should return a float tensor with the reward value for each environment, in the shape (num_envs,).

## `action_acceleration_l2(env, action_manager=None)`

Bases: `MdpFnClass`

Targets jittery oscillations (rather than smooth consistent movement), by penalize the second-order finite difference of actions (discrete acceleration) using the L2 squared kernel.

This encourages a smooth consistent movement, where a smooth ramp has zero acceleration even at high velocity.

0.5 → 0.6 → 0.7 → 0.8

- Velocities: 0.1, 0.1, 0.1
- Accelerations: 0.0, 0.0 (zero -- perfectly smooth)
- Penalty: zero

0.5 → 0.8 → 0.5 → 0.8

- Velocities: 0.3, -0.3, 0.3
- Accelerations: -0.6, 0.6 (large -- direction keeps reversing)
- Penalty: very large

The acceleration is computed as:

.. math::

```
\text{acc}_t = a_t - 2 \cdot a_{t-1} + a_{t-2}
```

and the penalty is :math:`\sum \text{acc}_t^2` across all action dimensions.

Parameters:

| Name             | Type                    | Description                                                                                         | Default    |
| ---------------- | ----------------------- | --------------------------------------------------------------------------------------------------- | ---------- |
| `env`            | `GenesisEnv`            | The Genesis environment containing the robot                                                        | *required* |
| `action_manager` | `PositionActionManager` | Optional action manager to source actions from. If not provided, actions are read from env.actions. | `None`     |

Source code in `genesis_forge/mdp/rewards.py`

```
def __init__(
    self,
    env: GenesisEnv,
    action_manager: PositionActionManager = None,
):
    super().__init__(env)
    self.env = env
    self._prev_action: torch.Tensor | None = None
    self._prev_prev_action: torch.Tensor | None = None
    self._action_log_count: torch.Tensor | None = None
```

### `env = env`

### `build()`

Called during the environment build phase and when MDP params are changed.

Source code in `genesis_forge/managers/config/mdp_fn_class.py`

```
def build(self):
    """Called during the environment build phase and when MDP params are changed."""
    pass
```

### `reset(envs_idx)`

Clear the action history for the specified environments.

Source code in `genesis_forge/mdp/rewards.py`

```
def reset(self, envs_idx):
    """
    Clear the action history for the specified environments.
    """
    if self._prev_action is None:
        return
    self._prev_action[envs_idx] = 0.0
    self._prev_prev_action[envs_idx] = 0.0
    self._action_log_count[envs_idx] = 0
```

## `body_acceleration_exp(env, entity_attr='robot', entity_manager=None, sensitivity=0.1)`

Bases: `MdpFnClass`

Penalize jerky body acceleration to encourage smooth locomotion.

Parameters:

| Name             | Type            | Description                                                                                                                                      | Default    |
| ---------------- | --------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | ---------- |
| `env`            | `GenesisEnv`    | The Genesis environment containing the robot                                                                                                     | *required* |
| `entity_manager` | `EntityManager` | The entity manager for the robot/entity the reward is being computed for. This is slightly more performant than using the entity_attr parameter. | `None`     |
| `entity_attr`    | `str`           | The attribute name of the entity in the environment. This isn't necessary if entity_manager is provided.                                         | `'robot'`  |
| `sensitivity`    | `float`         | The sensitivity of the exponential decay. A lower value means the reward is more sensitive to the error.                                         | `0.1`      |

Source code in `genesis_forge/mdp/rewards.py`

```
def __init__(
    self,
    env: GenesisEnv,
    entity_attr: str = "robot",
    entity_manager: EntityManager = None,
    sensitivity: float = 0.10,
):
    super().__init__(env)
```

### `env = env`

### `build()`

Called during the environment build phase and when MDP params are changed.

Source code in `genesis_forge/managers/config/mdp_fn_class.py`

```
def build(self):
    """Called during the environment build phase and when MDP params are changed."""
    pass
```

### `reset(envs_idx)`

Called when environments are reset. Override to clear per-env state.

Source code in `genesis_forge/managers/config/mdp_fn_class.py`

```
def reset(self, envs_idx):
    """Called when environments are reset. Override to clear per-env state."""
    pass
```

## `action_rate_l2(env)`

Penalize the rate of change of the actions using L2 squared kernel.

Parameters:

| Name  | Type         | Description                                  | Default    |
| ----- | ------------ | -------------------------------------------- | ---------- |
| `env` | `GenesisEnv` | The Genesis environment containing the robot | *required* |

Returns:

| Type     | Description                                  |
| -------- | -------------------------------------------- |
| `Tensor` | torch.Tensor: Penalty for changes in actions |

Source code in `genesis_forge/mdp/rewards.py`

```
def action_rate_l2(env: GenesisEnv) -> torch.Tensor:
    """
    Penalize the rate of change of the actions using L2 squared kernel.

    Args:
        env: The Genesis environment containing the robot

    Returns:
        torch.Tensor: Penalty for changes in actions
    """
    actions = env.actions
    last_actions = env.last_actions
    if last_actions is None:
        return torch.zeros_like(actions, device=gs.device)
    return torch.sum(torch.square(last_actions - actions), dim=1)
```

## `ang_vel_xy_l2(env, entity_attr='robot', entity_manager=None)`

Penalize xy-axis base angular velocity using L2 squared kernel.

Parameters:

| Name             | Type            | Description                                                                                                                                      | Default    |
| ---------------- | --------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | ---------- |
| `env`            | `GenesisEnv`    | The Genesis environment containing the entity                                                                                                    | *required* |
| `entity_manager` | `EntityManager` | The entity manager for the robot/entity the reward is being computed for. This is slightly more performant than using the entity_attr parameter. | `None`     |
| `entity_attr`    | `str`           | The attribute name of the entity in the environment. This isn't necessary if entity_manager is provided.                                         | `'robot'`  |

Returns:

| Type     | Description  |
| -------- | ------------ |
| `Tensor` | torch.Tensor |

Source code in `genesis_forge/mdp/rewards.py`

```
def ang_vel_xy_l2(
    env: GenesisEnv,
    entity_attr: str = "robot",
    entity_manager: EntityManager = None,
) -> torch.Tensor:
    """
    Penalize xy-axis base angular velocity using L2 squared kernel.

    Args:
        env: The Genesis environment containing the entity
        entity_manager: The entity manager for the robot/entity the reward is being computed for.
                        This is slightly more performant than using the `entity_attr` parameter.
        entity_attr: The attribute name of the entity in the environment. This isn't necessary if `entity_manager` is provided.

    Returns:
        torch.Tensor
    """
    angle_vel = None
    if entity_manager is not None:
        angle_vel = entity_manager.get_angular_velocity()
    else:
        robot = getattr(env, entity_attr)
        angle_vel = entity_ang_vel(robot)
    return torch.sum(torch.square(angle_vel[:, :2]), dim=1)
```

## `base_height(env, target_height=None, height_command=None, terrain_manager=None, entity_attr='robot', entity_manager=None)`

Penalize base height away from target, using the L2 squared kernel.

Parameters:

| Name              | Type                   | Description                                                                                                 | Default    |
| ----------------- | ---------------------- | ----------------------------------------------------------------------------------------------------------- | ---------- |
| `env`             | `GenesisEnv`           | The Genesis environment containing the robot                                                                | *required* |
| `target_height`   | `Union[float, Tensor]` | The target height to penalize the base height away from                                                     | `None`     |
| `height_command`  | `CommandManager`       | Get the target height from a height command manager. This expects the command to have a single range value. | `None`     |
| `terrain_manager` | `TerrainManager`       | The terrain manager will adjust the height based on the terrain height.                                     | `None`     |
| `entity_attr`     | `str`                  | The attribute name of the entity in the environment.                                                        | `'robot'`  |
| `entity_manager`  | `EntityManager`        | The entity manager for the entity.                                                                          | `None`     |

Returns:

| Type     | Description                                            |
| -------- | ------------------------------------------------------ |
| `Tensor` | torch.Tensor: Penalty for base height away from target |

Source code in `genesis_forge/mdp/rewards.py`

```
def base_height(
    env: GenesisEnv,
    target_height: Union[float, torch.Tensor] = None,
    height_command: CommandManager = None,
    terrain_manager: TerrainManager = None,
    entity_attr: str = "robot",
    entity_manager: EntityManager = None,
) -> torch.Tensor:
    """
    Penalize base height away from target, using the L2 squared kernel.

    Args:
        env: The Genesis environment containing the robot
        target_height: The target height to penalize the base height away from
        height_command: Get the target height from a height command manager. This expects the command to have a single range value.
        terrain_manager: The terrain manager will adjust the height based on the terrain height.
        entity_attr: The attribute name of the entity in the environment.
        entity_manager: The entity manager for the entity.

    Returns:
        torch.Tensor: Penalty for base height away from target
    """
    robot = None
    if entity_manager is not None:
        robot = entity_manager.entity
    else:
        robot = getattr(env, entity_attr)

    base_pos = robot.get_pos()
    height_offset = 0.0
    if terrain_manager is not None:
        height_offset = terrain_manager.get_terrain_height(
            base_pos[:, 0], base_pos[:, 1]
        )
    if height_command is not None:
        target_height = height_command.command.squeeze(-1)
    return torch.square(base_pos[:, 2] - height_offset - target_height)
```

## `command_tracking_ang_vel(env, commanded_ang_vel=None, vel_cmd_manager=None, sensitivity=0.25, entity_attr='robot', entity_manager=None)`

Reward for tracking commanded angular velocity (yaw)

Parameters:

| Name                | Type                     | Description                                                                                                                                      | Default    |
| ------------------- | ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------ | ---------- |
| `env`               | `GenesisEnv`             | The Genesis Forge environment                                                                                                                    | *required* |
| `commanded_ang_vel` | `Tensor`                 | The commanded angular velocity in the shape (num_envs, 1)                                                                                        | `None`     |
| `vel_cmd_manager`   | `VelocityCommandManager` | The velocity command manager                                                                                                                     | `None`     |
| `sensitivity`       | `float`                  | A lower value means the reward is more sensitive to the error                                                                                    | `0.25`     |
| `entity_manager`    | `EntityManager`          | The entity manager for the robot/entity the reward is being computed for. This is slightly more performant than using the entity_attr parameter. | `None`     |
| `entity_attr`       | `str`                    | The attribute name of the entity in the environment. This isn't necessary if entity_manager is provided.                                         | `'robot'`  |

Returns:

| Type     | Description                                                          |
| -------- | -------------------------------------------------------------------- |
| `Tensor` | torch.Tensor: Reward for tracking of angular velocity commands (yaw) |

Source code in `genesis_forge/mdp/rewards.py`

```
def command_tracking_ang_vel(
    env: GenesisEnv,
    commanded_ang_vel: torch.Tensor = None,
    vel_cmd_manager: VelocityCommandManager = None,
    sensitivity: float = 0.25,
    entity_attr: str = "robot",
    entity_manager: EntityManager = None,
) -> torch.Tensor:
    """
    Reward for tracking commanded angular velocity (yaw)

    Args:
        env: The Genesis Forge environment
        commanded_ang_vel: The commanded angular velocity in the shape (num_envs, 1)
        vel_cmd_manager: The velocity command manager
        sensitivity: A lower value means the reward is more sensitive to the error
        entity_manager: The entity manager for the robot/entity the reward is being computed for.
                        This is slightly more performant than using the `entity_attr` parameter.
        entity_attr: The attribute name of the entity in the environment. This isn't necessary if `entity_manager` is provided.

    Returns:
        torch.Tensor: Reward for tracking of angular velocity commands (yaw)
    """
    assert (
        commanded_ang_vel is not None or vel_cmd_manager is not None
    ), "Either commanded_ang_vel or vel_cmd_manager must be provided to command_tracking_ang_vel"

    angular_vel = None
    if entity_manager is not None:
        angular_vel = entity_manager.get_angular_velocity()
    else:
        robot = getattr(env, entity_attr)
        angular_vel = entity_ang_vel(robot)

    if vel_cmd_manager is not None:
        commanded_ang_vel = vel_cmd_manager.command[:, 2]

    ang_vel_error = torch.square(commanded_ang_vel - angular_vel[:, 2])
    return torch.exp(-ang_vel_error / sensitivity)
```

## `command_tracking_lin_vel(env, command=None, vel_cmd_manager=None, sensitivity=0.25, entity_attr='robot', entity_manager=None)`

Reward for tracking commanded linear velocity (xy axes)

Parameters:

| Name              | Type                     | Description                                                                                                                                      | Default    |
| ----------------- | ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------ | ---------- |
| `env`             | `GenesisEnv`             | The Genesis environment containing the robot                                                                                                     | *required* |
| `command`         | `Tensor`                 | The commanded XY linear velocity in the shape (num_envs, 2)                                                                                      | `None`     |
| `vel_cmd_manager` | `VelocityCommandManager` | The velocity command manager                                                                                                                     | `None`     |
| `sensitivity`     | `float`                  | A lower value means the reward is more sensitive to the error                                                                                    | `0.25`     |
| `entity_manager`  | `EntityManager`          | The entity manager for the robot/entity the reward is being computed for. This is slightly more performant than using the entity_attr parameter. | `None`     |
| `entity_attr`     | `str`                    | The attribute name of the entity in the environment. This isn't necessary if entity_manager is provided.                                         | `'robot'`  |

Returns:

| Type     | Description                                                             |
| -------- | ----------------------------------------------------------------------- |
| `Tensor` | torch.Tensor: Reward for tracking of linear velocity commands (xy axes) |

Source code in `genesis_forge/mdp/rewards.py`

```
def command_tracking_lin_vel(
    env: GenesisEnv,
    command: torch.Tensor = None,
    vel_cmd_manager: VelocityCommandManager = None,
    sensitivity: float = 0.25,
    entity_attr: str = "robot",
    entity_manager: EntityManager = None,
) -> torch.Tensor:
    """
    Reward for tracking commanded linear velocity (xy axes)

    Args:
        env: The Genesis environment containing the robot
        command: The commanded XY linear velocity in the shape (num_envs, 2)
        vel_cmd_manager: The velocity command manager
        sensitivity: A lower value means the reward is more sensitive to the error
        entity_manager: The entity manager for the robot/entity the reward is being computed for.
                        This is slightly more performant than using the `entity_attr` parameter.
        entity_attr: The attribute name of the entity in the environment. This isn't necessary if `entity_manager` is provided.

    Returns:
        torch.Tensor: Reward for tracking of linear velocity commands (xy axes)
    """
    assert (
        command is not None or vel_cmd_manager is not None
    ), "Either command or vel_cmd_manager must be provided to command_tracking_lin_vel"

    linear_vel_local = None
    if entity_manager is not None:
        linear_vel_local = entity_manager.get_linear_velocity()
    else:
        robot = getattr(env, entity_attr)
        linear_vel_local = entity_lin_vel(robot)

    if vel_cmd_manager is not None:
        command = vel_cmd_manager.command[:, :2]

    lin_vel_error = torch.sum(torch.square(command - linear_vel_local[:, :2]), dim=1)
    return torch.exp(-lin_vel_error / sensitivity)
```

## `contact_force(env, contact_manager, threshold=1.0)`

Reward for the total contact force acting on all the target links in the contact manager over the threshold.

Parameters:

| Name              | Type             | Description                                                | Default    |
| ----------------- | ---------------- | ---------------------------------------------------------- | ---------- |
| `env`             | `GenesisEnv`     | The Genesis Forge environment                              | *required* |
| `contact_manager` | `ContactManager` | The contact manager to check for contact                   | *required* |
| `threshold`       | `float`          | The force threshold for contact detection (default: 1.0 N) | `1.0`      |

Returns:

| Type     | Description                                                  |
| -------- | ------------------------------------------------------------ |
| `Tensor` | The total force for the contact manager for each environment |

Source code in `genesis_forge/mdp/rewards.py`

```
def contact_force(
    env: GenesisEnv, contact_manager: ContactManager, threshold: float = 1.0
) -> torch.Tensor:
    """
    Reward for the total contact force acting on all the target links in the contact manager over the threshold.

    Args:
        env: The Genesis Forge environment
        contact_manager: The contact manager to check for contact
        threshold: The force threshold for contact detection (default: 1.0 N)

    Returns:
        The total force for the contact manager for each environment
    """
    violation = torch.norm(contact_manager.contacts[:, :, :], dim=-1) - threshold
    return torch.sum(violation.clip(min=0.0), dim=1)
```

## `dof_similar_to_default(env, actuator_manager=None, action_manager=None)`

Penalize joint poses far away from default pose(s).

Pass `actuator_manager` as one manager or a non-empty list/tuple (e.g. per-limb stacks); penalties are summed per environment across all included DOFs.

Parameters:

| Name               | Type                    | Description                                                                                          | Default                                                                          |
| ------------------ | ----------------------- | ---------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------- |
| `env`              | `GenesisEnv`            | The Genesis environment containing the robot (unused today; accepted for MDP signature consistency). | *required*                                                                       |
| `actuator_manager` | \`ActuatorManager       | list[ActuatorManager]                                                                                | None\`                                                                           |
| `action_manager`   | \`PositionActionManager | None\`                                                                                               | (deprecated) One or more position-action managers. Use actuator_manager instead. |

Returns:

| Type     | Description                                                         |
| -------- | ------------------------------------------------------------------- |
| `Tensor` | torch.Tensor: Penalty summed over included DOFs, shape (num_envs,). |

Source code in `genesis_forge/mdp/rewards.py`

```
def dof_similar_to_default(
    env: GenesisEnv,
    actuator_manager: ActuatorManager | list[ActuatorManager] | None = None,
    action_manager: PositionActionManager | None = None,
) -> torch.Tensor:
    """
    Penalize joint poses far away from default pose(s).

    Pass ``actuator_manager`` as one manager or a non-empty list/tuple (e.g. per-limb
    stacks); penalties are summed per environment across all included DOFs.

    Args:
        env: The Genesis environment containing the robot (unused today; accepted for MDP signature consistency).
        actuator_manager: One or more actuator managers.
        action_manager: (deprecated) One or more position-action managers. Use
            ``actuator_manager`` instead.

    Returns:
        torch.Tensor: Penalty summed over included DOFs, shape ``(num_envs,)``.
    """
    if actuator_manager is not None:
        if isinstance(actuator_manager, list):
            total = None
            for mgr in actuator_manager:
                dof_pos = mgr.get_dofs_position()
                part = torch.sum(torch.abs(dof_pos - mgr.default_dofs_pos), dim=1)
                total = part if total is None else total + part
            return total
        else:
            dof_pos = actuator_manager.get_dofs_position()
            default_pos = actuator_manager.default_dofs_pos
            return torch.sum(torch.abs(dof_pos - default_pos), dim=1)
    elif action_manager is not None:
        dof_pos = action_manager.get_dofs_position()
        default_pos = action_manager.default_dofs_pos
        return torch.sum(torch.abs(dof_pos - default_pos), dim=1)

    raise ValueError("dof_similar_to_default: Either actuator_manager or action_manager must be provided")
```

## `dof_torque_l2(env, actuator_manager)`

Penalize joint torque effort using the L2 squared kernel.

Discourages the policy from applying unnecessary force, particularly when the robot is near equilibrium. This helps reduce actuator oscillation when the robot is stationary or moving slowly.

Parameters:

| Name               | Type              | Description                                       | Default    |
| ------------------ | ----------------- | ------------------------------------------------- | ---------- |
| `env`              | `GenesisEnv`      | The Genesis environment containing the robot      | *required* |
| `actuator_manager` | `ActuatorManager` | The actuator manager to retrieve DOF forces from. | *required* |

Returns:

| Type     | Description                                                      |
| -------- | ---------------------------------------------------------------- |
| `Tensor` | torch.Tensor: Penalty for joint torque effort, shape (num_envs,) |

Source code in `genesis_forge/mdp/rewards.py`

```
def dof_torque_l2(
    env: GenesisEnv,
    actuator_manager: ActuatorManager,
) -> torch.Tensor:
    """
    Penalize joint torque effort using the L2 squared kernel.

    Discourages the policy from applying unnecessary force, particularly when the
    robot is near equilibrium. This helps reduce actuator oscillation when the robot
    is stationary or moving slowly.

    Args:
        env: The Genesis environment containing the robot
        actuator_manager: The actuator manager to retrieve DOF forces from.

    Returns:
        torch.Tensor: Penalty for joint torque effort, shape (num_envs,)
    """
    torque = actuator_manager.get_dofs_control_force()
    return torch.sum(torch.square(torque), dim=1)
```

## `dof_velocity_l2(env, action_manager)`

Penalize joint angular velocities to encourage slow, deliberate motion.

Source code in `genesis_forge/mdp/rewards.py`

```
def dof_velocity_l2(env: GenesisEnv, action_manager: PositionActionManager) -> torch.Tensor:
    """Penalize joint angular velocities to encourage slow, deliberate motion."""
    dof_vel = action_manager.get_dofs_velocity()
    return torch.sum(torch.square(dof_vel), dim=1)
```

## `feet_air_time(env, contact_manager, time_threshold, time_threshold_max=None, vel_cmd_manager=None)`

Reward long steps taken by the feet using L2-kernel.

This function rewards the agent for taking steps that are longer than a threshold. This helps ensure that the robot lifts its feet off the ground and takes steps. The reward is computed as the sum of the time for which the feet are in the air.

If the velocity commands are small (i.e. the agent is not supposed to take a step), then the reward is zero.

Parameters:

| Name                 | Type                     | Description                                                 | Default                                                                                                          |
| -------------------- | ------------------------ | ----------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------- |
| `env`                | `GenesisEnv`             | The Genesis Forge environment                               | *required*                                                                                                       |
| `contact_manager`    | `ContactManager`         | The contact manager to check for contact                    | *required*                                                                                                       |
| `time_threshold`     | `float`                  | The minimum time (in seconds) the feet should be in the air | *required*                                                                                                       |
| `time_threshold_max` | \`float                  | None\`                                                      | (optional) The maximum time (in seconds) the feet should be in the air. The reward will be capped at this value. |
| `vel_cmd_manager`    | \`VelocityCommandManager | None\`                                                      | The velocity command manager                                                                                     |

Returns:

| Type     | Description                      |
| -------- | -------------------------------- |
| `Tensor` | The reward for the feet air time |

Source code in `genesis_forge/mdp/rewards.py`

```
def feet_air_time(
    env: GenesisEnv,
    contact_manager: ContactManager,
    time_threshold: float,
    time_threshold_max: float | None = None,
    vel_cmd_manager: VelocityCommandManager | None = None,
) -> torch.Tensor:
    """Reward long steps taken by the feet using L2-kernel.

    This function rewards the agent for taking steps that are longer than a threshold. This helps ensure
    that the robot lifts its feet off the ground and takes steps. The reward is computed as the sum of
    the time for which the feet are in the air.

    If the velocity commands are small (i.e. the agent is not supposed to take a step), then the reward is zero.

    Args:
        env: The Genesis Forge environment
        contact_manager: The contact manager to check for contact
        time_threshold: The minimum time (in seconds) the feet should be in the air
        time_threshold_max: (optional) The maximum time (in seconds) the feet should be in the air.
                            The reward will be capped at this value.
        vel_cmd_manager: The velocity command manager

    Returns:
        The reward for the feet air time
    """
    made_contact = contact_manager.has_made_contact(env.dt)
    last_air_time = contact_manager.last_air_time

    # Calculate the air time
    air_time = (last_air_time - time_threshold) * made_contact
    if time_threshold_max is not None:
        air_time = torch.clamp(air_time, max=time_threshold_max - time_threshold)
    reward = torch.sum(air_time, dim=1)

    # no reward for zero velocity command
    if vel_cmd_manager is not None:
        reward *= torch.norm(vel_cmd_manager.command[:, :2], dim=1) > 0.1
    return reward
```

## `feet_ground_time(env, contact_manager, time_threshold)`

Penalize brief ground contacts (foot tapping) using a linear kernel.

Fires at the moment a foot lifts off. The penalty is proportional to how much the stance duration fell below time_threshold. A proper stance phase (contact_time >= time_threshold) produces zero penalty.

Intended to be paired with feet_air_time (positive reward) to fully shape gait timing: feet_air_time rewards long swings while this penalizes taps. Use a negative weight in the RewardManager.

Parameters:

| Name              | Type             | Description                                                                                                                                                          | Default    |
| ----------------- | ---------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------- |
| `env`             | `GenesisEnv`     | The Genesis Forge environment                                                                                                                                        | *required* |
| `contact_manager` | `ContactManager` | The contact manager to check for contact                                                                                                                             | *required* |
| `time_threshold`  | `float`          | Contacts shorter than this (in seconds) are penalized. Set independently from the feet_air_time threshold based on the expected stance duration of your target gait. | *required* |

Returns:

| Type     | Description                                              |
| -------- | -------------------------------------------------------- |
| `Tensor` | The penalty for brief ground contacts, shape (num_envs,) |

Source code in `genesis_forge/mdp/rewards.py`

```
def feet_ground_time(
    env: GenesisEnv,
    contact_manager: ContactManager,
    time_threshold: float,
) -> torch.Tensor:
    """Penalize brief ground contacts (foot tapping) using a linear kernel.

    Fires at the moment a foot lifts off. The penalty is proportional to how
    much the stance duration fell below time_threshold. A proper stance phase
    (contact_time >= time_threshold) produces zero penalty.

    Intended to be paired with feet_air_time (positive reward) to fully shape
    gait timing: feet_air_time rewards long swings while this penalizes taps.
    Use a negative weight in the RewardManager.

    Args:
        env: The Genesis Forge environment
        contact_manager: The contact manager to check for contact
        time_threshold: Contacts shorter than this (in seconds) are penalized.
                        Set independently from the feet_air_time threshold based
                        on the expected stance duration of your target gait.

    Returns:
        The penalty for brief ground contacts, shape (num_envs,)
    """
    just_lifted = contact_manager.has_broken_contact(env.dt)
    last_contact_time = contact_manager.last_contact_time
    short_contact = (time_threshold - last_contact_time).clamp(min=0.0) * just_lifted
    return torch.sum(short_contact, dim=1)
```

## `feet_slide(env, contact_manager, entity_attr='robot')`

Penalize feet sliding.

This function penalizes the agent for sliding its feet on the ground. The reward is computed as the norm of the linear velocity of the feet multiplied by a binary contact sensor. This ensures that the agent is penalized only when the feet are in contact with the ground.

This penalty is less effective at longer foot-contact links (for example, long legs without dedicated foot links), because they might have some velocity while they're being used to move the robot. However, dedicated foot links will be stationary on the ground and not moving while pushing the robot forward.

Parameters:

| Name              | Type             | Description                                                           | Default    |
| ----------------- | ---------------- | --------------------------------------------------------------------- | ---------- |
| `env`             | `GenesisEnv`     | The Genesis Forge environment                                         | *required* |
| `contact_manager` | `ContactManager` | The contact manager for the feet                                      | *required* |
| `entity_attr`     | `str`            | The attribute name of the robot entity that the feet are attached to. | `'robot'`  |

Returns:

| Type     | Description                    |
| -------- | ------------------------------ |
| `Tensor` | The penalty for the feet slide |

Source code in `genesis_forge/mdp/rewards.py`

```
def feet_slide(
    env: GenesisEnv,
    contact_manager: ContactManager,
    entity_attr: str = "robot",
) -> torch.Tensor:
    """Penalize feet sliding.

    This function penalizes the agent for sliding its feet on the ground. The reward is computed as the
    norm of the linear velocity of the feet multiplied by a binary contact sensor. This ensures that the
    agent is penalized only when the feet are in contact with the ground.

    This penalty is less effective at longer foot-contact links (for example, long legs without dedicated foot links),
    because they might have some velocity while they're being used to move the robot. However, dedicated foot links
    will be stationary on the ground and not moving while pushing the robot forward.

    Args:
        env: The Genesis Forge environment
        contact_manager: The contact manager for the feet
        entity_attr: The attribute name of the robot entity that the feet are attached to.

    Returns:
        The penalty for the feet slide
    """
    # Get links in contact
    contacts = torch.norm(contact_manager.contacts[:, :, :], dim=-1) > 1.0

    # Get link velocities.
    # If the links aren't moving, then they're being used to move the robot and not sliding.
    link_ids = contact_manager.local_link_ids
    robot: RigidEntity = getattr(env, entity_attr)
    link_vel = robot.get_links_vel(links_idx_local=link_ids)

    return torch.sum(link_vel.norm(dim=-1) * contacts, dim=1)
```

## `flat_orientation_l2(env, entity_attr='robot', entity_manager=None)`

Penalize non-flat base orientation using L2 squared kernel. This is computed by penalizing the xy-components of the projected gravity vector.

Parameters:

| Name             | Type            | Description                                                                                                                                      | Default    |
| ---------------- | --------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | ---------- |
| `env`            | `GenesisEnv`    | The Genesis environment containing the robot                                                                                                     | *required* |
| `entity_manager` | `EntityManager` | The entity manager for the robot/entity the reward is being computed for. This is slightly more performant than using the entity_attr parameter. | `None`     |
| `entity_attr`    | `str`           | The attribute name of the entity in the environment. This isn't necessary if entity_manager is provided.                                         | `'robot'`  |

Returns:

| Type     | Description                                         |
| -------- | --------------------------------------------------- |
| `Tensor` | torch.Tensor: Penalty for non-flat base orientation |

Source code in `genesis_forge/mdp/rewards.py`

```
def flat_orientation_l2(
    env: GenesisEnv,
    entity_attr: str = "robot",
    entity_manager: EntityManager = None,
) -> torch.Tensor:
    """
    Penalize non-flat base orientation using L2 squared kernel.
    This is computed by penalizing the xy-components of the projected gravity vector.

    Args:
        env: The Genesis environment containing the robot
        entity_manager: The entity manager for the robot/entity the reward is being computed for.
                        This is slightly more performant than using the `entity_attr` parameter.
        entity_attr: The attribute name of the entity in the environment. This isn't necessary if `entity_manager` is provided.

    Returns:
        torch.Tensor: Penalty for non-flat base orientation
    """
    # Get the projected gravity vector in the robot's base frame
    # This represents how "tilted" the robot is from upright
    projected_gravity = None
    if entity_manager is not None:
        projected_gravity = entity_manager.get_projected_gravity()
    else:
        robot = getattr(env, entity_attr)
        projected_gravity = entity_projected_gravity(robot)

    # Penalize the xy-components (horizontal tilt) using L2 squared kernel
    # A flat orientation means these components should be close to zero
    return torch.sum(torch.square(projected_gravity[:, :2]), dim=1)
```

## `has_contact(env, contact_manager, threshold=1.0, min_contacts=1)`

One or more links in the contact manager are in contact with something.

Parameters:

| Name              | Type             | Description                                              | Default    |
| ----------------- | ---------------- | -------------------------------------------------------- | ---------- |
| `env`             | `GenesisEnv`     | The Genesis Forge environment                            | *required* |
| `contact_manager` | `ContactManager` | The contact manager to check for contact                 | *required* |
| `threshold`       | `float`          | The force threshold for contact detection (default: 1.0) | `1.0`      |
| `min_contacts`    | `int`            | The minimum number of contacts required. (default: 1)    | `1`        |

Returns:

| Type     | Description                              |
| -------- | ---------------------------------------- |
| `Tensor` | 1 for each contact meeting the threshold |

Source code in `genesis_forge/mdp/rewards.py`

```
def has_contact(
    env: GenesisEnv, contact_manager: ContactManager, threshold: float=1.0, min_contacts: int=1
) -> torch.Tensor:
    """
    One or more links in the contact manager are in contact with something.

    Args:
        env: The Genesis Forge environment
        contact_manager: The contact manager to check for contact
        threshold: The force threshold for contact detection (default: 1.0)
        min_contacts: The minimum number of contacts required. (default: 1)

    Returns:
        1 for each contact meeting the threshold
    """
    has_contact = contact_manager.contacts[:, :].norm(dim=-1) > threshold
    result = has_contact.sum(dim=1) >= min_contacts
    return result.float()
```

## `is_alive(env)`

Reward for being alive and not terminating this step. This assumes that `env.extras["terminations"]` is a boolean tensor with the termination signals for the environments.

Source code in `genesis_forge/mdp/rewards.py`

```
def is_alive(env: GenesisEnv) -> torch.Tensor:
    """
    Reward for being alive and not terminating this step.
    This assumes that `env.extras["terminations"]` is a boolean tensor with the termination signals for the environments.
    """
    terminations: torch.Tensor = env.extras["terminations"]
    return (~terminations).float().detach()
```

## `lin_vel_xy_l2(env, entity_manager)`

Penalize horizontal base linear velocity.

Source code in `genesis_forge/mdp/rewards.py`

```
def lin_vel_xy_l2(env: GenesisEnv, entity_manager) -> torch.Tensor:
    """Penalize horizontal base linear velocity."""
    lin_vel = entity_manager.get_linear_velocity()
    return torch.sum(torch.square(lin_vel[:, :2]), dim=1)
```

## `lin_vel_z_l2(env, entity_attr='robot', entity_manager=None)`

Penalize z axis base linear velocity

Parameters:

| Name             | Type            | Description                                                                                                                                      | Default    |
| ---------------- | --------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | ---------- |
| `env`            | `GenesisEnv`    | The Genesis environment containing the entity                                                                                                    | *required* |
| `entity_manager` | `EntityManager` | The entity manager for the robot/entity the reward is being computed for. This is slightly more performant than using the entity_attr parameter. | `None`     |
| `entity_attr`    | `str`           | The attribute name of the entity in the environment. This isn't necessary if entity_manager is provided.                                         | `'robot'`  |

Returns:

| Type     | Description                                           |
| -------- | ----------------------------------------------------- |
| `Tensor` | torch.Tensor: Penalty for z axis base linear velocity |

Source code in `genesis_forge/mdp/rewards.py`

```
def lin_vel_z_l2(
    env: GenesisEnv,
    entity_attr: str = "robot",
    entity_manager: EntityManager = None,
) -> torch.Tensor:
    """
    Penalize z axis base linear velocity

    Args:
        env: The Genesis environment containing the entity
        entity_manager: The entity manager for the robot/entity the reward is being computed for.
                        This is slightly more performant than using the `entity_attr` parameter.
        entity_attr: The attribute name of the entity in the environment. This isn't necessary if `entity_manager` is provided.

    Returns:
        torch.Tensor: Penalty for z axis base linear velocity
    """
    linear_vel = None
    if entity_manager is not None:
        linear_vel = entity_manager.get_linear_velocity()
    else:
        robot = getattr(env, entity_attr)
        linear_vel = entity_lin_vel(robot)
    return torch.square(linear_vel[:, 2])
```

## `stand_still_joint_deviation_l1(env, vel_cmd_manager, actuator_manager=None, command_threshold=0.06, action_manager=None)`

Penalize offsets from the default joint positions when the command is very small.

Parameters:

| Name                | Type                     | Description                                                              | Default    |
| ------------------- | ------------------------ | ------------------------------------------------------------------------ | ---------- |
| `env`               | `GenesisEnv`             | The Genesis Forge environment                                            | *required* |
| `command_threshold` | `float`                  | The threshold for the command to be considered small                     | `0.06`     |
| `vel_cmd_manager`   | `VelocityCommandManager` | The velocity command manager                                             | *required* |
| `actuator_manager`  | `ActuatorManager`        | The actuator manager to get the joint positions and recent actions from. | `None`     |
| `action_manager`    | `PositionActionManager`  | The action manager to get the joint positions and recent actions from.   | `None`     |

Returns:

| Type     | Description                                                                                       |
| -------- | ------------------------------------------------------------------------------------------------- |
| `Tensor` | torch.Tensor: Penalty for offsets from the default joint positions when the command is very small |

Source code in `genesis_forge/mdp/rewards.py`

```
def stand_still_joint_deviation_l1(
    env: GenesisEnv,
    vel_cmd_manager: VelocityCommandManager,
    actuator_manager: ActuatorManager = None,
    command_threshold: float = 0.06,
    action_manager: PositionActionManager = None,
) -> torch.Tensor:
    """
    Penalize offsets from the default joint positions when the command is very small.

    Args:
        env: The Genesis Forge environment
        command_threshold: The threshold for the command to be considered small
        vel_cmd_manager: The velocity command manager
        actuator_manager: The actuator manager to get the joint positions and recent actions from.
        action_manager: The action manager to get the joint positions and recent actions from.

    Returns:
        torch.Tensor: Penalty for offsets from the default joint positions when the command is very small
    """
    assert (
        actuator_manager is not None or action_manager is not None
    ), "Either actuator_manager or action_manager must be provided to stand_still_joint_deviation_l1"

    if actuator_manager is not None:
        joint_pos = actuator_manager.get_dofs_position()
        default_pos = actuator_manager.default_dofs_pos
    elif action_manager is not None:
        joint_pos = action_manager.get_dofs_position()
        default_pos = action_manager.default_dofs_pos
    joint_deviation = torch.sum(torch.abs(joint_pos - default_pos), dim=1)

    # Penalize motion when command is nearly zero.
    command = vel_cmd_manager.command
    return joint_deviation * (torch.norm(command[:, :2], dim=1) < command_threshold)
```

## `terminated(env)`

Penalize terminated episodes that terminated. This assumes that `env.extras["terminations"]` is a boolean tensor with the termination signals for the environments.

Source code in `genesis_forge/mdp/rewards.py`

```
def terminated(env: GenesisEnv) -> torch.Tensor:
    """
    Penalize terminated episodes that terminated.
    This assumes that `env.extras["terminations"]` is a boolean tensor with the termination signals for the environments.
    """
    terminations: torch.Tensor = env.extras["terminations"]
    return terminations.float().detach()
```

# Terminations

## `bad_orientation(env, limit_angle=40.0, entity_attr='robot', entity_manager=None, grace_steps=0)`

Terminate the environment if the robot is tipping over too much.

This function uses projected gravity to detect when the robot has tilted beyond a safe threshold. When the robot is perfectly upright, projected gravity should be [0, 0, -1] in the body frame. As the robot tilts, the x,y components increase, indicating roll and pitch angles.

Parameters:

| Name             | Type            | Description                                                                                                                                        | Default    |
| ---------------- | --------------- | -------------------------------------------------------------------------------------------------------------------------------------------------- | ---------- |
| `env`            | `GenesisEnv`    | The Genesis environment containing the robot                                                                                                       | *required* |
| `limit_angle`    | `float`         | Maximum allowed tilt angle in degrees (default: 40 degrees)                                                                                        | `40.0`     |
| `entity_manager` | `EntityManager` | The entity manager for the entity.                                                                                                                 | `None`     |
| `entity_attr`    | `str`           | The attribute name of the entity in the environment. This isn't necessary if entity_manager is provided.                                           | `'robot'`  |
| `grace_steps`    | `int`           | Number of steps at episode start to ignore tilt detection (default: 0) This gives the robot a chance to stabilize before tilt detection is active. | `0`        |

Returns:

| Type     | Description                                                                 |
| -------- | --------------------------------------------------------------------------- |
| `Tensor` | torch.Tensor: Boolean tensor indicating which environments should terminate |

Source code in `genesis_forge/mdp/terminations.py`

```
def bad_orientation(
    env: GenesisEnv,
    limit_angle: float = 40.0,
    entity_attr: str = "robot",
    entity_manager: EntityManager = None,
    grace_steps: int = 0,
) -> torch.Tensor:
    """
    Terminate the environment if the robot is tipping over too much.

    This function uses projected gravity to detect when the robot has tilted
    beyond a safe threshold. When the robot is perfectly upright, projected
    gravity should be [0, 0, -1] in the body frame. As the robot tilts,
    the x,y components increase, indicating roll and pitch angles.

    Args:
        env: The Genesis environment containing the robot
        limit_angle: Maximum allowed tilt angle in degrees (default: 40 degrees)
        entity_manager: The entity manager for the entity.
        entity_attr: The attribute name of the entity in the environment.
                        This isn't necessary if `entity_manager` is provided.
        grace_steps: Number of steps at episode start to ignore tilt detection (default: 0)
                     This gives the robot a chance to stabilize before tilt detection is active.

    Returns:
        torch.Tensor: Boolean tensor indicating which environments should terminate
    """
    in_grace_period = env.episode_length <= grace_steps

    # Get the projected gravity vector in body frame
    projected_gravity = None
    if entity_manager is not None:
        projected_gravity = entity_manager.get_projected_gravity()
    else:
        entity = getattr(env, entity_attr)
        projected_gravity = entity_projected_gravity(entity)

    # Calculate the magnitude of tilt (distance from perfectly upright)
    projected_gravity_xy = projected_gravity[:, :2]
    tilt_magnitude = torch.norm(projected_gravity_xy, dim=1)

    # Convert tilt magnitude to angle
    tilt_angle = torch.asin(torch.clamp(tilt_magnitude, max=0.99))

    # Terminate if tilt angle exceeds the limit
    return (~in_grace_period) & (tilt_angle > math.radians(limit_angle))
```

## `base_height_below_minimum(env, minimum_height=0.05, entity_attr='robot', entity_manager=None)`

Terminate the environment if the robot's base height falls below a minimum threshold.

Parameters:

| Name             | Type            | Description                                                                                              | Default    |
| ---------------- | --------------- | -------------------------------------------------------------------------------------------------------- | ---------- |
| `env`            | `GenesisEnv`    | The Genesis environment containing the robot                                                             | *required* |
| `minimum_height` | `float`         | Minimum allowed base height in meters                                                                    | `0.05`     |
| `entity_manager` | `EntityManager` | The entity manager for the entity.                                                                       | `None`     |
| `entity_attr`    | `str`           | The attribute name of the entity in the environment. This isn't necessary if entity_manager is provided. | `'robot'`  |

Returns:

| Type     | Description                                                                 |
| -------- | --------------------------------------------------------------------------- |
| `Tensor` | torch.Tensor: Boolean tensor indicating which environments should terminate |

Source code in `genesis_forge/mdp/terminations.py`

```
def base_height_below_minimum(
    env: GenesisEnv,
    minimum_height: float = 0.05,
    entity_attr: str = "robot",
    entity_manager: EntityManager = None,
) -> torch.Tensor:
    """
    Terminate the environment if the robot's base height falls below a minimum threshold.

    Args:
        env: The Genesis environment containing the robot
        minimum_height: Minimum allowed base height in meters
        entity_manager: The entity manager for the entity.
        entity_attr: The attribute name of the entity in the environment.
                        This isn't necessary if `entity_manager` is provided.

    Returns:
        torch.Tensor: Boolean tensor indicating which environments should terminate
    """
    base_pos = None
    if entity_manager is not None:
        base_pos = entity_manager.base_pos
    else:
        entity = getattr(env, entity_attr)
        base_pos = entity.get_pos()
    return base_pos[:, 2] < minimum_height
```

## `contact_force(env, contact_manager, threshold=1.0)`

Terminate if any link in the contact manager is in contact with something with a force greater than the threshold.

Parameters:

| Name              | Type             | Description                                                | Default    |
| ----------------- | ---------------- | ---------------------------------------------------------- | ---------- |
| `env`             | `GenesisEnv`     | The Genesis environment containing the robot               | *required* |
| `contact_manager` | `ContactManager` | The contact manager to check for contact                   | *required* |
| `threshold`       | `float`          | The force threshold for contact detection (default: 1.0 N) | `1.0`      |

Returns:

| Type     | Description                                                  |
| -------- | ------------------------------------------------------------ |
| `Tensor` | The total force for the contact manager for each environment |

Source code in `genesis_forge/mdp/terminations.py`

```
def contact_force(
    env: GenesisEnv, contact_manager: ContactManager, threshold: float = 1.0
) -> torch.Tensor:
    """
    Terminate if any link in the contact manager is in contact with something with a force greater than the threshold.

    Args:
        env: The Genesis environment containing the robot
        contact_manager: The contact manager to check for contact
        threshold: The force threshold for contact detection (default: 1.0 N)

    Returns:
        The total force for the contact manager for each environment
    """
    return torch.any(torch.norm(contact_manager.contacts, dim=-1) > threshold, dim=-1)
```

## `contact_force_with_grace_period(env, contact_manager, threshold=100.0, grace_steps=10)`

Terminate if contact force exceeds threshold, with a grace period at episode start.

This is useful for quadrupeds that may start in slightly unstable positions and need a few steps to stabilize before fall detection becomes active.

Parameters:

| Name              | Type             | Description                                                       | Default    |
| ----------------- | ---------------- | ----------------------------------------------------------------- | ---------- |
| `env`             | `GenesisEnv`     | The Genesis environment containing the robot                      | *required* |
| `contact_manager` | `ContactManager` | The contact manager to check for contact                          | *required* |
| `threshold`       | `float`          | The force threshold for contact detection (default: 100.0 N)      | `100.0`    |
| `grace_steps`     | `int`            | Number of steps at episode start to ignore contacts (default: 10) | `10`       |

Returns:

| Type     | Description                                                   |
| -------- | ------------------------------------------------------------- |
| `Tensor` | Boolean tensor indicating which environments should terminate |

Source code in `genesis_forge/mdp/terminations.py`

```
def contact_force_with_grace_period(
    env: GenesisEnv,
    contact_manager: ContactManager,
    threshold: float = 100.0,
    grace_steps: int = 10,
) -> torch.Tensor:
    """
    Terminate if contact force exceeds threshold, with a grace period at episode start.

    This is useful for quadrupeds that may start in slightly unstable positions
    and need a few steps to stabilize before fall detection becomes active.

    Args:
        env: The Genesis environment containing the robot
        contact_manager: The contact manager to check for contact
        threshold: The force threshold for contact detection (default: 100.0 N)
        grace_steps: Number of steps at episode start to ignore contacts (default: 10)

    Returns:
        Boolean tensor indicating which environments should terminate
    """
    # Don't terminate during grace period (early in episode)
    in_grace_period = env.episode_length <= grace_steps

    # Check contact forces
    contact_exceeded = torch.any(
        torch.norm(contact_manager.contacts, dim=-1) > threshold, dim=-1
    )

    # Only terminate if past grace period AND contact exceeded
    return (~in_grace_period) & contact_exceeded.detach()
```

## `dof_control_force_limit(env, actuator_manager, threshold=None)`

Terminate if any joint's commanded actuator force exceeds a limit (+/-).

Uses control/output force (what the actuator commands), not measured joint load. Suitable for teaching policies to stay within rated motor torque.

Parameters:

| Name               | Type              | Description                                | Default                                                                                       |
| ------------------ | ----------------- | ------------------------------------------ | --------------------------------------------------------------------------------------------- |
| `env`              | `GenesisEnv`      | The Genesis environment                    | *required*                                                                                    |
| `actuator_manager` | `ActuatorManager` | Actuator manager for the controlled joints | *required*                                                                                    |
| `threshold`        | \`float           | None\`                                     | Force/torque limit (in simulator units). If None, uses max_force value from actuator_manager. |

Returns:

| Type     | Description                                                   |
| -------- | ------------------------------------------------------------- |
| `Tensor` | Boolean tensor indicating which environments should terminate |

Source code in `genesis_forge/mdp/terminations.py`

```
def dof_control_force_limit(
    env: GenesisEnv,
    actuator_manager: ActuatorManager,
    threshold: float | None = None,
) -> torch.Tensor:
    """
    Terminate if any joint's commanded actuator force exceeds a limit (+/-).

    Uses control/output force (what the actuator commands), not measured joint load.
    Suitable for teaching policies to stay within rated motor torque.

    Args:
        env: The Genesis environment
        actuator_manager: Actuator manager for the controlled joints
        threshold: Force/torque limit (in simulator units). 
                   If None, uses `max_force` value from actuator_manager.

    Returns:
        Boolean tensor indicating which environments should terminate
    """
    force = actuator_manager.get_dofs_control_force()
    if threshold is None:
        threshold = actuator_manager.get_dofs_max_force()
    return torch.any(torch.abs(force) > threshold, dim=-1)
```

## `dof_velocity_limit(env, actuator_manager, threshold, unit='rad')`

Terminate if any of the actuator_manager's joints moves faster than a speed limit.

Parameters:

| Name               | Type                    | Description                                                                                 | Default    |
| ------------------ | ----------------------- | ------------------------------------------------------------------------------------------- | ---------- |
| `env`              | `GenesisEnv`            | The Genesis environment                                                                     | *required* |
| `actuator_manager` | `ActuatorManager`       | Actuator manager for the controlled joints                                                  | *required* |
| `threshold`        | `float`                 | Speed limit in the units given by unit                                                      | *required* |
| `unit`             | `Literal['rpm', 'rad']` | The speed units - "rad" for radians per second (default) - "rpm" for revolutions per minute | `'rad'`    |

Returns:

| Type     | Description                                                   |
| -------- | ------------------------------------------------------------- |
| `Tensor` | Boolean tensor indicating which environments should terminate |

Source code in `genesis_forge/mdp/terminations.py`

```
def dof_velocity_limit(
    env: GenesisEnv,
    actuator_manager: ActuatorManager,
    threshold: float,
    unit: Literal["rpm", "rad"] = "rad",
) -> torch.Tensor:
    """
    Terminate if any of the actuator_manager's joints moves faster than a speed limit.

    Args:
        env: The Genesis environment
        actuator_manager: Actuator manager for the controlled joints
        threshold: Speed limit in the units given by `unit`
        unit: The speed units
              - `"rad"` for radians per second (default)
              - `"rpm"` for revolutions per minute

    Returns:
        Boolean tensor indicating which environments should terminate
    """
    assert unit in ["rad", "rpm"], f"Unknown velocity unit '{unit}'. Use 'rad' or 'rpm'."

    if unit == "rpm":
        threshold = threshold * (2 * math.pi / 60)
    vel = actuator_manager.get_dofs_velocity()
    return torch.any(torch.abs(vel) > threshold, dim=-1)
```

## `has_contact(env, contact_manager, threshold=1.0, min_contacts=1)`

One or more links in the contact manager are in contact with something.

Parameters:

| Name              | Type             | Description                                                            | Default    |
| ----------------- | ---------------- | ---------------------------------------------------------------------- | ---------- |
| `env`             | `GenesisEnv`     | The Genesis environment containing the robot                           | *required* |
| `contact_manager` | `ContactManager` | The contact manager to check for contact                               | *required* |
| `threshold`       | `float`          | The force threshold, per contact, for contact detection (default: 1.0) | `1.0`      |
| `min_contacts`    | `int`            | The minimum number of contacts required to terminate (default: 1)      | `1`        |

Returns:

| Type     | Description                                |
| -------- | ------------------------------------------ |
| `Tensor` | True for each environment that has contact |

Source code in `genesis_forge/mdp/terminations.py`

```
def has_contact(
    env: GenesisEnv, contact_manager: ContactManager, threshold: float=1.0, min_contacts: int=1
) -> torch.Tensor:
    """
    One or more links in the contact manager are in contact with something.

    Args:
        env: The Genesis environment containing the robot
        contact_manager: The contact manager to check for contact
        threshold: The force threshold, per contact, for contact detection (default: 1.0)
        min_contacts: The minimum number of contacts required to terminate (default: 1)

    Returns:
        True for each environment that has contact
    """
    has_contact = contact_manager.contacts[:, :].norm(dim=-1) > threshold
    return has_contact.sum(dim=1) >= min_contacts
```

## `is_upsidedown(env, threshold=0.5, entity_attr='robot', entity_manager=None, grace_steps=0)`

Terminate when the robot is belly-up (inverted).

Uses projected gravity in the body frame: upright is approximately [0, 0, -1], belly-up is approximately [0, 0, +1]. Side-lying poses keep z below threshold.

Parameters:

| Name             | Type            | Description                                               | Default    |
| ---------------- | --------------- | --------------------------------------------------------- | ---------- |
| `env`            | `GenesisEnv`    | The Genesis environment                                   | *required* |
| `threshold`      | `float`         | Terminate when projected_gravity[:, 2] exceeds this value | `0.5`      |
| `entity_manager` | `EntityManager` | The entity manager for the robot                          | `None`     |
| `entity_attr`    | `str`           | Entity attribute if entity_manager is not provided        | `'robot'`  |
| `grace_steps`    | `int`           | Steps at episode start to ignore this check               | `0`        |

Source code in `genesis_forge/mdp/terminations.py`

```
def is_upsidedown(
    env: GenesisEnv,
    threshold: float = 0.5,
    entity_attr: str = "robot",
    entity_manager: EntityManager = None,
    grace_steps: int = 0,
) -> torch.Tensor:
    """
    Terminate when the robot is belly-up (inverted).

    Uses projected gravity in the body frame: upright is approximately [0, 0, -1],
    belly-up is approximately [0, 0, +1]. Side-lying poses keep z below threshold.

    Args:
        env: The Genesis environment
        threshold: Terminate when projected_gravity[:, 2] exceeds this value
        entity_manager: The entity manager for the robot
        entity_attr: Entity attribute if entity_manager is not provided
        grace_steps: Steps at episode start to ignore this check
    """
    in_grace_period = env.episode_length <= grace_steps

    if entity_manager is not None:
        projected_gravity = entity_manager.get_projected_gravity()
    else:
        entity = getattr(env, entity_attr)
        projected_gravity = entity_projected_gravity(entity)

    return (~in_grace_period) & (projected_gravity[:, 2] > threshold)
```

## `out_of_bounds(env, terrain_manager, subterrain=None, border_margin=0.5, entity_attr='robot')`

Terminate if the entity's base position is outside of the terrain.

Parameters:

| Name              | Type             | Description                                                                                                             | Default                                    |
| ----------------- | ---------------- | ----------------------------------------------------------------------------------------------------------------------- | ------------------------------------------ |
| `env`             | `GenesisEnv`     | The Genesis environment containing the robot                                                                            | *required*                                 |
| `terrain_manager` | `TerrainManager` | The terrain manager to check for out of bounds                                                                          | *required*                                 |
| `subterrain`      | \`str            | None\`                                                                                                                  | The subterrain to keep the robot inside of |
| `border_margin`   | `float`          | The margin (in meters) to add to the terrain bounds This terminates the episode before the robot falls off the terrain. | `0.5`                                      |
| `entity_attr`     | `str`            | The attribute name of the entity in the environment. This isn't necessary if entity_manager is provided.                | `'robot'`                                  |

Source code in `genesis_forge/mdp/terminations.py`

```
def out_of_bounds(
    env: GenesisEnv,
    terrain_manager: TerrainManager,
    subterrain: str | None = None,
    border_margin: float = 0.5,
    entity_attr: str = "robot",
) -> torch.Tensor:
    """
    Terminate if the entity's base position is outside of the terrain.

    Args:
        env: The Genesis environment containing the robot
        terrain_manager: The terrain manager to check for out of bounds
        subterrain: The subterrain to keep the robot inside of
        border_margin: The margin (in meters) to add to the terrain bounds
                       This terminates the episode before the robot falls off the terrain.
        entity_attr: The attribute name of the entity in the environment.
                        This isn't necessary if `entity_manager` is provided.
    """
    # Get the entity's base position
    entity = getattr(env, entity_attr)
    position = entity.get_pos()

    # Get terrain bounds
    (x_min, x_max, y_min, y_max) = terrain_manager.get_bounds(subterrain)
    x_min_bound, x_max_bound = x_min + border_margin, x_max - border_margin
    y_min_bound, y_max_bound = y_min + border_margin, y_max - border_margin

    # Check bounds
    x_pos, y_pos = position[:, 0], position[:, 1]
    return (
        (x_pos < x_min_bound)
        | (x_pos > x_max_bound)
        | (y_pos < y_min_bound)
        | (y_pos > y_max_bound)
    )
```

## `timeout(env)`

Terminate the environment if the episode length exceeds the maximum episode length.

Source code in `genesis_forge/mdp/terminations.py`

```
def timeout(env: GenesisEnv) -> torch.Tensor:
    """
    Terminate the environment if the episode length exceeds the maximum episode length.
    """
    if env.max_episode_length is None:
        return torch.zeros(env.num_envs, dtype=torch.bool)
    return env.episode_length > env.max_episode_length
```
# Wrappers

# Wrapper Overview

Environment wrappers are used to wrap your environment class to add functionality for training. This is similar to how wrappers work in Gymnasium and StableBaselines3.

For example, you might want to use the rsl_rl training framework and regularly capture videos during training:

```
    # Your environment
    env = Go2SimpleEnv(num_envs=10, headless=True)

    # Record a video of every 5th episode
    env = VideoWrapper(
        env,
        video_length_sec=12,
        out_dir=os.path.join(log_dir, "videos"),
        episode_trigger=lambda episode_id: episode_id % 5 == 0,
    )

    # Make the environment compatible with rsl_rl
    env = RslRlWrapper(env)

    # Build the environment
    env.build()
    env.reset()

    # Train
    runner = OnPolicyRunner(env, training_cfg, log_dir, device=gs.device)
    runner.learn(num_learning_iterations=max_iterations)
    env.close()
```

# RSL-RL

Bases: `Wrapper`

A wrapper that makes your genesis forge environment compatible with the rsl_rl training framework.

Warning

This should be the last wrapper, as the change in the step and get_observations methods might break other wrappers.

What it does

- Combines the terminated and truncated tensors into a single tensor (i.e. `terminated | truncated`).
- Add the truncated tensor to the extras dictionary as "time_outs".
- Returns observations and extras from the `get_observations` method.

Parameters:

| Name  | Type         | Description              | Default                                                                              |
| ----- | ------------ | ------------------------ | ------------------------------------------------------------------------------------ |
| `env` | `GenesisEnv` | The environment to wrap. | *required*                                                                           |
| `cfg` | \`dict       | object\`                 | The configuration for the wrapper that will be passed to the neptune or wandb logger |

Source code in `genesis_forge/wrappers/rsl_rl.py`

```
def __init__(self, env: GenesisEnv, cfg: dict | object = {}):
    super().__init__(env)

    self.rsl3 = False
    self.cfg = cfg
    try:
        major_version = int(metadata.version("rsl-rl-lib").split(".")[0])
        if major_version >= 3:
            self.rsl3 = True
    except:
        pass
```

## `action_space`

The action space of the environment.

## `can_be_wrapped = False`

## `cfg = cfg`

## `device`

## `dt`

The time step of the environment.

## `env = env`

## `extras`

The extras/infos dictionary that should be returned by the step and reset functions. This dictionary will be cleared at the start of every step.

## `num_actions`

The number of actions for each environment.

## `num_envs`

The number of parallel environments.

## `num_observations`

The number of observations for each environment.

## `observation_space`

The observation space of the environment.

## `robot`

Get the environment robot.

## `rsl3 = False`

## `scene`

Get the environment scene.

## `unwrapped`

Returns the base environment of the wrapper.

This will be the bare :class:`ManagedEnvironment` or :class:`GenesisEnv` environment, underneath all layers of wrappers.

## `build()`

Builds the scene and other supporting components necessary for the training environment. This assumes that the scene has already been constructed and assigned to the .scene attribute.

Source code in `genesis_forge/wrappers/wrapper.py`

```
def build(self) -> None:
    """
    Builds the scene and other supporting components necessary for the training environment.
    This assumes that the scene has already been constructed and assigned to the <env>.scene attribute.
    """
    self.env.build()
```

## `close()`

Closes the wrapper and :attr:`env`.

Source code in `genesis_forge/wrappers/wrapper.py`

```
def close(self):
    """Closes the wrapper and :attr:`env`."""
    return self.env.close()
```

## `get_observations()`

Returns observations as well as an extras dictionary with the observations added to the `extras["observations"]["critic"]` key.

Source code in `genesis_forge/wrappers/rsl_rl.py`

```
def get_observations(self):
    """
    Returns observations as well as an extras dictionary with the observations added to the `extras["observations"]["critic"]` key.
    """
    obs = self.env.get_observations()

    # rsl_rl 3.0+ just returns the observations
    if self.rsl3:
        obs = self._format_obs_group(obs, self.env.extras)
        return obs

    # Earlier versions of rsl_rl adds critic observations to the extras dictionary
    extras = self._add_observations_to_extras(obs, self.env.extras)
    return obs, extras
```

## `reset()`

Converts observations into a TensorDict for rsl_rl 3.0+

Source code in `genesis_forge/wrappers/rsl_rl.py`

```
def reset(self):
    """
    Converts observations into a TensorDict for rsl_rl 3.0+
    """
    (obs, extras) = self.env.reset()
    obs = self._format_obs_group(obs, extras)
    return obs, extras
```

## `step(actions)`

Returns a single "dones" tensor, instead of the terminated and truncated tensors (via `terminated | truncated`). Add the truncated tensor to the extras dictionary as "time_outs".

Source code in `genesis_forge/wrappers/rsl_rl.py`

```
def step(
    self, actions: torch.Tensor
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, dict[str, Any]]:
    """
    Returns a single "dones" tensor, instead of the terminated and truncated tensors (via `terminated | truncated`).
    Add the truncated tensor to the extras dictionary as "time_outs".
    """
    (
        obs,
        rewards,
        terminated,
        truncated,
        extras,
    ) = super().step(actions)

    # Combine terminated and truncated
    dones = terminated | truncated

    # Add observations and timeouts to extras
    if extras is None:
        extras = {}
    extras = self._add_observations_to_extras(obs, extras)

    obs = self._format_obs_group(obs, extras)
    return obs, rewards, dones, extras
```

# SkrlEnv

Bases: `Wrapper`, `Wrapper`

A wrapper that makes your genesis forge environment compatible with the skrl training framework.

Initialize the logging wrapper with the function to use for data logging.

Source code in `genesis_forge/wrappers/wrapper.py`

```
def __init__(self, env: GenesisEnv):
    """Initialize the logging wrapper with the function to use for data logging."""
    assert (
        env.can_be_wrapped
    ), f"An environment wrapped with {self.__class__.__name__} cannot be wrapped"

    self.env = env
    if not isinstance(env, GenesisEnv) and not isinstance(env, Wrapper):
        raise ValueError(
            f"Expected env to be a `GenesisEnv` or `Wrapper` but got {type(env)}"
        )
```

## `action_space`

The action space of the environment.

## `can_be_wrapped = False`

## `dt`

The time step of the environment.

## `env = env`

## `extras`

The extras/infos dictionary that should be returned by the step and reset functions. This dictionary will be cleared at the start of every step.

## `num_actions`

The number of actions for each environment.

## `num_envs`

The number of parallel environments.

## `num_observations`

The number of observations for each environment.

## `observation_space`

The observation space of the environment.

## `robot`

Get the environment robot.

## `scene`

Get the environment scene.

## `unwrapped`

Returns the base environment of the wrapper.

This will be the bare :class:`ManagedEnvironment` or :class:`GenesisEnv` environment, underneath all layers of wrappers.

## `build()`

Build the environment

Source code in `genesis_forge/wrappers/skrl.py`

```
def build(self) -> None:
    """Build the environment"""
    self._env.build()
```

## `close()`

Close the environment

Source code in `genesis_forge/wrappers/skrl.py`

```
def close(self) -> None:
    """Close the environment"""
    return self._env.close()
```

## `get_observations()`

Uses the :meth:`get_observations` of the :attr:`env` that can be overwritten to change the returned data.

Source code in `genesis_forge/wrappers/wrapper.py`

```
def get_observations(self) -> torch.Tensor:
    """Uses the :meth:`get_observations` of the :attr:`env` that can be overwritten to change the returned data."""
    return self.env.get_observations()
```

## `render(*args, **kwargs)`

Not implemented for Genesis Forge environments.

Source code in `genesis_forge/wrappers/skrl.py`

```
def render(self, *args, **kwargs) -> Any:
    """
    Not implemented for Genesis Forge environments.
    """
    pass
```

## `reset()`

Reset the environment

Raises:

| Type                  | Description     |
| --------------------- | --------------- |
| `NotImplementedError` | Not implemented |

Returns:

| Name    | Type                 | Description                       |
| ------- | -------------------- | --------------------------------- |
| `tuple` | `Tuple[Tensor, Any]` | Observation (tensor), info (dict) |

Source code in `genesis_forge/wrappers/skrl.py`

```
def reset(self) -> Tuple[torch.Tensor, Any]:
    """Reset the environment

    Raises:
        NotImplementedError: Not implemented

    Returns: 
        tuple: Observation (tensor), info (dict)
    """
    return self._env.reset()
```

## `state()`

Get the environment state

Returns:

| Type     | Description          |
| -------- | -------------------- |
| `Tensor` | State (torch.Tensor) |

Source code in `genesis_forge/wrappers/skrl.py`

```
def state(self) -> torch.Tensor:
    """Get the environment state

    Returns: 
        State (torch.Tensor)
    """
    return self.env.state()
```

## `step(actions)`

Perform a step in the environment

Parameters:

| Name      | Type     | Description            | Default    |
| --------- | -------- | ---------------------- | ---------- |
| `actions` | `Tensor` | The actions to perform | *required* |

Returns:

| Type                                         | Description                                                                                                               |
| -------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------- |
| `Tuple[Tensor, Tensor, Tensor, Tensor, Any]` | tuple of tensors and a dict: Observation (tensor) , reward (tensor), terminated (tensor), truncated (tensor), info (dict) |

Source code in `genesis_forge/wrappers/skrl.py`

```
def step(
    self, actions: torch.Tensor
) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor, Any]:
    """Perform a step in the environment

    Args:
        actions: The actions to perform

    Returns:
        tuple of tensors and a dict: 
            Observation (tensor) , reward (tensor), terminated (tensor), truncated (tensor), info (dict)
    """
    obs, rewards, terminations, timeouts, extras = self._env.step(actions)

    # Expand rewards, terminations and timeouts to the shape (num_envs, 1)
    rewards = rewards.unsqueeze(1)
    terminations = terminations.unsqueeze(1)
    timeouts = timeouts.unsqueeze(1)

    return obs, rewards, terminations, timeouts, extras
```

# Video

Bases: `Wrapper`

Automatically record videos during training at a regular step or episode intervals.

Based on the RecordVideo wrapper from Gymnasium: https://gymnasium.farama.org/main/api/wrappers/misc_wrappers/#gymnasium.wrappers.RecordVideo

Recordings will be made from a dedicated camera, which you need to add to your environment (see the example below).

To control how frequently recordings are made specify **either** `episode_trigger` **or** `step_trigger` (not both). They should be functions returning a boolean that indicates whether a recording should be started at the current episode or step, respectively. If neither :attr:`episode_trigger` nor `step_trigger` is passed, a default `episode_trigger` will be used, which records at the episode indices 0, 1, 8, 27, ..., :math:`k^3`, ..., 729, 1000, 2000, 3000,.

Parameters:

| Name               | Type                      | Description                                                                                 | Default                                                                                                                                                                   |
| ------------------ | ------------------------- | ------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `env`              | `GenesisEnv`              | GenesisEnv                                                                                  | *required*                                                                                                                                                                |
| `camera_attr`      | `str`                     | The attribute of the base environment that contains the camera to use for recording.        | `'camera'`                                                                                                                                                                |
| `episode_trigger`  | \`Callable\[[int], bool\] | None\`                                                                                      | Function that accepts an episode count integer and returns True if a recording should be started at this episode                                                          |
| `step_trigger`     | \`Callable\[[int], bool\] | None\`                                                                                      | Function that accepts a step count integer and returns True if a recording should be started at this step                                                                 |
| `video_length_sec` | `int`                     | Length of each video, in seconds.                                                           | `8`                                                                                                                                                                       |
| `out_dir`          | `str`                     | Directory to save the videos to.                                                            | `'./videos'`                                                                                                                                                              |
| `fps`              | `int`                     | Frames per second for the video.                                                            | `60`                                                                                                                                                                      |
| `env_idx`          | `int`                     | If triggering on episode, this is the index of the environment to be counting episodes for. | `0`                                                                                                                                                                       |
| `filename`         | \`str                     | None\`                                                                                      | The filename for the video. If None, the video will automatically be named for the current step. If defined, each video will overwrite the previous video with this name. |

Example::

```
class MyEnv(GenesisEnv):
    __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        # Construct the scene
        self.scene = gs.Scene()

        # Assign a camera to the `camera` env attribute
        self.camera = scene.add_camera(pos=(-2.5, -1.5, 1.0))


def train():
    env = MyEnv()
    env = VideoWrapper(
        env,
        camera_attr="camera",
        out_dir="./videos"
    )
    env.build()
    ...training code...
```

Record every 1500 steps::

```
env = MyEnv()
env = VideoWrapper(
    env,
    camera_attr="camera",
    out_dir="./videos",
    step_trigger=lambda step: step % 1500 == 0
)
```

Source code in `genesis_forge/wrappers/video.py`

```
def __init__(
    self,
    env: GenesisEnv,
    camera_attr: str = "camera",
    video_length_sec: int = 8,
    episode_trigger: Callable[[int], bool] | None = None,
    step_trigger: Callable[[int], bool] | None = None,
    out_dir: str = "./videos",
    fps: int = 60,
    env_idx: int = 0,
    filename: str | None = None,
    logging: bool = True,
):
    super().__init__(env)
    self._is_recording: bool = False
    self._logging: bool = logging
    self._current_step: int = 0
    self._current_episode: int = 0
    self._recording_start_step: int = 0
    self._recording_stop_step: int = 0

    self._cam: Camera = None
    self._camera_attr = camera_attr
    self._out_dir = out_dir
    self._filename = filename
    self._video_length_steps = math.ceil(video_length_sec / self.dt)
    self._steps_per_frame = max(1, round(1.0 / fps / self.dt)) # max prevents division by zero
    self._actual_fps = round(1.0 / self.dt / self._steps_per_frame)
    self._env_idx = env_idx

    if episode_trigger is None and step_trigger is None:
        episode_trigger = capped_cubic_episode_trigger

    trigger_count = sum(x is not None for x in [episode_trigger, step_trigger])
    assert trigger_count == 1, "Must specify only one trigger"

    self.episode_trigger = episode_trigger
    self.step_trigger = step_trigger

    os.makedirs(self._out_dir, exist_ok=True)
```

## `action_space`

The action space of the environment.

## `can_be_wrapped = True`

## `dt`

The time step of the environment.

## `env = env`

## `episode_trigger = episode_trigger`

## `extras`

The extras/infos dictionary that should be returned by the step and reset functions. This dictionary will be cleared at the start of every step.

## `num_actions`

The number of actions for each environment.

## `num_envs`

The number of parallel environments.

## `num_observations`

The number of observations for each environment.

## `observation_space`

The observation space of the environment.

## `robot`

Get the environment robot.

## `scene`

Get the environment scene.

## `step_trigger = step_trigger`

## `unwrapped`

Returns the base environment of the wrapper.

This will be the bare :class:`ManagedEnvironment` or :class:`GenesisEnv` environment, underneath all layers of wrappers.

## `video_length_steps`

The number of steps that will be recorded for each video.

## `build()`

Load the camera from the environment.

Source code in `genesis_forge/wrappers/video.py`

```
def build(self) -> None:
    """Load the camera from the environment."""
    super().build()
    self._cam = self.unwrapped.__getattribute__(self._camera_attr)
    assert (
        self._cam is not None
    ), f"Camera not found at attribute: {self.unwrapped.__class__.__name__}.{self._camera_attr}"
```

## `close()`

Finish recording on close

Source code in `genesis_forge/wrappers/video.py`

```
def close(self):
    """Finish recording on close"""
    if self._is_recording:
        self.finish_recording()
    super().close()
```

## `finish_recording()`

Stop recording and save the video.

Source code in `genesis_forge/wrappers/video.py`

```
def finish_recording(self):
    """
    Stop recording and save the video.
    """
    if not self._is_recording or self._cam is None:
        return

    # Save recording
    filename = self._filename or f"{self._recording_start_step}.mp4"
    filepath = os.path.join(self._out_dir, filename)
    if self._logging:
        print(f"Saving recording to {filepath}")
    self._cam.stop_recording(filepath, fps=self._actual_fps)

    # Reset recording state
    self._is_recording = False
    self._recording_stop_step = 0
```

## `get_observations()`

Uses the :meth:`get_observations` of the :attr:`env` that can be overwritten to change the returned data.

Source code in `genesis_forge/wrappers/wrapper.py`

```
def get_observations(self) -> torch.Tensor:
    """Uses the :meth:`get_observations` of the :attr:`env` that can be overwritten to change the returned data."""
    return self.env.get_observations()
```

## `reset(env_ids=None)`

Uses the :meth:`reset` of the :attr:`env` that can be overwritten to change the returned data.

Source code in `genesis_forge/wrappers/wrapper.py`

```
def reset(
    self,
    env_ids: Sequence[int] = None,
) -> tuple[torch.Tensor, dict[str, Any]]:
    """Uses the :meth:`reset` of the :attr:`env` that can be overwritten to change the returned data."""
    return self.env.reset(env_ids)
```

## `start_recording()`

Start recording a video.

Source code in `genesis_forge/wrappers/video.py`

```
def start_recording(self):
    """Start recording a video."""
    self._is_recording = True
    self._recording_start_step = self._current_step
    self._recording_stop_step = self._current_step + self._video_length_steps
    self._cam.start_recording()
```

## `step(actions)`

Record a video image at each step.

Source code in `genesis_forge/wrappers/video.py`

```
def step(
    self, actions: torch.Tensor
) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor, dict[str, Any]]:
    """Record a video image at each step."""
    (
        observations,
        rewards,
        terminateds,
        truncateds,
        extras,
    ) = super().step(actions)

    self._check_recording_trigger()
    if self._is_recording:
        if self._current_step % self._steps_per_frame == 0:
            self._cam.render()

        # Stop recording if the recording stop step is reached
        if self._recording_stop_step <= self._current_step:
            self.finish_recording()

    # Increment episode count if the watched environment has terminated or truncated
    if self._is_done(terminateds) or self._is_done(truncateds):
        self._current_episode += 1
    self._current_step += 1

    return (
        observations,
        rewards,
        terminateds,
        truncateds,
        extras,
    )
```
# Other

# Gamepad

Wrapper around SDL2 ControllerEventLoop that provides a polling-based interface.

This maintains axis and button state that can be queried at any time, similar to the old HID-based Gamepad API.

Example::

```
>>> gamepad = Gamepad()
>>> gamepad.axis(0)  # Get left stick X axis
0.0
>>> gamepad.buttons()  # Get list of pressed buttons
['a', 'b']
```

Initialize the SDL2 gamepad wrapper.

Parameters:

| Name    | Type   | Description                      | Default |
| ------- | ------ | -------------------------------- | ------- |
| `debug` | `bool` | If True, print debug information | `False` |

Source code in `genesis_forge/gamepads/gamepad.py`

```
def __init__(self, debug: bool = False):
    """
    Initialize the SDL2 gamepad wrapper.

    Args:
        debug: If True, print debug information
    """
    self._debug = debug
    self.is_running = True
    self._lock = threading.Lock()

    # State storage
    self._axis_values: list[float] = [0.0] * 6
    self._button_set: set[str] = set()
    self._dpad: str | None = None

    # Start the SDL2 event loop in a background thread
    self._event_loop = ControllerEventLoop(
        handle_key=self._handle_key,
        alive=threading.Event(),
    )
    self._event_loop.alive.set()
    self._read_thread = threading.Thread(target=self._event_loop.run, daemon=True)
    self._read_thread.start()

    if self._debug:
        print("SDL2 gamepad wrapper initialized")
```

## `dpad`

Get the current D-pad direction (e.g., 'up', 'down', 'left', 'right').

## `is_running = True`

## `axis(index)`

Get the value of an axis.

Parameters:

| Name    | Type  | Description          | Default    |
| ------- | ----- | -------------------- | ---------- |
| `index` | `int` | The axis index (0-5) | *required* |

Returns:

| Type    | Description                                                             |
| ------- | ----------------------------------------------------------------------- |
| `float` | The axis value in range [-1.0, 1.0] for sticks, [0.0, 1.0] for triggers |

Source code in `genesis_forge/gamepads/gamepad.py`

```
def axis(self, index: int) -> float:
    """
    Get the value of an axis.

    Args:
        index: The axis index (0-5)

    Returns:
        The axis value in range [-1.0, 1.0] for sticks, [0.0, 1.0] for triggers
    """
    with self._lock:
        if index >= len(self._axis_values):
            return 0.0
        return self._axis_values[index]
```

## `buttons()`

Get the list of currently pressed buttons.

Returns:

| Type        | Description                                                |
| ----------- | ---------------------------------------------------------- |
| `list[str]` | List of button names (lowercase, e.g., 'a', 'b', 'x', 'y') |

Source code in `genesis_forge/gamepads/gamepad.py`

```
def buttons(self) -> list[str]:
    """
    Get the list of currently pressed buttons.

    Returns:
        List of button names (lowercase, e.g., 'a', 'b', 'x', 'y')
    """
    with self._lock:
        return list(self._button_set)
```

## `stop()`

Stop reading gamepad input.

Source code in `genesis_forge/gamepads/gamepad.py`

```
def stop(self) -> None:
    """Stop reading gamepad input."""
    self.is_running = False
    self._event_loop.stop()
```

## Low-level SDL2 API

For advanced users who want direct access to the SDL2 event loop:

Minimal SDL2 controller event loop for genesis-forge.

This class runs in a background thread and processes SDL2 controller events, converting them to Key objects and passing them to a callback function.

Parameters:

| Name         | Type                    | Description                                                                      | Default                                                                           |
| ------------ | ----------------------- | -------------------------------------------------------------------------------- | --------------------------------------------------------------------------------- |
| `handle_key` | `Callable[[Key], None]` | Callback function that receives Key objects for each controller event            | *required*                                                                        |
| `alive`      | \`Event                 | None\`                                                                           | Optional threading.Event to control the event loop. When cleared, the loop exits. |
| `timeout`    | `int`                   | Milliseconds to wait for events before checking the alive flag (default: 2000ms) | `2000`                                                                            |

Example::

```
def on_key(key: Key):
    print(f"Key event: {key}")

loop = ControllerEventLoop(handle_key=on_key)
thread = threading.Thread(target=loop.run, daemon=True)
thread.start()

# Later, to stop:
loop.stop()
```

Source code in `genesis_forge/gamepads/sdl2.py`

```
def __init__(
    self,
    handle_key: Callable[[Key], None],
    alive: threading.Event | None = None,
    timeout: int = 2000,
) -> None:
    self.handle_key = handle_key
    self.alive = alive or threading.Event()
    if not self.alive.is_set():
        self.alive.set()
    self.timeout = timeout
    self._controllers: dict[int, sdl2.SDL_GameController] = {}
```

## `alive = alive or threading.Event()`

## `handle_key = handle_key`

## `timeout = timeout`

## `run()`

Source code in `genesis_forge/gamepads/sdl2.py`

```
def run(self) -> None:
    if sdl2.SDL_WasInit(sdl2.SDL_INIT_GAMECONTROLLER) == 0:
        sdl2.SDL_InitSubSystem(sdl2.SDL_INIT_GAMECONTROLLER)

    for idx in range(sdl2.SDL_NumJoysticks()):
        if sdl2.SDL_IsGameController(idx):
            self._open_controller(idx)

    event = sdl2.SDL_Event()
    try:
        while self.alive.is_set():
            if sdl2.SDL_WaitEventTimeout(event, self.timeout):
                self._handle_event(event)
    finally:
        for instance_id in list(self._controllers.keys()):
            self._close_controller(instance_id)
        sdl2.SDL_QuitSubSystem(sdl2.SDL_INIT_GAMECONTROLLER)
```

## `stop()`

Source code in `genesis_forge/gamepads/sdl2.py`

```
def stop(self) -> None:
    self.alive.clear()
```

Gamepad input abstraction shared across backends.

## `AXIS = 'Axis'`

## `BUTTON = 'Button'`

## `HAT = 'Hat'`

## `index`

## `keytype`

## `name = None`

## `value = None`

Source code in `genesis_forge/gamepads/sdl2.py`

```
def controller_key_from_event(
    event: sdl2.SDL_Event, deadzone: float = 0.05
) -> Key | None:
    etype = event.type

    if etype == sdl2.SDL_CONTROLLERAXISMOTION:
        axis = event.caxis.axis
        raw = event.caxis.value
        limits = (0.0, 1.0) if _is_trigger_axis(axis) else (-1.0, 1.0)
        value = _rescale(raw, -32768, 32767, limits[0], limits[1])
        if not _is_trigger_axis(axis) and abs(value) < deadzone:
            value = 0.0
        return Key(Key.AXIS, axis, AXIS_NAMES.get(axis), value)

    if etype in (sdl2.SDL_CONTROLLERBUTTONDOWN, sdl2.SDL_CONTROLLERBUTTONUP):
        button = event.cbutton.button
        val = 1 if etype == sdl2.SDL_CONTROLLERBUTTONDOWN else 0
        return Key(Key.BUTTON, button, BUTTON_NAMES.get(button), val)

    return None
```
