Skip to content

metric_base

MetricBase

Class for defining a programmatic environment metric that can be tracked over the course of each environment episode

Source code in OmniGibson/omnigibson/metrics/metric_base.py 
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
class MetricBase:
    """
    Class for defining a programmatic environment metric that can be tracked over the course of
    each environment episode
    """

    def __init__(self, env_idx=0, env_accessor=None):
        # Batched evaluation binds a metric to an accessor, which owns the mapping into the shared
        # environment. env_idx remains as a compatibility path for MetricsWrapper and external users.
        self.env_idx = env_idx
        self.env_accessor = env_accessor
        self.state = dict()

    def _resolve_env(self, env):
        if env is not None:
            return env
        if self.env_accessor is None:
            raise ValueError("Metric is not bound to an environment accessor; pass env explicitly.")
        return self.env_accessor

    def _scene(self, env=None):
        """Return the scene tracked by this metric for bound and legacy callers."""
        env = self._resolve_env(env)
        return env.scene if env is self.env_accessor else env.scenes[self.env_idx]

    @classmethod
    def is_compatible(cls, env):
        """
        Checks if this metric class is compatible with @env

        Args:
            env (og.Environment or EnvironmentWrapper): Environment to check compatibility

        Returns:
            bool: Whether this metric is compatible or not
        """
        return True

    @classmethod
    def validate_episode(cls, episode_metrics, **kwargs):
        """
        Validates the given @episode_metrics from self.aggregate_results using any specific @kwargs

        Args:
            episode_metrics (dict): Metrics aggregated using self.aggregate_results
            kwargs (Any): Any keyword arguments relevant to this specific MetricBase

        Returns:
            dict: Keyword-mapped dictionary mapping each validation test name to {"success": bool, "feedback": str} dict
                where "success" is True if the given @episode_metrics pass that specific test; if False, "feedback"
                provides information as to why the test failed
        """
        raise NotImplementedError

    def step(self, env=None, action=None, obs=None, reward=None, terminated=None, truncated=None, info=None):
        """
        Steps this metric, updating any internal values being tracked.

        Args:
            action (th.Tensor): action deployed resulting in @obs
            obs (dict): state, i.e. observation
            reward (float): reward, i.e. reward at this current timestep
            terminated (bool): terminated, i.e. whether this episode ended due to a failure or success
            truncated (bool): truncated, i.e. whether this episode ended due to a time limit etc.
            info (dict): info, i.e. dictionary with any useful information
        """
        env = self._resolve_env(env)
        step_metrics = self._compute_step_metrics(env, action, obs, reward, terminated, truncated, info)
        scene = self._scene(env)
        assert scene in self.state, f"Environment {scene} is not being tracked, please call 'self.reset()' to track!"
        state = self.state[scene]
        for k, v in step_metrics.items():
            if k not in state:
                state[k] = []
            state[k].append(v)

    def _compute_step_metrics(self, env, action, obs, reward, terminated, truncated, info):
        """
        Compute any step-wise metrics at the current environment step that just occurred

        Args:
            action (th.Tensor): action deployed resulting in @obs
            obs (dict): state, i.e. observation
            reward (float): reward, i.e. reward at this current timestep
            terminated (bool): terminated, i.e. whether this episode ended due to a failure or success
            truncated (bool): truncated, i.e. whether this episode ended due to a time limit etc.
            info (dict): info, i.e. dictionary with any useful information

        Returns:
            dict: Any per-step information that should be internally tracked
        """
        raise NotImplementedError

    def _compute_episode_metrics(self, env, episode_info):
        """
        Computes the aggregated metrics over the current trajectory episode in @env

        Args:
            episode_info (dict): Internal information that was tracked using @_compute_episode metrics. This
                information is is the same key-mapped dict as @_compute_step_metrics mapped to the
                list of values aggregated over the current trajectory episode

        Returns:
            dict: Any per-step information that should be internally tracked
        """
        raise NotImplementedError

    def aggregate(self, env=None):
        """
        Aggregates information over the current trajectory tracked by this metric.

        Returns:
            dict: Any relevant aggregated metric information
        """
        env = self._resolve_env(env)
        scene = self._scene(env)
        if scene in self.state:
            if self.state[scene] == dict():
                return dict()
            else:
                return self._compute_episode_metrics(env=env, episode_info=self.state[scene])
        else:
            print("Environment not yet tracked, skipping metric aggregation!")
            return dict()

    def reset(self, env=None):
        """
        Resets this metric for its bound logical environment.
        """
        self.state[self._scene(env)] = dict()

aggregate(env=None)

Aggregates information over the current trajectory tracked by this metric.

Returns:

Type Description
dict

Any relevant aggregated metric information
Source code in OmniGibson/omnigibson/metrics/metric_base.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def aggregate(self, env=None):
    """
    Aggregates information over the current trajectory tracked by this metric.

    Returns:
        dict: Any relevant aggregated metric information
    """
    env = self._resolve_env(env)
    scene = self._scene(env)
    if scene in self.state:
        if self.state[scene] == dict():
            return dict()
        else:
            return self._compute_episode_metrics(env=env, episode_info=self.state[scene])
    else:
        print("Environment not yet tracked, skipping metric aggregation!")
        return dict()

is_compatible(env) classmethod

Checks if this metric class is compatible with @env

Parameters:

Name Type Description Default
env Environment or EnvironmentWrapper

Environment to check compatibility

 required

Returns:

Type Description
bool

Whether this metric is compatible or not
Source code in OmniGibson/omnigibson/metrics/metric_base.py 
26
27
28
29
30
31
32
33
34
35
36
37
@classmethod
def is_compatible(cls, env):
    """
    Checks if this metric class is compatible with @env

    Args:
        env (og.Environment or EnvironmentWrapper): Environment to check compatibility

    Returns:
        bool: Whether this metric is compatible or not
    """
    return True

reset(env=None)

Resets this metric for its bound logical environment.

Source code in OmniGibson/omnigibson/metrics/metric_base.py 
126
127
128
129
130
def reset(self, env=None):
    """
    Resets this metric for its bound logical environment.
    """
    self.state[self._scene(env)] = dict()

step(env=None, action=None, obs=None, reward=None, terminated=None, truncated=None, info=None)

Steps this metric, updating any internal values being tracked.

Parameters:

Name Type Description Default
action Tensor

action deployed resulting in @obs

 None
obs dict

state, i.e. observation

 None
reward float

reward, i.e. reward at this current timestep

 None
terminated bool

terminated, i.e. whether this episode ended due to a failure or success

 None
truncated bool

truncated, i.e. whether this episode ended due to a time limit etc.

 None
info dict

info, i.e. dictionary with any useful information

 None
Source code in OmniGibson/omnigibson/metrics/metric_base.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def step(self, env=None, action=None, obs=None, reward=None, terminated=None, truncated=None, info=None):
    """
    Steps this metric, updating any internal values being tracked.

    Args:
        action (th.Tensor): action deployed resulting in @obs
        obs (dict): state, i.e. observation
        reward (float): reward, i.e. reward at this current timestep
        terminated (bool): terminated, i.e. whether this episode ended due to a failure or success
        truncated (bool): truncated, i.e. whether this episode ended due to a time limit etc.
        info (dict): info, i.e. dictionary with any useful information
    """
    env = self._resolve_env(env)
    step_metrics = self._compute_step_metrics(env, action, obs, reward, terminated, truncated, info)
    scene = self._scene(env)
    assert scene in self.state, f"Environment {scene} is not being tracked, please call 'self.reset()' to track!"
    state = self.state[scene]
    for k, v in step_metrics.items():
        if k not in state:
            state[k] = []
        state[k].append(v)

validate_episode(episode_metrics, **kwargs) classmethod

Validates the given @episode_metrics from self.aggregate_results using any specific @kwargs

Parameters:

Name Type Description Default
episode_metrics dict

Metrics aggregated using self.aggregate_results

 required
kwargs Any

Any keyword arguments relevant to this specific MetricBase

 {}

Returns:

Type Description
dict

Keyword-mapped dictionary mapping each validation test name to {"success": bool, "feedback": str} dict where "success" is True if the given @episode_metrics pass that specific test; if False, "feedback" provides information as to why the test failed
Source code in OmniGibson/omnigibson/metrics/metric_base.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@classmethod
def validate_episode(cls, episode_metrics, **kwargs):
    """
    Validates the given @episode_metrics from self.aggregate_results using any specific @kwargs

    Args:
        episode_metrics (dict): Metrics aggregated using self.aggregate_results
        kwargs (Any): Any keyword arguments relevant to this specific MetricBase

    Returns:
        dict: Keyword-mapped dictionary mapping each validation test name to {"success": bool, "feedback": str} dict
            where "success" is True if the given @episode_metrics pass that specific test; if False, "feedback"
            provides information as to why the test failed
    """
    raise NotImplementedError