(
      self,
      walker: Callable[..., 'legacy_base.Walker'],
      arena: composer.Arena,
      ref_path: Text,
      ref_steps: Sequence[int],
      dataset: Union[Text, types.ClipCollection],
      termination_error_threshold: float = 0.3,
      prop_termination_error_threshold: float = 0.1,
      min_steps: int = 10,
      reward_type: Text = 'termination_reward',
      physics_timestep: float = DEFAULT_PHYSICS_TIMESTEP,
      always_init_at_clip_start: bool = False,
      proto_modifier: Optional[Any] = None,
      prop_factory: Optional[Any] = None,
      disable_props: bool = False,
      ghost_offset: Optional[Sequence[Union[int, float]]] = None,
      body_error_multiplier: Union[int, float] = 1.0,
      actuator_force_coeff: float = 0.015,
      enabled_reference_observables: Optional[Sequence[Text]] = None,
  )
Abstract task that uses reference data.

    Args:
      walker: Walker constructor to be used.
      arena: Arena to be used.
      ref_path: Path to the dataset containing reference poses.
      ref_steps: tuples of indices of reference observation. E.g if
        ref_steps=(1, 2, 3) the walker/reference observation at time t will
        contain information from t+1, t+2, t+3.
      dataset: A ClipCollection instance or a name of a dataset that appears as
        a key in DATASETS in datasets.py
      termination_error_threshold: Error threshold for episode terminations for
        hand body position and joint error only.
      prop_termination_error_threshold: Error threshold for episode terminations
        for prop position.
      min_steps: minimum number of steps within an episode. This argument
        determines the latest allowable starting point within a given reference
        trajectory.
      reward_type: type of reward to use, must be a string that appears as a key
        in the REWARD_FN dict in rewards.py.
      physics_timestep: Physics timestep to use for simulation.
      always_init_at_clip_start: only initialize epsidodes at the start of a
        reference trajectory.
      proto_modifier: Optional proto modifier to modify reference trajectories,
        e.g. adding a vertical offset.
      prop_factory: Optional function that takes the mocap proto and returns
        the corresponding props for the trajectory.
      disable_props: If prop_factory is specified but disable_props is True,
        no props will be created.
      ghost_offset: if not None, include a ghost rendering of the walker with
        the reference pose at the specified position offset.
      body_error_multiplier: A multiplier that is applied to the body error term
        when determining failure termination condition.
      actuator_force_coeff: A coefficient for the actuator force reward channel.
      enabled_reference_observables: Optional iterable of enabled observables.
        If not specified, a reasonable default set will be enabled.
Abstract task that uses reference data.
Abstract task that uses reference data .
def __init__(
      self,
      walker: Callable[..., 'legacy_base.Walker'],
      arena: composer.Arena,
      ref_path: Text,
      ref_steps: Sequence[int],
      dataset: Union[Text, types.ClipCollection],
      termination_error_threshold: float = 0.3,
      prop_termination_error_threshold: float = 0.1,
      min_steps: int = 10,
      reward_type: Text = 'termination_reward',
      physics_timestep: float = DEFAULT_PHYSICS_TIMESTEP,
      always_init_at_clip_start: bool = False,
      proto_modifier: Optional[Any] = None,
      prop_factory: Optional[Any] = None,
      disable_props: bool = False,
      ghost_offset: Optional[Sequence[Union[int, float]]] = None,
      body_error_multiplier: Union[int, float] = 1.0,
      actuator_force_coeff: float = 0.015,
      enabled_reference_observables: Optional[Sequence[Text]] = None,
  ):
    """Abstract task that uses reference data.

    Args:
      walker: Walker constructor to be used.
      arena: Arena to be used.
      ref_path: Path to the dataset containing reference poses.
      ref_steps: tuples of indices of reference observation. E.g if
        ref_steps=(1, 2, 3) the walker/reference observation at time t will
        contain information from t+1, t+2, t+3.
      dataset: A ClipCollection instance or a name of a dataset that appears as
        a key in DATASETS in datasets.py
      termination_error_threshold: Error threshold for episode terminations for
        hand body position and joint error only.
      prop_termination_error_threshold: Error threshold for episode terminations
        for prop position.
      min_steps: minimum number of steps within an episode. This argument
        determines the latest allowable starting point within a given reference
        trajectory.
      reward_type: type of reward to use, must be a string that appears as a key
        in the REWARD_FN dict in rewards.py.
      physics_timestep: Physics timestep to use for simulation.
      always_init_at_clip_start: only initialize epsidodes at the start of a
        reference trajectory.
      proto_modifier: Optional proto modifier to modify reference trajectories,
        e.g. adding a vertical offset.
      prop_factory: Optional function that takes the mocap proto and returns
        the corresponding props for the trajectory.
      disable_props: If prop_factory is specified but disable_props is True,
        no props will be created.
      ghost_offset: if not None, include a ghost rendering of the walker with
        the reference pose at the specified position offset.
      body_error_multiplier: A multiplier that is applied to the body error term
        when determining failure termination condition.
      actuator_force_coeff: A coefficient for the actuator force reward channel.
      enabled_reference_observables: Optional iterable of enabled observables.
        If not specified, a reasonable default set will be enabled.
    """
    self._ref_steps = np.sort(ref_steps)
    self._max_ref_step = self._ref_steps[-1]
    self._termination_error_threshold = termination_error_threshold
    self._prop_termination_error_threshold = prop_termination_error_threshold
    self._reward_fn = rewards.get_reward(reward_type)
    self._reward_keys = rewards.get_reward_channels(reward_type)
    self._min_steps = min_steps
    self._always_init_at_clip_start = always_init_at_clip_start
    self._ghost_offset = ghost_offset
    self._body_error_multiplier = body_error_multiplier
    self._actuator_force_coeff = actuator_force_coeff
    logging.info('Reward type %s', reward_type)

    if isinstance(dataset, Text):
      try:
        dataset = datasets.DATASETS[dataset]
      except KeyError:
        logging.error('Dataset %s not found in datasets.py', dataset)
        raise
    self._load_reference_data(
        ref_path=ref_path, proto_modifier=proto_modifier, dataset=dataset)

    self._get_possible_starts()

    logging.info('%d starting points found.', len(self._possible_starts))

    # load a dummy trajectory
    self._current_clip_index = 0
    self._current_clip = self._loader.get_trajectory(
        self._dataset.ids[0], zero_out_velocities=False)
    # Create the environment.
    self._arena = arena
    self._walker = utils.add_walker(walker, self._arena)
    self.set_timesteps(
        physics_timestep=physics_timestep,
        control_timestep=self._current_clip.dt)

    # Identify the desired body components.
    try:
      walker_bodies = self._walker.mocap_tracking_bodies
    except AttributeError:
      logging.info('Walker must implement mocap bodies for this task.')
      raise

    walker_bodies_names = [bdy.name for bdy in walker_bodies]
    self._body_idxs = np.array(
        [walker_bodies_names.index(bdy) for bdy in walker_bodies_names])

    self._prop_factory = prop_factory
    if disable_props:
      self._props = []
    else:
      self._props = self._current_clip.create_props(prop_factory=prop_factory)
    for prop in self._props:
      self._arena.add_free_entity(prop)

    # Create the observables.
    self._add_observables(enabled_reference_observables)

    # initialize counters etc.
    self._time_step = 0
    self._current_start_time = 0.0
    self._last_step = 0
    self._current_clip_index = 0
    self._reference_observations = dict()
    self._end_mocap = False
    self._should_truncate = False

    # Set up required dummy quantities for observations
    self._prop_prefixes = []

    self._disable_props = disable_props
    if not disable_props:
      if len(self._props) == 1:
        self._prop_prefixes += ['prop/']
      else:
        self._prop_prefixes += [f'prop_{i}/' for i in range(len(self._props))]
    self._clip_reference_features = self._current_clip.as_dict()
    self._strip_reference_prefix()

    self._walker_joints = self._clip_reference_features['joints'][0]
    self._walker_features = tree.map_structure(lambda x: x[0],
                                               self._clip_reference_features)
    self._walker_features_prev = tree.map_structure(
        lambda x: x[0], self._clip_reference_features)

    self._current_reference_features = dict()
    self._reference_ego_bodies_quats = collections.defaultdict(dict)
    # if requested add ghost body to visualize motion capture reference.
    if self._ghost_offset is not None:
      self._ghost = utils.add_walker(
          walker, self._arena, name='ghost', ghost=True)
      self._ghost.observables.disable_all()

      if disable_props:
        self._ghost_props = []
      else:
        self._ghost_props = self._current_clip.create_props(
            prop_factory=self._ghost_prop_factory)
        for prop in self._ghost_props:
          self._arena.add_free_entity(prop)
          prop.observables.disable_all()
    else:
      self._ghost_props = []

    # initialize reward channels
    self._reset_reward_channels()
def __init__ ( self , walker : Callable [ ... , 'legacy_base.Walker' ] , arena : composer . Arena , ref_path : Text , ref_steps : Sequence [ int ] , dataset : Union [ Text , types . ClipCollection ] , termination_error_threshold : float = 0.3 , prop_termination_error_threshold : float = 0.1 , min_steps : int = 10 , reward_type : Text = 'termination_reward' , physics_timestep : float = DEFAULT_PHYSICS_TIMESTEP , always_init_at_clip_start : bool = False , proto_modifier : Optional [ Any ] = None , prop_factory : Optional [ Any ] = None , disable_props : bool = False , ghost_offset : Optional [ Sequence [ Union [ int , float ] ] ] = None , body_error_multiplier : Union [ int , float ] = 1.0 , actuator_force_coeff : float = 0.015 , enabled_reference_observables : Optional [ Sequence [ Text ] ] = None , ) : self . _ref_steps = np . sort ( ref_steps ) self . _max_ref_step = self . _ref_steps [ - 1 ] self . _termination_error_threshold = termination_error_threshold self . _prop_termination_error_threshold = prop_termination_error_threshold self . _reward_fn = rewards . get_reward ( reward_type ) self . _reward_keys = rewards . get_reward_channels ( reward_type ) self . _min_steps = min_steps self . _always_init_at_clip_start = always_init_at_clip_start self . _ghost_offset = ghost_offset self . _body_error_multiplier = body_error_multiplier self . _actuator_force_coeff = actuator_force_coeff logging . info ( 'Reward type %s' , reward_type ) if isinstance ( dataset , Text ) : try : dataset = datasets . DATASETS [ dataset ] except KeyError : logging . error ( 'Dataset %s not found in datasets.py' , dataset ) raise self . _load_reference_data ( ref_path = ref_path , proto_modifier = proto_modifier , dataset = dataset ) self . _get_possible_starts ( ) logging . info ( '%d starting points found.' , len ( self . _possible_starts ) ) # load a dummy trajectory self . _current_clip_index = 0 self . _current_clip = self . _loader . get_trajectory ( self . _dataset . ids [ 0 ] , zero_out_velocities = False ) # Create the environment. self . _arena = arena self . _walker = utils . add_walker ( walker , self . _arena ) self . set_timesteps ( physics_timestep = physics_timestep , control_timestep = self . _current_clip . dt ) # Identify the desired body components. try : walker_bodies = self . _walker . mocap_tracking_bodies except AttributeError : logging . info ( 'Walker must implement mocap bodies for this task.' ) raise walker_bodies_names = [ bdy . name for bdy in walker_bodies ] self . _body_idxs = np . array ( [ walker_bodies_names . index ( bdy ) for bdy in walker_bodies_names ] ) self . _prop_factory = prop_factory if disable_props : self . _props = [ ] else : self . _props = self . _current_clip . create_props ( prop_factory = prop_factory ) for prop in self . _props : self . _arena . add_free_entity ( prop ) # Create the observables. self . _add_observables ( enabled_reference_observables ) # initialize counters etc. self . _time_step = 0 self . _current_start_time = 0.0 self . _last_step = 0 self . _current_clip_index = 0 self . _reference_observations = dict ( ) self . _end_mocap = False self . _should_truncate = False # Set up required dummy quantities for observations self . _prop_prefixes = [ ] self . _disable_props = disable_props if not disable_props : if len ( self . _props ) == 1 : self . _prop_prefixes += [ 'prop/' ] else : self . _prop_prefixes += [ f'prop_{i}/' for i in range ( len ( self . _props ) ) ] self . _clip_reference_features = self . _current_clip . as_dict ( ) self . _strip_reference_prefix ( ) self . _walker_joints = self . _clip_reference_features [ 'joints' ] [ 0 ] self . _walker_features = tree . map_structure ( lambda x : x [ 0 ] , self . _clip_reference_features ) self . _walker_features_prev = tree . map_structure ( lambda x : x [ 0 ] , self . _clip_reference_features ) self . _current_reference_features = dict ( ) self . _reference_ego_bodies_quats = collections . defaultdict ( dict ) # if requested add ghost body to visualize motion capture reference. if self . _ghost_offset is not None : self . _ghost = utils . add_walker ( walker , self . _arena , name = 'ghost' , ghost = True ) self . _ghost . observables . disable_all ( ) if disable_props : self . _ghost_props = [ ] else : self . _ghost_props = self . _current_clip . create_props ( prop_factory = self . _ghost_prop_factory ) for prop in self . _ghost_props : self . _arena . add_free_entity ( prop ) prop . observables . disable_all ( ) else : self . _ghost_props = [ ] # initialize reward channels self . _reset_reward_channels ( )