Creating a Modular Environment

Creating a Modular Environment#

This tutorial demonstrates how to create sophisticated robotic environments using EmbodiChain’s modular architecture. You’ll learn how to use the advanced envs.EmbodiedEnv class with configuration-driven setup, event managers, observation managers, and randomization systems.

The Code#

The tutorial corresponds to the modular_env.py script in the scripts/tutorials/gym directory.

Code for modular_env.py

# ----------------------------------------------------------------------------
# Copyright (c) 2021-2026 DexForce Technology Co., Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ----------------------------------------------------------------------------

from __future__ import annotations

import argparse
import random
from copy import deepcopy

from embodichain.cli.sim import (
    add_sim_args_to_parser,
    add_seed_arg_to_parser,
    resolve_seed,
)


def build_parser() -> argparse.ArgumentParser:
    """Build CLI options without initializing simulation resources."""
    parser = argparse.ArgumentParser()
    add_sim_args_to_parser(parser)
    add_seed_arg_to_parser(parser, scope="task environment")
    return parser


if __name__ == "__main__":
    # Parse before importing optional simulation/planning dependencies.
    _cli_args = build_parser().parse_args()


import torch

from typing import List, Dict, Any

import embodichain.lab.gym.envs.managers.randomization as rand
import embodichain.lab.gym.envs.managers.events as events
import embodichain.lab.gym.envs.managers.observations as obs

from embodichain.lab.gym.envs.managers import (
    EventCfg,
    SceneEntityCfg,
    ObservationCfg,
)
from embodichain.lab.gym.envs import EmbodiedEnv, EmbodiedEnvCfg
from embodichain.lab.gym.utils.registration import register_env
from embodichain_robots import DexforceW1Cfg
from embodichain.lab.sim.sensors import StereoCameraCfg, SensorCfg
from embodichain.lab.sim.shapes import MeshCfg
from embodichain.lab.sim.cfg import (
    RenderCfg,
    physics_cfg_for_backend,
    LightCfg,
    ArticulationCfg,
    RobotCfg,
    RigidObjectCfg,
    RigidBodyPhysicsCfg,
)
from embodichain.data import get_data_path
from embodichain.utils import configclass
from embodichain.utils.file import get_all_files_in_directory
from embodichain.lab.gym.envs.managers.event_manager import _derive_functor_seed


class _ReplacePreselectedFork(events.replace_assets_from_group):
    """Consume the first reset draw without replacing an already selected fork."""

    def __call__(
        self,
        env: EmbodiedEnv,
        env_ids: torch.Tensor | None,
        entity_cfg: SceneEntityCfg,
        folder_path: str,
    ) -> None:
        initial_path = env._initial_fork_path
        env._initial_fork_path = None
        if initial_path is not None:
            # Preview the actual event stream: reset(seed=...) may have changed
            # it since construction. Leave fallback sampling to the base term.
            preview = random.Random()
            preview.setstate(random.getstate())
            if (
                preview.choice(self._asset_group_path) == initial_path
                and self.asset_cfg.shape.fpath == initial_path
            ):
                random.choice(self._asset_group_path)
                # As with a real replacement, keep the template separate from
                # the live object modified by later mass randomization terms.
                self.asset_cfg = deepcopy(self.asset_cfg)
                return
        super().__call__(env, env_ids, entity_cfg, folder_path)


@configclass
class ExampleEventCfg:

    replace_obj: EventCfg = EventCfg(
        func=_ReplacePreselectedFork,
        mode="reset",
        params={
            "entity_cfg": SceneEntityCfg(
                uid="fork",
            ),
            "folder_path": get_data_path("TableWare/tableware/fork/"),
        },
    )

    randomize_fork_mass: EventCfg = EventCfg(
        func=rand.randomize_rigid_object_mass,
        mode="reset",
        params={
            "entity_cfg": SceneEntityCfg(
                uid="fork",
            ),
            "mass_range": (0.1, 2.0),
        },
    )

    randomize_table_mat: EventCfg = EventCfg(
        func=rand.randomize_visual_material,
        mode="interval",
        interval_step=25,
        params={
            "entity_cfg": SceneEntityCfg(
                uid="table",
            ),
            "random_texture_prob": 0.5,
            "texture_path": get_data_path("CocoBackground/coco"),
            "base_color_range": [[0.2, 0.2, 0.2], [1.0, 1.0, 1.0]],
        },
    )


@configclass
class ObsCfg:

    obj_pose: ObservationCfg = ObservationCfg(
        func=obs.get_rigid_object_pose,
        mode="add",
        name="fork_pose",
        params={"entity_cfg": SceneEntityCfg(uid="fork")},
    )


@configclass
class ExampleCfg(EmbodiedEnvCfg):

    # Define the robot configuration using DexforceW1Cfg
    robot: RobotCfg = DexforceW1Cfg.from_dict(
        {
            "uid": "dexforce_w1",
            "version": "v021",
            "init_pos": [0.0, 0, 0.0],
        }
    )

    # Define the sensor configuration using StereoCameraCfg
    sensor: List[SensorCfg] = [
        StereoCameraCfg(
            uid="eye_in_head",
            width=960,
            height=540,
            enable_mask=True,
            enable_depth=True,
            left_to_right_pos=(0.06, 0, 0),
            intrinsics=(450, 450, 480, 270),
            intrinsics_right=(450, 450, 480, 270),
            extrinsics=StereoCameraCfg.ExtrinsicsCfg(
                parent="eyes",
            ),
        )
    ]

    background: List[RigidObjectCfg] = [
        RigidObjectCfg(
            uid="table",
            shape=MeshCfg(
                fpath=get_data_path("CircleTableSimple/circle_table_simple.ply"),
                compute_uv=True,
            ),
            attrs=RigidBodyPhysicsCfg.from_dict(
                {
                    "mass_props": {"mass": 10.0},
                    "material_props": {
                        "static_friction": 0.95,
                        "dynamic_friction": 0.85,
                        "restitution": 0.01,
                    },
                }
            ),
            body_type="kinematic",
            init_pos=(0.80, 0, 0.8),
            init_rot=(0, 90, 0),
        ),
    ]

    rigid_object: List[RigidObjectCfg] = [
        RigidObjectCfg(
            uid="fork",
            shape=MeshCfg(
                fpath=get_data_path("TableWare/tableware/fork/standard_fork_scale.ply"),
            ),
            body_scale=(0.75, 0.75, 1.0),
            init_pos=(0.8, 0, 1.0),
        ),
    ]

    articulation_cfg: List[ArticulationCfg] = [
        ArticulationCfg(
            uid="drawer",
            fpath="SlidingBoxDrawer/SlidingBoxDrawer.urdf",
            init_pos=(0.5, 0.0, 0.85),
        )
    ]

    events = ExampleEventCfg()

    observations = ObsCfg()


def _preselect_initial_fork(
    cfg: EmbodiedEnvCfg,
) -> tuple[EmbodiedEnvCfg, str | None]:
    """Prepare this tutorial's first reset asset before scene materialization.

    Use the same named event stream and sorted asset list as EventManager.
    Keep unseeded runs and customized replacement terms on their normal path.
    """
    term = getattr(cfg.events, "replace_obj", None)
    if (
        cfg.seed is None
        or cfg.seed < 0
        or term is None
        or term.func is not _ReplacePreselectedFork
        or term.mode != "reset"
    ):
        return cfg, None
    folder_path = term.params["folder_path"]
    if not folder_path.endswith("/"):
        return cfg, None
    paths = sorted(get_all_files_in_directory(get_data_path(folder_path)))
    if not paths:
        raise ValueError(f"No fork assets found in {folder_path!r}.")
    stream_seed = _derive_functor_seed(cfg.seed, "call", "reset", "replace_obj", 0)
    path = random.Random(stream_seed).choice(paths)
    prepared = deepcopy(cfg)
    uid = term.params["entity_cfg"].uid
    asset_cfg = next(asset for asset in prepared.rigid_object if asset.uid == uid)
    asset_cfg.shape.fpath = path
    return prepared, path


@register_env("ModularEnv-v1", max_episode_steps=100, override=True)
class ModularEnv(EmbodiedEnv):
    """
    An example of a modular environment that inherits from EmbodiedEnv
    and uses custom event and observation managers.
    """

    def __init__(self, cfg: EmbodiedEnvCfg, **kwargs):
        cfg, self._initial_fork_path = _preselect_initial_fork(cfg)
        super().__init__(cfg, **kwargs)


if __name__ == "__main__":
    import gymnasium as gym
    import argparse

    from embodichain.lab.sim import SimulationManagerCfg
    from embodichain.cli.sim import (
        add_sim_args_to_parser,
        add_seed_arg_to_parser,
        resolve_seed,
    )
    from embodichain.lab.visualization import visualization_cfg_from_args

    parser = build_parser()
    args = _cli_args
    seed = resolve_seed(args.seed)
    print(f"[INFO]: Environment seed: {seed}", flush=True)

    env_cfg = ExampleCfg(
        seed=seed,
        sim_cfg=SimulationManagerCfg(
            render_cfg=RenderCfg(renderer=args.renderer),
            headless=args.headless,
            device=args.device,
            num_envs=args.num_envs,
            physics_cfg=physics_cfg_for_backend(args.physics),
            visualization=visualization_cfg_from_args(args),
        ),
        num_envs=args.num_envs,
    )

    # Create the Gym environment
    env = gym.make("ModularEnv-v1", cfg=env_cfg)

    for i in range(5):
        obs, info = env.reset()

        for i in range(100):
            action = torch.zeros(env.action_space.shape, dtype=torch.float32)
            obs, reward, done, truncated, info = env.step(action)

The Code Explained#

This tutorial showcases EmbodiChain’s most powerful environment creation approach using the envs.EmbodiedEnv class. Unlike the basic environment tutorial, this approach uses declarative configuration classes and manager systems for maximum flexibility and reusability.

Event Configuration#

Events define automated behaviors that occur during simulation. There are three types of supported modes:

startup: triggers once when the environment is initialized
reset: triggers every time the environment is reset
interval: triggers at fixed step intervals during simulation

The ExampleEventCfg demonstrates three types of events:

if __name__ == "__main__":
    # Parse before importing optional simulation/planning dependencies.
    _cli_args = build_parser().parse_args()


import torch

from typing import List, Dict, Any

import embodichain.lab.gym.envs.managers.randomization as rand
import embodichain.lab.gym.envs.managers.events as events
import embodichain.lab.gym.envs.managers.observations as obs

from embodichain.lab.gym.envs.managers import (
    EventCfg,
    SceneEntityCfg,
    ObservationCfg,
)
from embodichain.lab.gym.envs import EmbodiedEnv, EmbodiedEnvCfg
from embodichain.lab.gym.utils.registration import register_env
from embodichain_robots import DexforceW1Cfg
from embodichain.lab.sim.sensors import StereoCameraCfg, SensorCfg
from embodichain.lab.sim.shapes import MeshCfg
from embodichain.lab.sim.cfg import (
    RenderCfg,
    physics_cfg_for_backend,
    LightCfg,
    ArticulationCfg,
    RobotCfg,
    RigidObjectCfg,
    RigidBodyPhysicsCfg,
)
from embodichain.data import get_data_path
from embodichain.utils import configclass
from embodichain.utils.file import get_all_files_in_directory
from embodichain.lab.gym.envs.managers.event_manager import _derive_functor_seed


class _ReplacePreselectedFork(events.replace_assets_from_group):

Asset Replacement Event

The replace_obj event demonstrates dynamic asset swapping:

Function: envs.managers.events.replace_assets_from_group()
Mode: "reset" - triggers at environment reset
Purpose: Randomly selects different fork models from a folder

Light Randomization Event

The randomize_light event creates dynamic lighting conditions:

Function: envs.managers.randomization.rendering.randomize_light()
Mode: "interval" - triggers every 5 steps
Parameters: Randomizes position, color, and intensity within specified ranges

Material Randomization Event

The randomize_table_mat event varies visual appearance:

Function: envs.managers.randomization.rendering.randomize_visual_material()
Mode: "interval" - triggers every 10 steps
Features: Random textures from COCO dataset and base color variations

For more randomization events, please refer to Event Functors.

Observation Configuration#

The default observation from envs.EmbodiedEnv includes: - robot: robot proprioceptive data (joint positions, velocities, efforts) - sensor: all available sensor data (images, depth, segmentation, etc.)

However, users always need to define some custom observation for specified learning tasks. To handle this, the observation manager system allows users to declaratively specify additional observations.

    def __call__(
        self,
        env: EmbodiedEnv,
        env_ids: torch.Tensor | None,
        entity_cfg: SceneEntityCfg,
        folder_path: str,
    ) -> None:
        initial_path = env._initial_fork_path
        env._initial_fork_path = None

This configuration:

Function: envs.managers.observations.get_rigid_object_pose()
Mode: "add" - appends data to observation dictionary
Name: Custom key for the observation data
Target: Tracks the fork object’s pose in the scene

For details documentation, see envs.managers.cfg.ObservationCfg.

Environment Configuration#

The main environment configuration inherits from envs.EmbodiedEnvCfg and defines all scene components:

Robot Configuration

    robot: RobotCfg = DexforceW1Cfg.from_dict(
        {
            "uid": "dexforce_w1",
            "version": "v021",
            "init_pos": [0.0, 0, 0.0],
        }
    )

Uses the pre-configured DexforceW1Cfg with customizations:

Version: Specific robot variant (v021)
Position: Initial placement in the scene

Sensor Configuration

@configclass
class ExampleEventCfg:

    replace_obj: EventCfg = EventCfg(
        func=_ReplacePreselectedFork,
        mode="reset",
        params={
            "entity_cfg": SceneEntityCfg(
                uid="fork",
            ),
            "folder_path": get_data_path("TableWare/tableware/fork/"),
        },
    )

Configures a stereo camera system using StereoCameraCfg:

Resolution: 960x540 pixels for realistic visual input
Features: Depth sensing and segmentation masks enabled
Stereo Setup: 6cm baseline between left and right cameras
Mounting: Attached to robot’s “eyes” frame

Lighting Configuration

        func=rand.randomize_rigid_object_mass,
        mode="reset",
        params={
            "entity_cfg": SceneEntityCfg(
                uid="fork",
            ),
            "mass_range": (0.1, 2.0),
        },
    )

    randomize_table_mat: EventCfg = EventCfg(

Defines scene illumination with configurable lights:

Types: Supports "point", "sun", "direction", "spot", "rect", and "mesh".
Global lights: "sun" and "direction" are global scene lights (single instance, infinite distance).
Properties: Configurable color, intensity, position, and direction.
UID: Named reference for event system manipulation.

Rigid Objects

        mode="interval",
        interval_step=25,
        params={
            "entity_cfg": SceneEntityCfg(
                uid="table",
            ),
            "random_texture_prob": 0.5,
            "texture_path": get_data_path("CocoBackground/coco"),
            "base_color_range": [[0.2, 0.2, 0.2], [1.0, 1.0, 1.0]],
        },
    )


@configclass
class ObsCfg:

    obj_pose: ObservationCfg = ObservationCfg(
        func=obs.get_rigid_object_pose,
        mode="add",
        name="fork_pose",
        params={"entity_cfg": SceneEntityCfg(uid="fork")},
    )


@configclass
class ExampleCfg(EmbodiedEnvCfg):

Multiple objects demonstrate different physics properties:

Table Configuration:

Shape: Custom PLY mesh with UV mapping
Physics: Kinematic body (movable but not affected by forces)
Material: Friction and restitution properties for realistic contact

Fork Configuration:

Shape: Detailed mesh from asset library
Scale: Proportionally scaled for scene consistency
Physics: Dynamic body affected by gravity and collisions

Articulated Objects

    # Define the robot configuration using DexforceW1Cfg
    robot: RobotCfg = DexforceW1Cfg.from_dict(
        {
            "uid": "dexforce_w1",
            "version": "v021",
            "init_pos": [0.0, 0, 0.0],
        }
    )

    # Define the sensor configuration using StereoCameraCfg
    sensor: List[SensorCfg] = [

Demonstrates complex mechanisms with moving parts:

URDF: Sliding drawer with joints and constraints
Positioning: Placed on table surface for interaction

Environment Implementation#

The actual environment class is remarkably simple due to the configuration-driven approach:

@register_env("ModularEnv-v1", max_episode_steps=100, override=True)
class ModularEnv(EmbodiedEnv):
    """
    An example of a modular environment that inherits from EmbodiedEnv
    and uses custom event and observation managers.
    """

    def __init__(self, cfg: EmbodiedEnvCfg, **kwargs):
        cfg, self._initial_fork_path = _preselect_initial_fork(cfg)
        super().__init__(cfg, **kwargs)

The envs.EmbodiedEnv base class automatically:

Loads all configured scene components
Sets up observation and action spaces
Initializes event and observation managers
Handles environment lifecycle (reset, step, etc.)

The Code Execution#

To run the modular environment:

cd /path/to/embodichain
python scripts/tutorials/gym/modular_env.py

The script demonstrates the complete workflow:

Configuration: Creates an instance of ExampleCfg
Registration: Uses the registered environment ID
Execution: Runs episodes with zero actions to observe automatic behaviors

Manager System Benefits#

The manager-based architecture provides several key advantages:

Event Managers

Modularity: Reusable event functions across environments
Timing Control: Flexible scheduling (reset, interval, condition-based)
Parameter Binding: Type-safe configuration with validation
Extensibility: Easy to add custom event behaviors

Observation Managers

Flexible Data: Any simulation data can become an observation
Processing Pipeline: Built-in normalization and transformation
Dynamic Composition: Runtime observation space modification
Performance: Efficient data collection and GPU acceleration

Key Features Demonstrated#

This tutorial showcases the most advanced features of EmbodiChain environments:

Configuration-Driven Design: Declarative environment specification
Manager Systems: Modular event and observation handling
Asset Management: Dynamic loading and randomization
Sensor Integration: Realistic camera systems with stereo vision
Physics Simulation: Complex articulated and rigid body dynamics
Visual Randomization: Automated domain randomization
Extensible Architecture: Easy customization and extension points

This tutorial demonstrates the full power of EmbodiChain’s modular environment system, providing the foundation for creating sophisticated robotic learning scenarios.

Tip

Using an AI coding agent? These skills can help you build on this tutorial:

/add-task-env — Scaffold a new task environment with the correct file structure, @register_env decorator, base class methods, __init__.py update, and test stub.
/add-functor — Add observation, reward, event, or randomization functors with the correct signature and module placement.

Next Steps#

Configure and Run an Embodied Task Program — Compose a modular environment with a declarative Task Program, embodiment, scene binding, and execution policy.
Expert Data Expansion — Record expert demonstrations from modular environments.