Spaces:
Sleeping
Sleeping
File size: 3,044 Bytes
af8c4cf 1d63133 af8c4cf a631394 af8c4cf 1d63133 | 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 | from __future__ import annotations
from RevLogiXenv_v0 import (
AutonomousReturnsEnv,
DispositionAction,
HeuristicPolicy,
ReturnsAction,
)
def _run_episode(task: str, seed: int) -> tuple[float, list[float], list[str]]:
env = AutonomousReturnsEnv(task)
obs = env.reset(seed=seed)
policy = HeuristicPolicy()
rewards: list[float] = []
while not obs.done:
action = policy(obs)
obs = env.step(action)
if obs.reward_detail is not None:
rewards.append(obs.reward_detail.value)
history_actions = [r.action.value for r in env._full_resolution_history]
return env.state.total_ledger, rewards, history_actions
def test_reward_bounds_across_tasks() -> None:
policy = HeuristicPolicy()
for task in ("easy", "medium", "hard"):
env = AutonomousReturnsEnv(task)
obs = env.reset(seed=42)
while not obs.done:
obs = env.step(policy(obs))
if obs.reward_detail is not None:
assert 0.01 <= obs.reward_detail.value <= 0.99
def test_repeatability_for_fixed_seed() -> None:
result_1 = _run_episode(task="hard", seed=1337)
result_2 = _run_episode(task="hard", seed=1337)
assert result_1 == result_2
def test_wait_on_active_item_is_penalized() -> None:
env = AutonomousReturnsEnv("easy")
obs = env.reset(seed=42)
assert obs.current_item is not None
obs = env.step(ReturnsAction(action=DispositionAction.WAIT))
assert obs.reward_detail is not None
assert obs.reward_detail.raw_step_reward < 0.0
assert "wait_with_active_item" in obs.reward_detail.penalties
def test_wait_on_tail_has_no_invalid_penalty() -> None:
env = AutonomousReturnsEnv("medium")
obs = env.reset(seed=42)
while obs.current_item is not None and not obs.done:
obs = env.step(ReturnsAction(action=DispositionAction.RESELL_DISCOUNT_30))
assert obs.current_item is None
assert env.pending_queue
obs = env.step(ReturnsAction(action=DispositionAction.WAIT))
assert obs.reward_detail is not None
assert "invalid_action_when_idle" not in obs.reward_detail.penalties
assert "non_wait_on_tail" not in obs.reward_detail.penalties
def test_step_after_done_is_guarded() -> None:
env = AutonomousReturnsEnv("easy")
obs = env.reset(seed=7)
policy = HeuristicPolicy()
while not obs.done:
obs = env.step(policy(obs))
ledger_before = env.state.total_ledger
obs_after = env.step(ReturnsAction(action=DispositionAction.RESELL_FULL))
assert obs_after.done is True
assert obs_after.reward_detail is not None
assert obs_after.reward_detail.raw_step_reward == 0.0
assert "post_done_step" in obs_after.reward_detail.penalties
assert env.state.total_ledger == ledger_before
def test_reset_task_override_changes_active_task() -> None:
env = AutonomousReturnsEnv("easy")
obs = env.reset(seed=11, task="hard")
assert env.task == "hard"
assert env.state.task == "hard"
assert obs.remaining_items == 40
|