Observation extractor

---

Air Force Research Laboratory (AFRL) Autonomous Capabilities Team (ACT3) Reinforcement Learning (RL) Core.

This is a US Government Work not subject to copyright protection in the US.

The use, dissemination or disclosure of data in this file is subject to limitation or restriction. See accompanying README and LICENSE for details.

---

Observation Extractor

ExtractorSet (tuple)

Class defining the set of extractors to pull information about a specific observation

Source code in corl/libraries/observation_extractor.py

class ExtractorSet(typing.NamedTuple):
    """Class defining the set of extractors to pull information about a specific observation
    """
    value: typing.Callable
    space: typing.Callable
    unit: typing.Callable

__getnewargs__(self) special

Return self as a plain tuple. Used by copy and pickle.

Source code in corl/libraries/observation_extractor.py

def __getnewargs__(self):
    'Return self as a plain tuple.  Used by copy and pickle.'
    return _tuple(self)

__new__(_cls, value, space, unit) special staticmethod

Create new instance of ExtractorSet(value, space, unit)

__repr__(self) special

Return a nicely formatted representation string

Source code in corl/libraries/observation_extractor.py

def __repr__(self):
    'Return a nicely formatted representation string'
    return self.__class__.__name__ + repr_fmt % self

ObservationExtractor(observation, fields, indices=None)

This function extracts a metric produced by a specific glue in observation space of an agent

---- Arguments ---- observation - A dict of an agent's observations platforms - The platforms the glue is observing, needed to compute the glue's prefix fields - Fields the extractor will attempt to walk through indicies - These will be accessed after the fields have been accessed, allowing users to reduce arrays to single values ---- Raises ---- RuntimeError - Thrown when the fields do not exists in the glue's measurments

Source code in corl/libraries/observation_extractor.py

def ObservationExtractor(
    observation,
    fields: typing.List[str],
    indices: typing.Union[int, typing.List[int]] = None,
):
    """ This function extracts a metric produced by a specific glue in observation space of an agent

    ---- Arguments ----
    observation         - A dict of an agent's observations
    platforms           - The platforms the glue is observing, needed to compute the glue's prefix
    fields              - Fields the extractor will attempt to walk through
    indicies            - These will be accessed after the fields have been accessed, allowing
                          users to reduce arrays to single values
    ---- Raises ----
    RuntimeError        - Thrown when the fields do not exists in the glue's measurments
    """
    if indices is None:
        indices = []
    if not isinstance(indices, typing.List):
        indices = [indices]

    value = observation
    for field in fields:
        if field not in value:
            raise RuntimeError(f"The field {field} is not present in the observation, the requested fields were {fields}, ")
        value = value[field]
    for index in indices:
        value = value[index]
    return value

ObservationSpaceExtractor(observation_space, fields)

Extract the observation space from a glue

---- Arguments ---- observation - A dict of an agent's observations fields - Fields the extractor will attempt to walk through

---- Raises ---- RuntimeError - Thrown when the field does not exists in the glue's obs space

Source code in corl/libraries/observation_extractor.py

def ObservationSpaceExtractor(
    observation_space,
    fields: typing.List[str],
):
    """Extract the observation space from a glue

     ---- Arguments ----
    observation         - A dict of an agent's observations
    fields              - Fields the extractor will attempt to walk through

    ---- Raises ----
    RuntimeError        - Thrown when the field does not exists in the glue's obs space
    """
    space = observation_space
    for field in fields:
        if field not in space.spaces:
            raise RuntimeError(
                f"The field {field} is not present in the observation space,"
                f"the requested fields were {fields}, space is {observation_space}"
            )
        space = space[field]
    return space