Skip to content

termination_condition_base

BaseTerminationCondition

Bases: Registerable

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

Source code in omnigibson/termination_conditions/termination_condition_base.py 
 19
 20
 21
 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
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._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:
            bool: whether environment should terminate or not
        """
        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:
                - bool: whether environment should terminate or not
                - bool: 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 and self._terminate_is_success

        return self._done, success

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

        Args:
            task (BaseTask): Task instance
            env (Environment): Environment instance
        """
        # Reset internal vars
        self._done = None

    @property
    def done(self):
        """
        Returns:
            bool: Whether this termination condition has triggered or not
        """
        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:
            bool: Whether this termination condition has been evaluated as a success or not
        """
        assert self._done is not None, "At least one step() must occur before success can be calculated!"
        return self._done and 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:

Name Type Description
bool

Whether this termination condition has triggered or not

success property

Returns:

Name Type Description
bool

Whether this termination condition has been evaluated as a success or not

reset(task, env)

Termination condition-specific reset

Parameters:

Name Type Description Default
task BaseTask

Task instance

 required
env Environment

Environment instance

 required
Source code in omnigibson/termination_conditions/termination_condition_base.py 
65
66
67
68
69
70
71
72
73
74
def reset(self, task, env):
    """
    Termination condition-specific reset

    Args:
        task (BaseTask): Task instance
        env (Environment): Environment instance
    """
    # Reset internal vars
    self._done = None

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: - bool: whether environment should terminate or not - bool: whether a success was reached under this termination condition
Source code in omnigibson/termination_conditions/termination_condition_base.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
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:
            - bool: whether environment should terminate or not
            - bool: 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 and self._terminate_is_success

    return self._done, success

FailureCondition

Bases: BaseTerminationCondition

Termination condition corresponding to a failure

Source code in omnigibson/termination_conditions/termination_condition_base.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
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/termination_conditions/termination_condition_base.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
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