Skip to content

core.base_abstractions.preprocessor#

[view_source]

Preprocessor#

class Preprocessor(abc.ABC)

[view_source]

Represents a preprocessor that transforms data from a sensor or another preprocessor to the input of agents or other preprocessors. The user of this class needs to implement the process method and the user is also required to set the below attributes:

Attributes:#

input_uuids : List of input universally unique ids. uuid : Universally unique id. observation_space : gym.Space object corresponding to processed observation spaces.

Preprocessor.process#

 | @abc.abstractmethod
 | process(obs: Dict[str, Any], *args: Any, **kwargs: Any) -> Any

[view_source]

Returns processed observations from sensors or other preprocessors.

Parameters

  • obs : Dict with available observations and processed observations.

Returns

Processed observation.

PreprocessorGraph#

class PreprocessorGraph()

[view_source]

Represents a graph of preprocessors, with each preprocessor being identified through a universally unique id.

Attributes

  • preprocessors: List containing preprocessors with required input uuids, output uuid of each sensor must be unique.

PreprocessorGraph.__init__#

 | __init__(preprocessors: List[Union[Preprocessor, Builder[Preprocessor]]]) -> None

[view_source]

Initializer.

Parameters

  • preprocessors : The preprocessors that will be included in the graph.

PreprocessorGraph.get#

 | get(uuid: str) -> Preprocessor

[view_source]

Return preprocessor with the given uuid.

Parameters

  • uuid : The unique id of the preprocessor.

Returns

The preprocessor with unique id uuid.

PreprocessorGraph.get_observations#

 | get_observations(obs: Dict[str, Any], *args: Any, **kwargs: Any) -> Dict[str, Any]

[view_source]

Get processed observations.

Returns

Collect observations processed from all sensors and return them packaged inside a Dict.

ObservationSet#

class ObservationSet()

[view_source]

Represents a list of source_ids, corresponding to sensors and preprocessors, with each source being identified through a unique id.

Attributes

  • source_ids: List containing sensor and preprocessor ids to be consumed by agents. Each source uuid must be unique.
  • graph: Computation graph for all preprocessors.
  • observation_spaces: Observation spaces of all output sources.
  • device: Device where the PreprocessorGraph is executed.

ObservationSet.__init__#

 | __init__(source_ids: List[str], all_preprocessors: List[Union[Preprocessor, Builder[Preprocessor]]], all_sensors: List[Sensor]) -> None

[view_source]

Initializer.

Parameters

  • source_ids : The sensors and preprocessors that will be included in the set.
  • all_preprocessors : The entire list of preprocessors to be executed.
  • all_sensors : The entire list of sensors.

ObservationSet.get#

 | get(uuid: str) -> Preprocessor

[view_source]

Return preprocessor with the given uuid.

Parameters

  • uuid : The unique id of the preprocessor.

Returns

The preprocessor with unique id uuid.

ObservationSet.get_observations#

 | get_observations(obs: Dict[str, Any], *args: Any, **kwargs: Any) -> Dict[str, Any]

[view_source]

Get all observations within a dictionary.

Returns

Collect observations from all sources and return them packaged inside a Dict.