(self)
return enabled_specs
The observation specification for this environment.

    Returns a dict mapping the names of enabled observations to their
    corresponding `Array` or `BoundedArray` specs.

    If an obs has a BoundedArray spec, but uses an aggregator that
    does not preserve those bounds (such as `sum`), it will be mapped to an
    (unbounded) `Array` spec. If using a bounds-preserving custom aggregator
    `my_agg`, give it an attribute `my_agg.preserves_bounds = True` to indicate
    to this method that it is bounds-preserving.

    The returned specification is only valid as of the previous call
    to `reset`. In particular, it is an error to call this function before
    the first call to `reset`.

    Returns:
      A dict mapping observation name to `Array` or `BoundedArray` spec
      containing the observation shape and dtype, and possibly bounds.

    Raises:
      RuntimeError: If this method is called before `reset` has been called.
The observation specification for this environment.
The observation specification for this environment .
def observation_spec(self):
    """The observation specification for this environment.

    Returns a dict mapping the names of enabled observations to their
    corresponding `Array` or `BoundedArray` specs.

    If an obs has a BoundedArray spec, but uses an aggregator that
    does not preserve those bounds (such as `sum`), it will be mapped to an
    (unbounded) `Array` spec. If using a bounds-preserving custom aggregator
    `my_agg`, give it an attribute `my_agg.preserves_bounds = True` to indicate
    to this method that it is bounds-preserving.

    The returned specification is only valid as of the previous call
    to `reset`. In particular, it is an error to call this function before
    the first call to `reset`.

    Returns:
      A dict mapping observation name to `Array` or `BoundedArray` spec
      containing the observation shape and dtype, and possibly bounds.

    Raises:
      RuntimeError: If this method is called before `reset` has been called.
    """
    if self._enabled_structure is None:
      raise RuntimeError('`reset` must be called before `observation_spec`.')

    def make_observation_spec_dict(enabled_dict):
      """Makes a dict of enabled observation specs from of observables."""
      out_dict = type(enabled_dict)()
      for name, enabled in enabled_dict.items():

        if isinstance(enabled.observable.array_spec, specs.BoundedArray):
          bounds = (enabled.observable.array_spec.minimum,
                    enabled.observable.array_spec.maximum)
        else:
          bounds = None

        if enabled.observable.aggregator:
          aggregator = enabled.observable.aggregator
          aggregated = aggregator(np.zeros(enabled.buffer.shape,
                                           dtype=enabled.buffer.dtype))
          shape = aggregated.shape
          dtype = aggregated.dtype

          # Ditch bounds if the aggregator isn't known to be bounds-preserving.
          if bounds:
            if not hasattr(aggregator, 'preserves_bounds'):
              logging.warning('Ignoring the bounds of this observable\'s spec, '
                              'as its aggregator method has no boolean '
                              '`preserves_bounds` attrubute.')
              bounds = None
            elif not aggregator.preserves_bounds:
              bounds = None
        else:
          shape = enabled.buffer.shape
          dtype = enabled.buffer.dtype

        if bounds:
          spec = specs.BoundedArray(minimum=bounds[0],
                                    maximum=bounds[1],
                                    shape=shape,
                                    dtype=dtype,
                                    name=name)
        else:
          spec = specs.Array(shape=shape, dtype=dtype, name=name)

        out_dict[name] = spec
      return out_dict

    if self._is_nested:
      enabled_specs = type(self._enabled_structure)(
          make_observation_spec_dict(enabled_dict)
          for enabled_dict in self._enabled_structure)
    else:
      enabled_specs = make_observation_spec_dict(self._enabled_structure)

    return enabled_specs
def observation_spec ( self ) : if self . _enabled_structure is None : raise RuntimeError ( '`reset` must be called before `observation_spec`.' ) def make_observation_spec_dict ( enabled_dict ) : """Makes a dict of enabled observation specs from of observables.""" out_dict = type ( enabled_dict ) ( ) for name , enabled in enabled_dict . items ( ) : if isinstance ( enabled . observable . array_spec , specs . BoundedArray ) : bounds = ( enabled . observable . array_spec . minimum , enabled . observable . array_spec . maximum ) else : bounds = None if enabled . observable . aggregator : aggregator = enabled . observable . aggregator aggregated = aggregator ( np . zeros ( enabled . buffer . shape , dtype = enabled . buffer . dtype ) ) shape = aggregated . shape dtype = aggregated . dtype # Ditch bounds if the aggregator isn't known to be bounds-preserving. if bounds : if not hasattr ( aggregator , 'preserves_bounds' ) : logging . warning ( 'Ignoring the bounds of this observable\'s spec, ' 'as its aggregator method has no boolean ' '`preserves_bounds` attrubute.' ) bounds = None elif not aggregator . preserves_bounds : bounds = None else : shape = enabled . buffer . shape dtype = enabled . buffer . dtype if bounds : spec = specs . BoundedArray ( minimum = bounds [ 0 ] , maximum = bounds [ 1 ] , shape = shape , dtype = dtype , name = name ) else : spec = specs . Array ( shape = shape , dtype = dtype , name = name ) out_dict [ name ] = spec return out_dict if self . _is_nested : enabled_specs = type ( self . _enabled_structure ) ( make_observation_spec_dict ( enabled_dict ) for enabled_dict in self . _enabled_structure ) else : enabled_specs = make_observation_spec_dict ( self . _enabled_structure ) return enabled_specs