eval_utils

find_start_point(base_vel)

Find the first point where the base velocity is non-zero. This is used to skip the initial part of the dataset where the robot is not moving.

Source code in OmniGibson/omnigibson/eval/utils/eval_utils.py

def find_start_point(base_vel):
    """
    Find the first point where the base velocity is non-zero.
    This is used to skip the initial part of the dataset where the robot is not moving.
    """
    start_idx = np.where(np.linalg.norm(base_vel, axis=-1) > 1e-5)[0]
    if len(start_idx) == 0:
        return 0
    return min(start_idx[0], 500)  # Limit to the first 100 points to avoid long initial periods

flatten_obs_dict(obs, parent_key='')

Process the observation dictionary by recursively flattening the keys. so obs["robot_r1"]["camera"]["rgb"] will become obs["robot_r1:::rgb"].

Source code in OmniGibson/omnigibson/eval/utils/eval_utils.py

def flatten_obs_dict(obs: dict, parent_key: str = "") -> dict:
    """
    Process the observation dictionary by recursively flattening the keys.
    so obs["robot_r1"]["camera"]["rgb"] will become obs["robot_r1::camera:::rgb"].
    """
    processed_obs = {}
    for key, value in obs.items():
        new_key = f"{parent_key}::{key}" if parent_key else key
        if isinstance(value, dict):
            processed_obs.update(flatten_obs_dict(value, parent_key=new_key))
        else:
            processed_obs[new_key] = value
    return processed_obs

generate_basic_environment_config(task_name, task_cfg)

Generate a basic environment configuration

Parameters:

Name	Type	Description	Default
task_name	str	Name of the task	required
task_cfg		Dictionary of task config	required

Returns:

Type	Description
dict	Environment configuration

Source code in OmniGibson/omnigibson/eval/utils/eval_utils.py

def generate_basic_environment_config(task_name, task_cfg):
    """
    Generate a basic environment configuration

    Args:
        task_name (str): Name of the task
        task_cfg: Dictionary of task config

    Returns:
        dict: Environment configuration
    """
    cfg = {
        "env": {
            "action_frequency": 30,
            "rendering_frequency": 30,
            "physics_frequency": 120,
        },
        "scene": {
            "type": "InteractiveTraversableScene",
            "scene_model": task_cfg["scene_model"],
            "load_room_types": None,
            "load_room_instances": task_cfg.get("load_room_instances", None),
            "include_robots": False,
        },
        "task": {
            "type": "BehaviorTask",
            "activity_name": task_name,
            "activity_definition_id": 0,
            "activity_instance_id": 0,
            "online_object_sampling": False,
            "debug_object_sampling": False,
            "highlight_task_relevant_objects": False,
            "termination_config": {
                "max_steps": 5000,
            },
            "reward_config": {
                "r_potential": 1.0,
            },
            "include_obs": False,
        },
    }
    return cfg

get_robot_camera_names(robot_name, robot_eval_config)

Return flattened observation names for cameras configured by eval role.

Robot configs may include

eval: camera_sensor_names: head: robot_r1:zed_link:Camera:0

Values are robot sensor names, i.e. keys in robot.sensors. The flattened observation name adds the robot observation namespace.

Source code in OmniGibson/omnigibson/eval/utils/eval_utils.py

def get_robot_camera_names(robot_name: str, robot_eval_config: dict | None) -> dict[str, str]:
    """
    Return flattened observation names for cameras configured by eval role.

    Robot configs may include:
        eval:
          camera_sensor_names:
            head: robot_r1:zed_link:Camera:0

    Values are robot sensor names, i.e. keys in robot.sensors. The flattened
    observation name adds the robot observation namespace.
    """
    robot_eval_config = robot_eval_config or {}
    camera_sensor_names = robot_eval_config.get("camera_sensor_names", {})
    return {camera_id: f"{robot_name}::{sensor_name}" for camera_id, sensor_name in camera_sensor_names.items()}

seed_everything(seed, deterministic_torch=False)

Seed Python, NumPy, and Torch RNGs used by eval.

Setting PYTHONHASHSEED here helps child processes and documents the intended seed, but hash randomization for the current Python process is fixed at interpreter startup.

Source code in OmniGibson/omnigibson/eval/utils/eval_utils.py

def seed_everything(seed: int, deterministic_torch: bool = False) -> int:
    """
    Seed Python, NumPy, and Torch RNGs used by eval.

    Setting PYTHONHASHSEED here helps child processes and documents the intended seed, but
    hash randomization for the current Python process is fixed at interpreter startup.
    """
    seed = int(seed)
    os.environ["PYTHONHASHSEED"] = str(seed)
    random.seed(seed)
    np.random.seed(seed)
    th.manual_seed(seed)
    if th.cuda.is_available():
        th.cuda.manual_seed(seed)
        th.cuda.manual_seed_all(seed)
    try:
        import warp as wp

        wp.rand_init(seed)
    except ImportError:
        pass

    if hasattr(th.backends, "cudnn"):
        th.backends.cudnn.benchmark = False
        th.backends.cudnn.deterministic = True

    if deterministic_torch:
        os.environ.setdefault("CUBLAS_WORKSPACE_CONFIG", ":4096:8")
        try:
            th.use_deterministic_algorithms(True, warn_only=True)
        except TypeError:
            th.use_deterministic_algorithms(True)

    return seed