in compiler_gym/envs/compiler_env.py [0:0]
def validate(self, state: Optional[CompilerEnvState] = None) -> ValidationResult:
"""Validate an environment's state.
:param state: A state to environment. If not provided, the current state
is validated.
:returns: A :class:`ValidationResult <compiler_gym.ValidationResult>`.
"""
if state:
self.reset(benchmark=state.benchmark)
in_place = False
else:
state = self.state
in_place = True
assert self.in_episode
errors: ValidationError = []
validation = {
"state": state,
"actions_replay_failed": False,
"reward_validated": False,
"reward_validation_failed": False,
"benchmark_semantics_validated": False,
"benchmark_semantics_validation_failed": False,
}
fkd = self.fork()
try:
with Timer() as walltime:
replay_target = self if in_place else fkd
replay_target.reset(benchmark=state.benchmark)
# Use a while loop here so that we can `break` early out of the
# validation process in case a step fails.
while True:
try:
replay_target.apply(state)
except (ValueError, OSError) as e:
validation["actions_replay_failed"] = True
errors.append(
ValidationError(
type="Action replay failed",
data={
"exception": str(e),
"exception_type": type(e).__name__,
},
)
)
break
if state.reward is not None and self.reward_space is None:
warnings.warn(
"Validating state with reward, but "
"environment has no reward space set"
)
elif (
state.reward is not None
and self.reward_space
and self.reward_space.deterministic
):
validation["reward_validated"] = True
# If reward deviates from the expected amount record the
# error but continue with the remainder of the validation.
if not isclose(
state.reward,
replay_target.episode_reward,
rel_tol=1e-5,
abs_tol=1e-10,
):
validation["reward_validation_failed"] = True
errors.append(
ValidationError(
type=(
f"Expected reward {state.reward} but "
f"received reward {replay_target.episode_reward}"
),
data={
"expected_reward": state.reward,
"actual_reward": replay_target.episode_reward,
},
)
)
benchmark = replay_target.benchmark
if benchmark.is_validatable():
validation["benchmark_semantics_validated"] = True
semantics_errors = benchmark.validate(replay_target)
if semantics_errors:
validation["benchmark_semantics_validation_failed"] = True
errors += semantics_errors
# Finished all checks, break the loop.
break
finally:
fkd.close()
return ValidationResult.construct(
walltime=walltime.time,
errors=errors,
**validation,
)