def observation()

in robogym/wrappers/randomizations.py [0:0]

• 28 lines of code
• 7 McCabe index (conditional complexity)

    def observation(self, observation):
        randomized_observation = OrderedDict()
        for key in observation:
            randomized_observation[key] = observation[key]
        for key in sorted(self._levels):
            key_len = self.key_length(key)
            uncorrelated_bias = (
                self.random_state.randn(key_len)
                * self._levels[key].get("uncorrelated", 0.0)
                * self._uncorrelated_multipler
            )
            additive_bias = self._additive_bias[key] + uncorrelated_bias

            if f"noisy_{key}" in observation:
                # There is already noisy value available for this observation key,
                # we apply noise on top of the noisy value.
                obs_key = f"noisy_{key}"
            else:
                # Apply noise on top of noiseless observation if no noisy value available.
                obs_key = key

            new_value = observation[obs_key].copy()

            if not key.endswith("_quat"):
                new_value *= self._multiplicative_bias[key]
                new_value += additive_bias
            else:
                assert np.allclose(self._multiplicative_bias[key], 1.0)
                noise_axis = self.random_state.uniform(-1.0, 1.0, size=(3,))
                additive_bias *= QUAT_NOISE_CORRECTION
                noise_quat = quat_from_angle_and_axis(additive_bias, noise_axis)
                new_value = quat_normalize(quat_mul(new_value, noise_quat))

            randomized_observation[f"noisy_{key}"] = new_value

        return randomized_observation