Source code for minestudio.simulator.callbacks.commands

'''
Date: 2024-11-11 19:31:53
LastEditors: caishaofei-mus1 1744260356@qq.com
LastEditTime: 2025-01-16 18:09:12
FilePath: /ROCKET-2/var/nfs-shared/shaofei/nfs-workspace/MineStudio/minestudio/simulator/callbacks/commands.py
'''
import os
import yaml
from typing import Dict, List, Tuple, Union, Sequence, Mapping, Any, Optional, Literal
from minestudio.simulator.callbacks.callback import MinecraftCallback
from minestudio.utils.register import Registers

[docs] @Registers.simulator_callback.register class CommandsCallback(MinecraftCallback): """ Executes a list of Minecraft commands at specific lifecycle events. This callback is typically used to run setup commands after an environment reset. """
[docs] def create_from_conf(source: Union[str, Dict]): """Creates a CommandsCallback instance from a configuration. The configuration can be a path to a YAML file or a dictionary. It should contain a list of command strings under the key 'custom_init_commands' or 'commands'. :param source: Configuration source (file path or dict). :type source: Union[str, Dict] :returns: A CommandsCallback instance or None if no commands are specified. :rtype: Optional[CommandsCallback] """ data = MinecraftCallback.load_data_from_conf(source) available_keys = ['custom_init_commands', 'commands'] for key in available_keys: if key in data: commands = data[key] return CommandsCallback(commands) return None
def __init__(self, commands: List[str]): """Initializes the CommandsCallback with a list of commands. :param commands: A list of command strings to execute. :type commands: List[str] """ super().__init__() self.commands = commands
[docs] def after_reset(self, sim, obs: Dict, info: Dict) -> Tuple[Dict, Dict]: """Executes the configured commands after the environment resets. Each command is run sequentially, and the observation and info dictionaries are updated with the results from each command execution. :param sim: The simulator instance. :param obs: The current observation dictionary. :param info: The current info dictionary. :returns: Updated observation and info dictionaries. :rtype: Tuple[Dict, Dict] """ for command in self.commands: _obs, reward, done, info = sim.env.execute_cmd(command) obs.update(_obs) info.update(info) obs, info = sim._wrap_obs_info(obs, info) return obs, info
def __repr__(self) -> str: """Returns a string representation of the CommandsCallback. :returns: String representation of the instance. :rtype: str """ return f"CommandsCallback(commands={self.commands})"