# Copyright (c) 2022-2025, The Isaac Lab Project Developers.# All rights reserved.## SPDX-License-Identifier: BSD-3-Clausefrom__future__importannotationsimportcopyimportinspectimportweakreffromabcimportABC,abstractmethodfromcollections.abcimportSequencefromtypingimportTYPE_CHECKING,Anyimportomni.logimportomni.timelineimportisaaclab.utils.stringasstring_utilsfromisaaclab.utilsimportstring_to_callablefrom.manager_term_cfgimportManagerTermBaseCfgfrom.scene_entity_cfgimportSceneEntityCfgifTYPE_CHECKING:fromisaaclab.envsimportManagerBasedEnv
[docs]classManagerTermBase(ABC):"""Base class for manager terms. Manager term implementations can be functions or classes. If the term is a class, it should inherit from this base class and implement the required methods. Each manager is implemented as a class that inherits from the :class:`ManagerBase` class. Each manager class should also have a corresponding configuration class that defines the configuration terms for the manager. Each term should the :class:`ManagerTermBaseCfg` class or its subclass. Example pseudo-code for creating a manager: .. code-block:: python from isaaclab.utils import configclass from isaaclab.utils.mdp import ManagerBase, ManagerTermBaseCfg @configclass class MyManagerCfg: my_term_1: ManagerTermBaseCfg = ManagerTermBaseCfg(...) my_term_2: ManagerTermBaseCfg = ManagerTermBaseCfg(...) my_term_3: ManagerTermBaseCfg = ManagerTermBaseCfg(...) # define manager instance my_manager = ManagerBase(cfg=ManagerCfg(), env=env) """
[docs]def__init__(self,cfg:ManagerTermBaseCfg,env:ManagerBasedEnv):"""Initialize the manager term. Args: cfg: The configuration object. env: The environment instance. """# store the inputsself.cfg=cfgself._env=env
""" Properties. """@propertydefnum_envs(self)->int:"""Number of environments."""returnself._env.num_envs@propertydefdevice(self)->str:"""Device on which to perform computations."""returnself._env.device""" Operations. """
[docs]defreset(self,env_ids:Sequence[int]|None=None)->None:"""Resets the manager term. Args: env_ids: The environment ids. Defaults to None, in which case all environments are considered. """pass
def__call__(self,*args)->Any:"""Returns the value of the term required by the manager. In case of a class implementation, this function is called by the manager to get the value of the term. The arguments passed to this function are the ones specified in the term configuration (see :attr:`ManagerTermBaseCfg.params`). .. attention:: To be consistent with memory-less implementation of terms with functions, it is recommended to ensure that the returned mutable quantities are cloned before returning them. For instance, if the term returns a tensor, it is recommended to ensure that the returned tensor is a clone of the original tensor. This prevents the manager from storing references to the tensors and altering the original tensors. Args: *args: Variable length argument list. Returns: The value of the term. """raiseNotImplementedError("The method '__call__' should be implemented by the subclass.")
[docs]classManagerBase(ABC):"""Base class for all managers."""
[docs]def__init__(self,cfg:object,env:ManagerBasedEnv):"""Initialize the manager. This function is responsible for parsing the configuration object and creating the terms. If the simulation is not playing, the scene entities are not resolved immediately. Instead, the resolution is deferred until the simulation starts. This is done to ensure that the scene entities are resolved even if the manager is created after the simulation has already started. Args: cfg: The configuration object. If None, the manager is initialized without any terms. env: The environment instance. """# store the inputsself.cfg=copy.deepcopy(cfg)self._env=env# if the simulation is not playing, we use callbacks to trigger the resolution of the scene# entities configuration. this is needed for cases where the manager is created after the# simulation, but before the simulation is playing.# FIXME: Once Isaac Sim supports storing this information as USD schema, we can remove this# callback and resolve the scene entities directly inside `_prepare_terms`.ifnotself._env.sim.is_playing():# note: Use weakref on all callbacks to ensure that this object can be deleted when its destructor# is called# The order is set to 20 to allow asset/sensor initialization to complete before the scene entities# are resolved. Those have the order 10.timeline_event_stream=omni.timeline.get_timeline_interface().get_timeline_event_stream()self._resolve_terms_handle=timeline_event_stream.create_subscription_to_pop_by_type(int(omni.timeline.TimelineEventType.PLAY),lambdaevent,obj=weakref.proxy(self):obj._resolve_terms_callback(event),order=20,)else:self._resolve_terms_handle=None# parse config to create terms informationifself.cfg:self._prepare_terms()
def__del__(self):"""Delete the manager."""ifself._resolve_terms_handle:self._resolve_terms_handle.unsubscribe()self._resolve_terms_handle=None""" Properties. """@propertydefnum_envs(self)->int:"""Number of environments."""returnself._env.num_envs@propertydefdevice(self)->str:"""Device on which to perform computations."""returnself._env.device@property@abstractmethoddefactive_terms(self)->list[str]|dict[str,list[str]]:"""Name of active terms."""raiseNotImplementedError""" Operations. """
[docs]defreset(self,env_ids:Sequence[int]|None=None)->dict[str,float]:"""Resets the manager and returns logging information for the current time-step. Args: env_ids: The environment ids for which to log data. Defaults None, which logs data for all environments. Returns: Dictionary containing the logging information. """return{}
[docs]deffind_terms(self,name_keys:str|Sequence[str])->list[str]:"""Find terms in the manager based on the names. This function searches the manager for terms based on the names. The names can be specified as regular expressions or a list of regular expressions. The search is performed on the active terms in the manager. Please check the :meth:`~isaaclab.utils.string_utils.resolve_matching_names` function for more information on the name matching. Args: name_keys: A regular expression or a list of regular expressions to match the term names. Returns: A list of term names that match the input keys. """# resolve search keysifisinstance(self.active_terms,dict):list_of_strings=[]fornamesinself.active_terms.values():list_of_strings.extend(names)else:list_of_strings=self.active_terms# return the matching namesreturnstring_utils.resolve_matching_names(name_keys,list_of_strings)[1]
[docs]defget_active_iterable_terms(self,env_idx:int)->Sequence[tuple[str,Sequence[float]]]:"""Returns the active terms as iterable sequence of tuples. The first element of the tuple is the name of the term and the second element is the raw value(s) of the term. Returns: The active terms. """raiseNotImplementedError
""" Implementation specific. """@abstractmethoddef_prepare_terms(self):"""Prepare terms information from the configuration object."""raiseNotImplementedError""" Internal callbacks. """def_resolve_terms_callback(self,event):"""Resolve configurations of terms once the simulation starts. Please check the :meth:`_process_term_cfg_at_play` method for more information. """# check if config is dict alreadyifisinstance(self.cfg,dict):cfg_items=self.cfg.items()else:cfg_items=self.cfg.__dict__.items()# iterate over all the termsforterm_name,term_cfgincfg_items:# check for non configifterm_cfgisNone:continue# process attributes at runtime# these properties are only resolvable once the simulation starts playingself._process_term_cfg_at_play(term_name,term_cfg)""" Internal functions. """def_resolve_common_term_cfg(self,term_name:str,term_cfg:ManagerTermBaseCfg,min_argc:int=1):"""Resolve common attributes of the term configuration. Usually, called by the :meth:`_prepare_terms` method to resolve common attributes of the term configuration. These include: * Resolving the term function and checking if it is callable. * Checking if the term function's arguments are matched by the parameters. * Resolving special attributes of the term configuration like ``asset_cfg``, ``sensor_cfg``, etc. * Initializing the term if it is a class. The last two steps are only possible once the simulation starts playing. By default, all term functions are expected to have at least one argument, which is the environment object. Some other managers may expect functions to take more arguments, for instance, the environment indices as the second argument. In such cases, the ``min_argc`` argument can be used to specify the minimum number of arguments required by the term function to be called correctly by the manager. Args: term_name: The name of the term. term_cfg: The term configuration. min_argc: The minimum number of arguments required by the term function to be called correctly by the manager. Raises: TypeError: If the term configuration is not of type :class:`ManagerTermBaseCfg`. ValueError: If the scene entity defined in the term configuration does not exist. AttributeError: If the term function is not callable. ValueError: If the term function's arguments are not matched by the parameters. """# check if the term is a valid term configifnotisinstance(term_cfg,ManagerTermBaseCfg):raiseTypeError(f"Configuration for the term '{term_name}' is not of type ManagerTermBaseCfg."f" Received: '{type(term_cfg)}'.")# get the corresponding function or functional classifisinstance(term_cfg.func,str):term_cfg.func=string_to_callable(term_cfg.func)# check if function is callableifnotcallable(term_cfg.func):raiseAttributeError(f"The term '{term_name}' is not callable. Received: {term_cfg.func}")# check if the term is a class of valid typeifinspect.isclass(term_cfg.func):ifnotissubclass(term_cfg.func,ManagerTermBase):raiseTypeError(f"Configuration for the term '{term_name}' is not of type ManagerTermBase."f" Received: '{type(term_cfg.func)}'.")func_static=term_cfg.func.__call__min_argc+=1# forward by 1 to account for 'self' argumentelse:func_static=term_cfg.func# check if function is callableifnotcallable(func_static):raiseAttributeError(f"The term '{term_name}' is not callable. Received: {term_cfg.func}")# check statically if the term's arguments are matched by paramsterm_params=list(term_cfg.params.keys())args=inspect.signature(func_static).parametersargs_with_defaults=[argforarginargsifargs[arg].defaultisnotinspect.Parameter.empty]args_without_defaults=[argforarginargsifargs[arg].defaultisinspect.Parameter.empty]args=args_without_defaults+args_with_defaults# ignore first two arguments for env and env_ids# Think: Check for cases when kwargs are set inside the function?iflen(args)>min_argc:ifset(args[min_argc:])!=set(term_params+args_with_defaults):raiseValueError(f"The term '{term_name}' expects mandatory parameters: {args_without_defaults[min_argc:]}"f" and optional parameters: {args_with_defaults}, but received: {term_params}.")# process attributes at runtime# these properties are only resolvable once the simulation starts playingifself._env.sim.is_playing():self._process_term_cfg_at_play(term_name,term_cfg)def_process_term_cfg_at_play(self,term_name:str,term_cfg:ManagerTermBaseCfg):"""Process the term configuration at runtime. This function is called when the simulation starts playing. It is used to process the term configuration at runtime. This includes: * Resolving the scene entity configuration for the term. * Initializing the term if it is a class. Since the above steps rely on PhysX to parse over the simulation scene, they are deferred until the simulation starts playing. Args: term_name: The name of the term. term_cfg: The term configuration. """forkey,valueinterm_cfg.params.items():ifisinstance(value,SceneEntityCfg):# load the entitytry:value.resolve(self._env.scene)exceptValueErrorase:raiseValueError(f"Error while parsing '{term_name}:{key}'. {e}")# log the entity for checking latermsg=f"[{term_cfg.__class__.__name__}:{term_name}] Found entity '{value.name}'."ifvalue.joint_idsisnotNone:msg+=f"\n\tJoint names: {value.joint_names} [{value.joint_ids}]"ifvalue.body_idsisnotNone:msg+=f"\n\tBody names: {value.body_names} [{value.body_ids}]"# print the informationomni.log.info(msg)# store the entityterm_cfg.params[key]=value# initialize the term if it is a classifinspect.isclass(term_cfg.func):omni.log.info(f"Initializing term '{term_name}' with class '{term_cfg.func.__name__}'.")term_cfg.func=term_cfg.func(cfg=term_cfg,env=self._env)