Terminating Functions#

In this section we describe the terminating function protocol, the terminating function registry, and how to write your own custom terminating functions.

The TerminatingFunction Protocol#

A terminating function is a deterministic mapping from a state-action-state transition to a boolean indicating the termination of an episode. Using the typing standard library, the terminating function type is defined as a typing.Callable which receives a State, an Action, and another State, and returns a bool.

TerminatingFunction = <class 'gym_gridverse.envs.terminating_functions.TerminatingFunction'>[source]

Signature for functions to determine whether a transition is terminal

Note

A terminating function may (and often does) accept additional arguments; this is possible so long as the extra arguments either have default values, or are binded to specific values later on, e.g., using functools.partial().

The Terminating Function Registry#

The terminating_functions module contains some pre-defined terminating functions and the terminating_function_registry, a dictionary-like object through which to register and retrieve terminating functions. Terminating functions are registered using the register() method, which can be used as a decorator (also see Custom Terminating Functions). As a dictionary, terminating_function_registry has a keys() method which returns the names of registered functions.

Custom Terminating Functions#

Custom terminating functions can be defined so long as they satisfy some basic rules; A custom terminating function:

  • MUST satisfy the TerminatingFunction protocol.

  • MUST NOT edit the input states.

  • SHOULD be wholly deterministic.

Practical Example 1#

Note

The examples shown here can be found in the examples/ folder.

In this example, we are going to write a terminating function in which the episode terminates if the agent has not changed from the previous time-step, i.e., it’s position, orientation, and object held are the same.

from gym_gridverse.action import Action
from gym_gridverse.envs.terminating_functions import (
    terminating_function_registry,
)
from gym_gridverse.state import State


@terminating_function_registry.register
def static_agent(state: State, action: Action, next_state: State) -> bool:
    return state.agent == next_state.agent

Practical Example 2#

In this example, we are going to write a terminating function in which the episode terminates if the agent does not observe a given object; this could be an interesting way to encode a “escort mission” task, where the agent needs to keep an eye on a moving target.

Note

The following example has a couple of minor issues. First of all, the observation used in the terminating function might not be the same received by the agent; this could happen if the observation_function itself is different, or if it’s stochastic. Second, because the TerminatingFunction protocol does not have an rng argument, this terminating function does not allow to fully reproduce executions, if the given observation function is stochastic.

from typing import Type

from gym_gridverse.action import Action
from gym_gridverse.envs.observation_functions import ObservationFunction
from gym_gridverse.envs.terminating_functions import (
    terminating_function_registry,
)
from gym_gridverse.grid_object import GridObject
from gym_gridverse.state import State


@terminating_function_registry.register
def concealed_terminating(
    state: State,
    action: Action,
    next_state: State,
    *,
    object_type: Type[GridObject],
    observation_function: ObservationFunction
) -> bool:
    # NOTE:  even if the observation_function is the same, this observation
    # might be different from that received by the agent if it is stochastic.
    observation = observation_function(state)
    return any(
        isinstance(observation.grid[position], object_type)
        for position in observation.grid.area.positions()
    )