Skip to content

termination_condition_base

BaseTerminationCondition

Bases: Registerable

Base TerminationCondition class Condition-specific _step() method is implemented in subclasses

Source code in OmniGibson/omnigibson/termination_conditions/termination_condition_base.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
class BaseTerminationCondition(Registerable, metaclass=ABCMeta):
    """
    Base TerminationCondition class
    Condition-specific _step() method is implemented in subclasses
    """

    def __init__(self):
        # Initialize internal vars that will be filled in at runtime
        self._num_envs = None
        self._done = None

    @abstractmethod
    def _step(self, task, env, action):
        """
        Step the termination condition and return whether the episode should terminate. Overwritten by subclasses.

        Args:
            task (BaseTask): Task instance
            env (Environment): Environment instance
            action (n-array): 1D flattened array of actions executed by all agents in the environment

        Returns:
            th.Tensor: (num_envs,) bool tensor indicating whether each env should terminate
        """
        raise NotImplementedError()

    def step(self, task, env, action):
        """
        Step the termination condition and return whether the episode should terminate.

        Args:
            task (BaseTask): Task instance
            env (Environment): Environment instance
            action (n-array): 1D flattened array of actions executed by all agents in the environment

        Returns:
            2-tuple:
                - th.Tensor: (num_envs,) bool tensor indicating whether each env should terminate or not
                - th.Tensor: (num_envs,) bool tensor indicating whether a success was reached under this termination condition
        """
        # Step internally and store the done state internally as well
        self._done = self._step(task=task, env=env, action=action)

        # We are successful if done is True AND this is a success condition
        success = self._done & self._terminate_is_success

        return self._done, success

    def reset(self, task, env, env_indices):
        """
        Termination condition-specific reset

        Args:
            task (BaseTask): Task instance
            env (Environment): Environment instance
            env_indices (th.Tensor): Indices of environments to reset
        """
        if self._num_envs is None:
            self._num_envs = env.num_envs
            self._done = th.zeros(self._num_envs, dtype=th.bool)
        self._done[env_indices] = False

    @property
    def done(self):
        """
        Returns:
            th.Tensor: (num_envs,) bool tensor indicating whether this termination condition has triggered
        """
        assert self._done is not None, "At least one step() must occur before done can be calculated!"
        return self._done

    @property
    def success(self):
        """
        Returns:
            th.Tensor: (num_envs,) bool tensor indicating whether this termination condition is a success
        """
        assert self._done is not None, "At least one step() must occur before success can be calculated!"
        return self._done & self._terminate_is_success

    @classproperty
    def _terminate_is_success(cls):
        """
        Returns:
            bool: Whether this termination condition corresponds to a success
        """
        raise NotImplementedError()

    @classproperty
    def _do_not_register_classes(cls):
        # Don't register this class since it's an abstract template
        classes = super()._do_not_register_classes
        classes.add("BaseTerminationCondition")
        return classes

    @classproperty
    def _cls_registry(cls):
        # Global registry
        global REGISTERED_TERMINATION_CONDITIONS
        return REGISTERED_TERMINATION_CONDITIONS

done property

Returns:

Type Description
Tensor

(num_envs,) bool tensor indicating whether this termination condition has triggered

success property

Returns:

Type Description
Tensor

(num_envs,) bool tensor indicating whether this termination condition is a success

reset(task, env, env_indices)

Termination condition-specific reset

Parameters:

Name Type Description Default
task BaseTask

Task instance

 required
env Environment

Environment instance

 required
env_indices Tensor

Indices of environments to reset

 required
Source code in OmniGibson/omnigibson/termination_conditions/termination_condition_base.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
def reset(self, task, env, env_indices):
    """
    Termination condition-specific reset

    Args:
        task (BaseTask): Task instance
        env (Environment): Environment instance
        env_indices (th.Tensor): Indices of environments to reset
    """
    if self._num_envs is None:
        self._num_envs = env.num_envs
        self._done = th.zeros(self._num_envs, dtype=th.bool)
    self._done[env_indices] = False

step(task, env, action)

Step the termination condition and return whether the episode should terminate.

Parameters:

Name Type Description Default
task BaseTask

Task instance

 required
env Environment

Environment instance

 required
action n - array

1D flattened array of actions executed by all agents in the environment

 required

Returns:

Type Description
2 - tuple
  • th.Tensor: (num_envs,) bool tensor indicating whether each env should terminate or not
  • th.Tensor: (num_envs,) bool tensor indicating whether a success was reached under this termination condition
Source code in OmniGibson/omnigibson/termination_conditions/termination_condition_base.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def step(self, task, env, action):
    """
    Step the termination condition and return whether the episode should terminate.

    Args:
        task (BaseTask): Task instance
        env (Environment): Environment instance
        action (n-array): 1D flattened array of actions executed by all agents in the environment

    Returns:
        2-tuple:
            - th.Tensor: (num_envs,) bool tensor indicating whether each env should terminate or not
            - th.Tensor: (num_envs,) bool tensor indicating whether a success was reached under this termination condition
    """
    # Step internally and store the done state internally as well
    self._done = self._step(task=task, env=env, action=action)

    # We are successful if done is True AND this is a success condition
    success = self._done & self._terminate_is_success

    return self._done, success

FailureCondition

Bases: BaseTerminationCondition

Termination condition corresponding to a failure

Source code in OmniGibson/omnigibson/termination_conditions/termination_condition_base.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
class FailureCondition(BaseTerminationCondition):
    """
    Termination condition corresponding to a failure
    """

    def __init_subclass__(cls, **kwargs):
        # Register as part of locomotion controllers
        super().__init_subclass__(**kwargs)
        register_failure_condition(cls)

    @classproperty
    def _terminate_is_success(cls):
        # Done --> not success
        return False

    @classproperty
    def _do_not_register_classes(cls):
        # Don't register this class since it's an abstract template
        classes = super()._do_not_register_classes
        classes.add("FailureCondition")
        return classes

SuccessCondition

Bases: BaseTerminationCondition

Termination condition corresponding to a success

Source code in OmniGibson/omnigibson/termination_conditions/termination_condition_base.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
class SuccessCondition(BaseTerminationCondition):
    """
    Termination condition corresponding to a success
    """

    def __init_subclass__(cls, **kwargs):
        # Register as part of locomotion controllers
        super().__init_subclass__(**kwargs)
        register_success_condition(cls)

    @classproperty
    def _terminate_is_success(cls):
        # Done --> success
        return True

    @classproperty
    def _do_not_register_classes(cls):
        # Don't register this class since it's an abstract template
        classes = super()._do_not_register_classes
        classes.add("SuccessCondition")
        return classes