original_docstring: Register all envs and components.
docstring: Register all envs and components.
docstring_tokens: Register all envs and components .
code: def register(registry_name: str,
               assert_override: bool = True,
               assert_format: bool = True,
               optional: bool = True):
    """Register all envs and components."""
    global ENVS, REGISTRIES, ENVS_BY_TRACKS
  
    assert (optional or registry_name not in REGISTRIES
           ), f'non-optional register() conflicts: {registry_name}'
    if registry_name in REGISTRIES:
      return REGISTRIES[registry_name]
  
    lib = importlib.import_module(f'{ROOT_PATH}.{registry_name}')
    metadata = {}
    if assert_format:  # assert submission format
      for attr in ('AUTHORS', 'CONTACTS', 'AFFILIATIONS', 'DESCRIPTIONS'):
        assert hasattr(lib, attr), f'{attr} must be defined as a tuple of strs'
        values = getattr(lib, attr)
        assert isinstance(
            values, tuple), f'{attr}={values} must be defined as a tuple of strs'
        assert all(isinstance(v, str) for v in values
                  ), f'{attr}={values} must be defined as a tuple of strs'
        metadata[attr] = values
  
    envs = lib.ENVS or {}
    components = lib.COMPONENTS or {}
    envs = {registry.get_env_name(registry_name, k): v for k, v in envs.items()}
    components = {
        registry.get_comp_name(registry_name, k): v
        for k, v in components.items()
    }
    task_envs = []
  
    # register environments
    for env_name, env_info in sorted(envs.items()):
      if assert_override:
        assert not exists(env_name), f'{list_env()} contains {env_name}'
      assert 'module' in env_info, env_info
      env_module = env_info['module']
      if isinstance(env_module, str):  # e.g. jump_cheetah:JumpCheetah
        env_lib_path, env_mod_name = env_module.split(':', 1)
        env_lib = importlib.import_module(
            f'{ROOT_PATH}.{registry_name}.envs.{env_lib_path}')
        env_module = getattr(env_lib, env_mod_name)
      if composer_envs.is_env_desc(env_module):
        # register a ComposerEnv
        composer_envs.register_env(env_name, env_module, override=True)
      else:
        # register a standard Env
        ENVS[env_name] = env_module
      tracks = env_info.get('tracks', ['rl'])
      for track in tracks:
        assert track in OPEN_ENDED_TRACKS, f'{track} not in {OPEN_ENDED_TRACKS}'
        ENVS_BY_TRACKS['open_ended'][track] += (env_name,)
      if 'mimax' in tracks:
        # (MI-Max only) register obs_indices
        for indices_type, indices in env_info.get('obs_indices', {}).items():
          obs_indices.register_indices(env_name, indices_type, indices)
  
    # register components
    for comp_name, comp_info in sorted(components.items()):
      if assert_override:
        assert not composer_components.exists(
            comp_name
        ), f'{composer_components.list_components()} contains {comp_name}'
      comp_module = comp_info['module']
      comp_lib = composer_components.register_component(
          comp_name,
          load_path=f'{ROOT_PATH}.{registry_name}.components.{comp_module}',
          override=True)
      component_params = get_func_kwargs(comp_lib.get_specs)[0]
      for track in comp_info.get('tracks', []):
        assert (track
                in GOAL_ORIENTED_TRACKS), f'{track} not in {GOAL_ORIENTED_TRACKS}'
        track_env_name = tasks.get_task_env_name(track, comp_name)
        if assert_override:
          assert not exists(
              track_env_name), f'{list_env()} contains {track_env_name}'
        track_env_module = functools.partial(tasks.TASKS[track], comp_name,
                                             **component_params)
        # register a ComposerEnv
        composer_envs.register_env(
            track_env_name, track_env_module, override=True)
        task_envs += [track_env_name]
        ENVS_BY_TRACKS['goal_oriented'][track] += (track_env_name,)
        COMPONENTS_BY_TRACKS[track] += (comp_name,)
  
    assert envs or task_envs, 'no envs registered'
    REGISTRIES[registry_name] = (sorted(envs), sorted(components),
                                 sorted(task_envs), metadata)
    return REGISTRIES[registry_name]
code_tokens: def register ( registry_name : str , assert_override : bool = True , assert_format : bool = True , optional : bool = True ) : global ENVS , REGISTRIES , ENVS_BY_TRACKS assert ( optional or registry_name not in REGISTRIES ) , f'non-optional register() conflicts: {registry_name}' if registry_name in REGISTRIES : return REGISTRIES [ registry_name ] lib = importlib . import_module ( f'{ROOT_PATH}.{registry_name}' ) metadata = { } if assert_format : for attr in ( 'AUTHORS' , 'CONTACTS' , 'AFFILIATIONS' , 'DESCRIPTIONS' ) : assert hasattr ( lib , attr ) , f'{attr} must be defined as a tuple of strs' values = getattr ( lib , attr ) assert isinstance ( values , tuple ) , f'{attr}={values} must be defined as a tuple of strs' assert all ( isinstance ( v , str ) for v in values ) , f'{attr}={values} must be defined as a tuple of strs' metadata [ attr ] = values envs = lib . ENVS or { } components = lib . COMPONENTS or { } envs = { registry . get_env_name ( registry_name , k ) : v for k , v in envs . items ( ) } components = { registry . get_comp_name ( registry_name , k ) : v for k , v in components . items ( ) } task_envs = [ ] for env_name , env_info in sorted ( envs . items ( ) ) : if assert_override : assert not exists ( env_name ) , f'{list_env()} contains {env_name}' assert 'module' in env_info , env_info env_module = env_info [ 'module' ] if isinstance ( env_module , str ) : env_lib_path , env_mod_name = env_module . split ( ':' , 1 ) env_lib = importlib . import_module ( f'{ROOT_PATH}.{registry_name}.envs.{env_lib_path}' ) env_module = getattr ( env_lib , env_mod_name ) if composer_envs . is_env_desc ( env_module ) : composer_envs . register_env ( env_name , env_module , override = True ) else : ENVS [ env_name ] = env_module tracks = env_info . get ( 'tracks' , [ 'rl' ] ) for track in tracks : assert track in OPEN_ENDED_TRACKS , f'{track} not in {OPEN_ENDED_TRACKS}' ENVS_BY_TRACKS [ 'open_ended' ] [ track ] += ( env_name , ) if 'mimax' in tracks : for indices_type , indices in env_info . get ( 'obs_indices' , { } ) . items ( ) : obs_indices . register_indices ( env_name , indices_type , indices ) for comp_name , comp_info in sorted ( components . items ( ) ) : if assert_override : assert not composer_components . exists ( comp_name ) , f'{composer_components.list_components()} contains {comp_name}' comp_module = comp_info [ 'module' ] comp_lib = composer_components . register_component ( comp_name , load_path = f'{ROOT_PATH}.{registry_name}.components.{comp_module}' , override = True ) component_params = get_func_kwargs ( comp_lib . get_specs ) [ 0 ] for track in comp_info . get ( 'tracks' , [ ] ) : assert ( track in GOAL_ORIENTED_TRACKS ) , f'{track} not in {GOAL_ORIENTED_TRACKS}' track_env_name = tasks . get_task_env_name ( track , comp_name ) if assert_override : assert not exists ( track_env_name ) , f'{list_env()} contains {track_env_name}' track_env_module = functools . partial ( tasks . TASKS [ track ] , comp_name , ** component_params ) composer_envs . register_env ( track_env_name , track_env_module , override = True ) task_envs += [ track_env_name ] ENVS_BY_TRACKS [ 'goal_oriented' ] [ track ] += ( track_env_name , ) COMPONENTS_BY_TRACKS [ track ] += ( comp_name , ) assert envs or task_envs , 'no envs registered' REGISTRIES [ registry_name ] = ( sorted ( envs ) , sorted ( components ) , sorted ( task_envs ) , metadata ) return REGISTRIES [ registry_name ]
short_docstring: Register all envs and components.
short_docstring_tokens: Register all envs and components .
comment: """Register all envs and components."""$SEP$# assert submission format$SEP$# register environments$SEP$# e.g. jump_cheetah:JumpCheetah$SEP$# register a ComposerEnv$SEP$# register a standard Env$SEP$# (MI-Max only) register obs_indices$SEP$# register components$SEP$# register a ComposerEnv