fixing broken tests

Signed-off-by: DONNOT Benjamin <[email protected]>

Author: BDonnot

grid2op/Environment/_env_prev_state.py

@@ -6,10 +6,11 @@
 # SPDX-License-Identifier: MPL-2.0
 # This file is part of Grid2Op, Grid2Op a testbed platform to model sequential decision making in power systems.
 
-from typing import Optional, Type
+from typing import Optional, Type, Any
 import numpy as np
 from grid2op.Space import GridObjects
 import grid2op.Backend
+from grid2op.typing_variables import CLS_AS_DICT_TYPING
 from grid2op.Exceptions import Grid2OpException
 
 
@@ -26,7 +27,9 @@ def __init__(self,
                  init_shunt_q : np.ndarray,
                  init_shunt_bus : np.ndarray):
         self._can_modif = True
-        self._grid_obj_cls : Type[GridObjects] = grid_obj_cls
+        self._grid_obj_cls : CLS_AS_DICT_TYPING = grid_obj_cls.cls_to_dict()
+        self._n_storage = len(self._grid_obj_cls["name_storage"])  # to avoid typing that over and over again
+        
         self._load_p : np.ndarray = 1. * init_load_p
         self._load_q : np.ndarray = 1. * init_load_q
         self._gen_p : np.ndarray = 1. * init_gen_p
@@ -50,21 +53,21 @@ def update(self,
         if not self._can_modif:
             raise Grid2OpException(f"Impossible to modifiy this _EnvPreviousState")
 
-        self._aux_update(topo_vect[self._grid_obj_cls.load_pos_topo_vect],
+        self._aux_update(topo_vect[self._grid_obj_cls["load_pos_topo_vect"]],
                          self._load_p,
                          load_p,
                          self._load_q,
                          load_q)
-        self._aux_update(topo_vect[self._grid_obj_cls.gen_pos_topo_vect],
+        self._aux_update(topo_vect[self._grid_obj_cls["gen_pos_topo_vect"]],
                          self._gen_p,
                          gen_p,
                          self._gen_v,
                          gen_v)
         self._topo_vect[topo_vect > 0] = 1 * topo_vect[topo_vect > 0]
 
         # update storage units
-        if self._grid_obj_cls.n_storage > 0:
-            self._aux_update(topo_vect[self._grid_obj_cls.storage_pos_topo_vect],
+        if self._n_storage > 0:
+            self._aux_update(topo_vect[self._grid_obj_cls["storage_pos_topo_vect"]],
                             self._storage_p,
                             storage_p)
 
@@ -84,7 +87,7 @@ def update_from_backend(self,
         topo_vect = backend.get_topo_vect()
         load_p, load_q, *_ = backend.loads_info()
         gen_p, gen_q, gen_v = backend.generators_info()
-        if self._grid_obj_cls.n_storage > 0:
+        if self._n_storage > 0:
             storage_p, *_ = backend.storages_info()
         else:
             storage_p = None
@@ -109,7 +112,12 @@ def update_from_other(self,
                         "_shunt_p",
                         "_shunt_q",
                         "_shunt_bus"]:
-            getattr(self, attr_nm)[:] = getattr(other, attr_nm)
+            tmp = getattr(self, attr_nm)
+            if tmp.size > 1:
+                # works only for array of size 2 or more
+                tmp[:] = getattr(other, attr_nm)
+            else:
+                setattr(self, attr_nm, getattr(other, attr_nm))
 
     def prevent_modification(self):
         for attr_nm in ["_load_p",
@@ -121,7 +129,10 @@ def prevent_modification(self):
                         "_shunt_p",
                         "_shunt_q",
                         "_shunt_bus"]:
-            getattr(self, attr_nm).flags.writeable = False
+            tmp = getattr(self, attr_nm)
+            if tmp.size > 1:
+                # can't set flags on array of size 1 apparently
+                tmp.flags.writeable = False
         self._can_modif = False
 
     def _aux_update(self,

grid2op/Space/GridObjects.py

@@ -4159,7 +4159,7 @@ def _make_cls_dict_extended(cls, res: CLS_AS_DICT_TYPING, as_list=True, copy_=Tr
             cls._CLS_DICT_EXTENDED = res.copy()
 
     @classmethod
-    def cls_to_dict(cls):
+    def cls_to_dict(cls) -> CLS_AS_DICT_TYPING:
         """
         INTERNAL
 

grid2op/tests/test_simulate_disco_load.py

@@ -56,7 +56,7 @@ def test_backend_action(self):
         obs._obs_env._backend_action_set += self.env.action_space({"set_bus": {"loads_id": [(l_id, -1)]}})
         assert obs._obs_env._backend_action_set.current_topo.values[l_pos] == -1
         tmp = obs._obs_env._backend_action_set()  # do as if the action has been processed
-        assert not obs._obs_env._backend_action_set.load_p.changed[l_id]  # it's disconnected, so marked as unchanged now
+        assert obs._obs_env._backend_action_set.load_p.changed[l_id]  # it is not changed because disconnected
         assert np.allclose(obs._obs_env._backend_action_set.load_p.values[l_id], 22.3), f"{obs._obs_env._backend_action_set.load_p.values[l_id]:.2f} vs 22.3"
 
 

grid2op/tests/test_ts_handlers.py

@@ -13,6 +13,7 @@
 from grid2op.tests.helper_path_test import *
 
 import grid2op
+from grid2op.Agent import BaseAgent
 from grid2op.Exceptions import NoForecastAvailable
 from grid2op.Chronics import GridStateFromFileWithForecasts, GridStateFromFile, GridStateFromFileWithForecastsWithoutMaintenance, FromHandlers
 from grid2op.Chronics.handlers import (CSVHandler,
@@ -34,6 +35,20 @@
 # TODO check when there is also redispatching
 
 
+class TestAgent(BaseAgent):
+    def __init__(self, action_space, tester):
+        super().__init__(action_space)
+        self.tester = tester
+    def act(self, obs, reward, done=False):
+        # size of the forecast is always 12 even if it's "after" the size of the episode
+        self.tester._aux_test_obs(obs, max_it=12)
+        _ = self.tester.env.step(self.action_space())   # for TestPerfectForecastHandler: self.tester.env should be synch with the runner env...
+        return self.action_space()
+    def reset(self, obs):
+        self.tester.env.reset()  # for TestPerfectForecastHandler
+        return super().reset(obs)
+            
+            
 def _load_next_chunk_in_memory_hack(self):
     self._nb_call += 1
     # i load the next chunk as dataframes
@@ -512,7 +527,7 @@ def tearDown(self) -> None:
         return super().tearDown()  
 
     def _aux_test_obs(self, obs, max_it=12, tol=1e-5):
-        assert len(obs._forecasted_inj) == max_it + 1 # 12 + 1
+        assert len(obs._forecasted_inj) == max_it + 1, f"{len(obs._forecasted_inj)} vs {max_it + 1}"
         init_obj = obs._forecasted_inj[0]
         for for_h, el in enumerate(obs._forecasted_inj):
             for k_ in ["load_p", "load_q"]:
@@ -553,20 +568,7 @@ def test_copy(self):
             for k_ in ["load_p", "load_q", "prod_p"]:
                 assert np.all(el[1]["injection"][k_] == el_cpy[1]["injection"][k_])
 
-    def test_runner(self):
-        from grid2op.Agent import BaseAgent
-        class TestAgent(BaseAgent):
-            def __init__(self, action_space, tester):
-                super().__init__(action_space)
-                self.tester = tester
-            def act(self, obs, reward, done=False):
-                self.tester._aux_test_obs(obs, max_it=5 - obs.current_step)
-                _ = self.tester.env.step(self.action_space())   # for TestPerfectForecastHandler: self.tester.env should be synch with the runner env...
-                return self.action_space()
-            def reset(self, obs):
-                self.tester.env.reset()  # for TestPerfectForecastHandler
-                return super().reset(obs)
-            
+    def test_runner(self):            
         testagent = TestAgent(self.env.action_space, self)
         self.env.set_id(0)  # for TestPerfectForecastHandler
         runner = Runner(**self.env.get_params_for_runner(), agentClass=None, agentInstance=testagent)