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()
)