test load_results on log files produced by the monitor wrapper
(tmp_path)
| 68 | |
| 69 | |
| 70 | def test_monitor_load_results(tmp_path): |
| 71 | """ |
| 72 | test load_results on log files produced by the monitor wrapper |
| 73 | """ |
| 74 | original_tmp_path = tmp_path |
| 75 | tmp_path = str(tmp_path) |
| 76 | env1 = gym.make("CartPole-v1") |
| 77 | env1.reset(seed=0) |
| 78 | with pytest.raises(LoadMonitorResultsError): |
| 79 | load_results(tmp_path) |
| 80 | |
| 81 | monitor_file1 = os.path.join(tmp_path, f"stable_baselines-test-{uuid.uuid4()}.monitor.csv") |
| 82 | monitor_env1 = Monitor(env1, monitor_file1) |
| 83 | |
| 84 | monitor_files = get_monitor_files(tmp_path) |
| 85 | assert len(monitor_files) == 1 |
| 86 | assert monitor_file1 in monitor_files |
| 87 | |
| 88 | monitor_env1.reset() |
| 89 | episode_count1 = 0 |
| 90 | for _ in range(1000): |
| 91 | _, _, terminated, truncated, _ = monitor_env1.step(monitor_env1.action_space.sample()) |
| 92 | if terminated or truncated: |
| 93 | episode_count1 += 1 |
| 94 | monitor_env1.reset() |
| 95 | |
| 96 | results_size1 = len(load_results(tmp_path).index) |
| 97 | assert results_size1 == episode_count1 |
| 98 | |
| 99 | env2 = gym.make("CartPole-v1") |
| 100 | env2.reset(seed=0) |
| 101 | monitor_file2 = os.path.join(tmp_path, f"stable_baselines-test-{uuid.uuid4()}.monitor.csv") |
| 102 | monitor_env2 = Monitor(env2, monitor_file2) |
| 103 | monitor_files = get_monitor_files(tmp_path) |
| 104 | assert len(monitor_files) == 2 |
| 105 | assert monitor_file1 in monitor_files |
| 106 | assert monitor_file2 in monitor_files |
| 107 | |
| 108 | episode_count2 = 0 |
| 109 | for _ in range(2): |
| 110 | # Test appending to existing file |
| 111 | monitor_env2 = Monitor(env2, monitor_file2, override_existing=False) |
| 112 | monitor_env2.reset() |
| 113 | for _ in range(1000): |
| 114 | _, _, terminated, truncated, _ = monitor_env2.step(monitor_env2.action_space.sample()) |
| 115 | if terminated or truncated: |
| 116 | episode_count2 += 1 |
| 117 | monitor_env2.reset() |
| 118 | |
| 119 | results_size2 = len(load_results(tmp_path).index) |
| 120 | |
| 121 | assert results_size2 == (results_size1 + episode_count2) |
| 122 | |
| 123 | empty_monitor = original_tmp_path / "demo" / "empty_monitor.csv" |
| 124 | empty_monitor.parent.mkdir() |
| 125 | |
| 126 | empty_monitor.write_text(EMPTY_MONITOR) |
| 127 | empty_df = load_results(empty_monitor.parent) |
nothing calls this directly
no test coverage detected
searching dependent graphs…