adressing issue Grid2op#374 and fix also behaviour for maintenance

grid2op/Environment/_ObsEnv.py

@@ -348,6 +348,12 @@ def init(
         self._topo_vect[:] = topo_vect
         # TODO update maintenance time, duration and cooldown accordingly (see all todos in `update_grid`)
 
+        # update the cooldowns (see issue https://github.com/rte-france/Grid2Op/issues/374)
+        tmp_ = self._times_before_line_status_actionable
+        tmp_[tmp_ > 0] += time_step
+        tmp_ = self._times_before_topology_actionable
+        tmp_[tmp_ > 0] += time_step
+
         # TODO set the shunts here
         # update the action that set the grid to the real value
         (
@@ -554,6 +560,7 @@ def simulate(self, action):
         if maybe_exc is not None:
             raise maybe_exc
         self._reset_to_orig_state()
+
         obs, reward, done, info = self.step(action)
         return obs, reward, done, info
 

grid2op/tests/test_ObservationHazard_Maintenance.py

@@ -360,6 +360,10 @@ def test_1_generating_obs_withmaintenance(self):
         )
 
     def test_simulate_disco_planned_maintenance(self):
+
+        reco_line = self.env.action_space()
+        reco_line.line_set_status = [(4, +1)]
+
         obs = self.env.get_obs()
         assert obs.line_status[4]
         assert obs.time_next_maintenance[4] == 1
@@ -370,19 +374,26 @@ def test_simulate_disco_planned_maintenance(self):
         assert not sim_obs.line_status[4]
         assert sim_obs.time_next_maintenance[4] == 0
         assert sim_obs.duration_next_maintenance[4] == 11
+
         # simulation at current step
         sim_obs, *_ = obs.simulate(self.env.action_space(), time_step=0)
         assert sim_obs.line_status[4]
         assert sim_obs.time_next_maintenance[4] == 1
         assert sim_obs.duration_next_maintenance[4] == 12
+
         # line will be disconnected next time step
         sim_obs, *_ = obs.simulate(self.env.action_space(), time_step=1)
         assert not sim_obs.line_status[4]
         assert sim_obs.time_next_maintenance[4] == 0
         assert sim_obs.duration_next_maintenance[4] == 11
 
-        for ts in range(12):
+        for ts in range(11):
             obs, reward, done, info = self.env.step(self.env.action_space())
+            assert obs.time_next_maintenance[4] == 0
+            assert obs.duration_next_maintenance[4] == 12-ts, f"should be {12-ts} but is {obs.duration_next_maintenance[4]} (step ts)"
+
+        # at this step if I attempt a reco it fails
+        obs, reward, done, info = self.env.step(self.env.action_space())
         # maintenance will be over next time step
         assert not obs.line_status[4]
         assert obs.time_next_maintenance[4] == 0
@@ -393,22 +404,36 @@ def test_simulate_disco_planned_maintenance(self):
         assert not sim_obs.line_status[4]
         assert sim_obs.time_next_maintenance[4] == -1
         assert sim_obs.duration_next_maintenance[4] == 0
-
-        # i have the right to reconnect it (if i simulate in the future)
-        act = self.env.action_space()
-        act.line_set_status = [(4, +1)]
-        sim_obs, reward, done, info = obs.simulate(act, time_step=1)
-        assert not info["is_illegal"]
-        assert sim_obs.line_status[4]
-        assert sim_obs.time_next_maintenance[4] == -1
-        assert sim_obs.duration_next_maintenance[4] == 0
-
+
         # i don't have the right to reconnect it if i don't simulate in the future
-        sim_obs, reward, done, info = obs.simulate(act, time_step=0)
+        sim_obs, reward, done, info = obs.simulate(reco_line, time_step=0)
         assert info["is_illegal"]
         assert not sim_obs.line_status[4]
         assert sim_obs.time_next_maintenance[4] == 0
         assert sim_obs.duration_next_maintenance[4] == 1
+
+        # I still have to wait 1 step before reconnection, so this raises
+        sim_obs, reward, done, info = obs.simulate(reco_line, time_step=1)
+        assert info["is_illegal"], f"there should be no error, but action is illegal"
+        assert not sim_obs.line_status[4]
+        assert sim_obs.time_next_maintenance[4] == -1
+        assert sim_obs.duration_next_maintenance[4] == 0
+
+        # at this step if I attempt a reco it fails
+        obs, reward, done, info = self.env.step(reco_line)
+        # maintenance will be over next time step
+        assert info["is_illegal"]
+        assert not obs.line_status[4]
+        assert obs.time_next_maintenance[4] == -1
+        assert obs.duration_next_maintenance[4] == 0
+
+        # I can reco the line next step
+        sim_obs, reward, done, info = obs.simulate(reco_line, time_step=1)
+        assert not info["is_illegal"], f"there should be no error, but action is illegal"
+        assert sim_obs.line_status[4]
+        assert sim_obs.time_next_maintenance[4] == -1
+        assert sim_obs.duration_next_maintenance[4] == 0
+
         # TODO be careful here, if the rules allows for reconnection, then the
         # TODO action becomes legal, and the powerline is reconnected
         # TODO => this is because the "_obs_env" do not attempt to force the disconnection

grid2op/tests/test_issue_374.py

@@ -0,0 +1,88 @@
+# Copyright (c) 2019-2022, RTE (https://www.rte-france.com)
+# See AUTHORS.txt and https://github.com/rte-france/Grid2Op/pull/319
+# This Source Code Form is subject to the terms of the Mozilla Public License, version 2.0.
+# If a copy of the Mozilla Public License, version 2.0 was not distributed with this file,
+# you can obtain one at http://mozilla.org/MPL/2.0/.
+# SPDX-License-Identifier: MPL-2.0
+# This file is part of Grid2Op, Grid2Op a testbed platform to model sequential decision making in power systems.
+
+import pdb
+import grid2op
+import unittest
+import warnings
+
+import re
+
+from grid2op.Parameters import Parameters
+
+
+class Issue367Tester(unittest.TestCase):
+    def setUp(self) -> None:
+        with warnings.catch_warnings():
+            warnings.filterwarnings("ignore")
+            # Creation of the environment
+            param = Parameters()
+            param.NB_TIMESTEP_COOLDOWN_SUB = 3
+            param.NB_TIMESTEP_COOLDOWN_LINE = 3
+            param.NO_OVERFLOW_DISCONNECTION = True
+            self.env = grid2op.make('l2rpn_wcci_2022', param=param)
+
+        self.env.set_id(0) 
+        self.env.seed(0)
+        self.obs = self.env.reset()
+
+    def test_cooldown(self):
+        # second, get some action that has no actual effect on the grid
+        action_space = self.env.action_space
+        env = self.env
+        do_nothing = action_space()
+
+        # take an action
+        set_sub_10 = action_space.get_all_unitary_topologies_set(action_space, sub_id=10)
+        set_sub10_to_bus1 = set_sub_10[5]
+
+        # do the action to trigger the cooldown
+        obs, _, done, info = env.step(set_sub10_to_bus1)
+        assert obs.time_before_cooldown_sub[10] == 3
+        assert info['exception'] == []
+
+        # check cooldown
+        obs, _, done, info = env.step(do_nothing)
+        assert obs.time_before_cooldown_sub[10] == 2
+        assert info['exception'] == []
+        # check i cannot simulate
+        obs_, _, done, info = obs.simulate(set_sub10_to_bus1)
+        assert  info['exception'] != [], "simulate should have raised an error"
+
+        # check cooldown
+        obs, _, done, info = env.step(do_nothing)
+        assert obs.time_before_cooldown_sub[10] == 1
+        assert info['exception'] == []
+
+        # check i cannot simulate
+        obs_, _, done, info = obs.simulate(set_sub10_to_bus1)
+        assert obs_.time_before_cooldown_sub[10] == 0
+        assert  info['exception'] != [], "simulate should have raised an error"
+        # check i cannot "step"
+        obs, _, done, info = env.step(set_sub10_to_bus1)
+        assert obs.time_before_cooldown_sub[10] == 0
+        assert info['exception'] != [], "step should have raised an error"
+
+
+        # check i can simulate
+        obs_, _, done, info = obs.simulate(set_sub10_to_bus1)
+        assert obs_.time_before_cooldown_sub[10] == 3
+        assert  info['exception'] == []
+        # check i can "step"
+        obs, _, done, info = env.step(set_sub10_to_bus1)
+        assert obs.time_before_cooldown_sub[10] == 3
+        assert info['exception'] == []
+
+
+    def tearDown(self) -> None:
+        self.env.close()
+        return super().tearDown()
+
+
+if __name__ == "__main__":
+    unittest.main()