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.	None
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]] | None	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.	None
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
delay_step	int | tuple[int, int]	Steps to delay actions by, to emulate bus latency. This is either a fixed step value or a min/max range to create random delays from. See ActionDelayBuffer.	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 = ".*",
    limit: tuple[float, float] | dict[str, tuple[float, float]] | None = None,
    soft_limit_scale_factor: float = 1.0,
    delay_step: int | tuple[int, int] = 0,
):
    super().__init__(
        env,
        actuator_manager=actuator_manager,
        actuator_joints=actuator_joints,
        delay_step=delay_step,
    )
    self._limit_cfg = ensure_dof_pattern(limit if limit is not None else {})
    self._soft_limit_scale_factor = soft_limit_scale_factor

action_space property

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

actions property

The processed actions for for the current step.

actuator_dof_filter property

An index filter for the actuator DOF buffer values.

actuator_manager property

Get the actuator manager.

default_dofs_pos property

Return the default DOF positions.

delay property

The action delay buffer, which holds each environment's current delay in delay.delay_steps.

deploy_type = 'position_within_limits' class-attribute instance-attribute

dofs property

Get a dictionary of the DOF names and their indices

dofs_idx property

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

enabled = True instance-attribute

env = env instance-attribute

joint_action_index property

One action index per joint, in dofs order.

None when every joint has its own action. Otherwise action_groups has joints sharing one, and this says which.

last_actions property

The processed actions for for the previous step.

num_actions property

The number of values the policy provides for this manager on each step. This is one per controlled DOF, unless action_groups ties several DOFs to one action.

num_dofs property

The number of DOFs this manager drives. The same as num_actions unless action_groups is used.

raw_actions property

The raw actions the policy emitted for this manager this step, before any processing or delay.

type = type instance-attribute

actions_to_dof_targets(actions)

Turn this manager's slice of the policy output into per-DOF targets.

Takes (num_envs, num_actions): fans the actions out to the DOFs they drive, then processes them. Override process_actions rather than this.

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

def actions_to_dof_targets(self, actions: torch.Tensor) -> torch.Tensor:
    """Turn this manager's slice of the policy output into per-DOF targets.

    Takes `(num_envs, num_actions)`: fans the actions out to the DOFs they
    drive, then processes them. Override `process_actions` rather than this.
    """
    return self.process_actions(self._actions_for_dofs(actions))

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 target actions for the environments.

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

def get_actions(self) -> torch.Tensor:
    """
    Get the current target actions for the environments.
    """
    if self._actions is None:
        return torch.zeros((self.env.num_envs, self.num_dofs), device=gs.device)
    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, :], strict=True
        )
    }

get_deployment_config()

Export this manager's process: pre-clip to [-1, 1], then map into limits.

Overrides :class:PositionActionManager deliberately. This manager inherits the parent's build() -- so _scale_values, _offset_values and _clip_values all exist -- but its process_actions ignores them in favour of _scale/_offset and applies no post-clip. Exporting the parent's parameters would describe processing this manager never performs.

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

def get_deployment_config(self) -> DeploymentActionConfig:
    """Export this manager's process: pre-clip to [-1, 1], then map into limits.

    Overrides :class:`PositionActionManager` deliberately. This manager
    inherits the parent's ``build()`` -- so ``_scale_values``, ``_offset_values``
    and ``_clip_values`` all exist -- but its ``process_actions`` ignores them
    in favour of ``_scale``/``_offset`` and applies no post-clip. Exporting the
    parent's parameters would describe processing this manager never performs.
    """
    return DeploymentActionConfig(
        deploy_type=self.deploy_type,
        config={
            "raw_action_clip": [-1.0, 1.0],
            "scale": self.per_joint_deployment_values(self._scale, "scale"),
            "offset": self.per_joint_deployment_values(self._offset, "offset"),
        },
        joint_action_index=self.joint_action_index,
    )

get_dofs_control_force(clip_to_max_force=False)

A wrapper for RigidEntity.get_dofs_control_force that returns the force output 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 output by the enabled DOFs.

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

def get_dofs_control_force(self, clip_to_max_force: bool = False) -> torch.Tensor:
    """
    A wrapper for `RigidEntity.get_dofs_control_force` that returns the force output 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 output by the enabled DOFs.
    """
    return self.actuator_manager.get_dofs_control_force(
        clip_to_max_force=clip_to_max_force, dofs_idx=self.dofs_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] | None	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 = 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
    )

per_joint_deployment_values(tensor, name)

Turn a per-joint parameter into a list of floats for the deployment bundle.

tensor holds one value per controlled joint, either flat as (num_dofs,) or with a row per environment as (num_envs, num_dofs).

A flat tensor is returned as-is. A per-environment tensor is returned as its shared row, and only if every environment holds the same values.

A bundle describes one robot, so when the environments differ this raises rather than guess which one is right. That only happens when something randomizes this particular parameter per environment.

Parameters:

Name	Type	Description	Default
tensor	Tensor	The parameter, such as a scale or offset.	required
name	str	How to refer to the parameter in error messages.	required

Returns:

Type	Description
list[float]	One float per joint, in dofs order.

Raises:

Type	Description
ValueError	The environments hold different values, or the tensor does not hold one value per joint.

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

def per_joint_deployment_values(
    self, tensor: torch.Tensor, name: str
) -> list[float]:
    """Turn a per-joint parameter into a list of floats for the deployment bundle.

    ``tensor`` holds one value per controlled joint, either flat as
    ``(num_dofs,)`` or with a row per environment as ``(num_envs, num_dofs)``.

    A flat tensor is returned as-is. A per-environment tensor is returned as
    its shared row, and only if every environment holds the same values.

    A bundle describes one robot, so when the environments differ this raises
    rather than guess which one is right. That only happens when something
    randomizes this particular parameter per environment.

    Args:
        tensor: The parameter, such as a scale or offset.
        name: How to refer to the parameter in error messages.

    Returns:
        One float per joint, in ``dofs`` order.

    Raises:
        ValueError: The environments hold different values, or the tensor does
            not hold one value per joint.
    """
    values = tensor.detach()
    num_dofs = self.num_dofs
    manager = type(self).__name__

    if values.ndim == 2 and values.shape[1] == num_dofs:
        if values.shape[0] > 1 and not bool((values == values[0]).all()):
            spread = float(
                (values.max(dim=0).values - values.min(dim=0).values).max()
            )
            raise ValueError(
                f"Cannot export '{name}' from action manager '{manager}': it is "
                f"not the same in every parallel environment (largest spread "
                f"{spread:g}), so there is no single value to put in the bundle. "
                f"Something is randomizing '{name}' per environment. Keep it "
                f"uniform across environments for the export run; randomizing "
                f"other things, such as friction or gains, does not affect it."
            )
        values = values[0]

    if values.ndim != 1 or values.shape[0] != num_dofs:
        raise ValueError(
            f"Cannot export '{name}' from action manager '{manager}': expected "
            f"one value per joint ({num_dofs}), found shape {tuple(tensor.shape)}."
        )

    return [float(item) for item in values.cpu().tolist()]

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=None)

Clear the action history of the reset environments, so the previous episode's actions are neither delivered by the delay buffer nor reported as the last actions.

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

def reset(self, envs_idx: torch.Tensor | None = None):
    """
    Clear the action history of the reset environments, so the previous episode's
    actions are neither delivered by the delay buffer nor reported as the last actions.
    """
    if envs_idx is None:
        envs_idx = self.env.all_envs_idx
    self._delay.reset(envs_idx)
    if self._actions is not None:
        self._raw_actions[envs_idx] = 0.0
        self._actions[envs_idx] = 0.0
        self._last_actions[envs_idx] = 0.0

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.
    """
    self._raw_actions = actions  # store raw actions before the delay buffer
    actions = self._delay.push(actions)

    if self._actions is None:
        shape = (actions.shape[0], self.num_dofs)
        self._actions = torch.zeros(shape, device=gs.device, dtype=actions.dtype)
        self._last_actions = torch.zeros(
            shape, device=gs.device, dtype=actions.dtype
        )
    self._last_actions[:] = self._actions[:]

    self._actions[:] = self.actions_to_dof_targets(actions)

    return self._actions