adding test for N1 rewards for DC

lightsim2grid/rewards/n1ContingencyReward.py

@@ -133,6 +133,7 @@ def __call__(self, action, env, has_error, is_done, is_illegal, is_ambiguous):
         # synch the contingency analyzer
         # self._contingecy_analyzer.update_grid(self._backend_action)
         contingecy_analyzer = ContingencyAnalysis(self._backend)
+        contingecy_analyzer.computer.change_solver(self._solver_type)
         contingecy_analyzer.add_multiple_contingencies(*self._l_ids)
         now_ = time.perf_counter()
         self._timer_pre_proc += now_ - beg
@@ -142,16 +143,18 @@ def __call__(self, action, env, has_error, is_done, is_illegal, is_ambiguous):
         self._timer_compute += now_2 - now_
         if self._dc:
             # In DC is study p, but take into account q in the limits
-            res = tmp[0]  # this is Por
+            res = np.abs(tmp[0])  # this is Por
             # now transform the limits in A in MW
-            por, qor, aor, vor = env.backend.lines_or_info()
-            p_sq = (th_lim_a * np.sqrt(3.) * vor) - qor
+            por, qor, vor, aor = env.backend.lines_or_info()
+            p_sq = (1e-3*th_lim_a)**2 * 3. * vor**2 - qor**2
             p_sq[p_sq <= 0.] = 0.
             limits = np.sqrt(p_sq)
         else:
             res = tmp[1]
             limits = th_lim_a
-
+        # print("Reward:")
+        # print(res)
+        # print(self._threshold_margin * limits)
         res = ((res > self._threshold_margin * limits) | (~np.isfinite(res))).any(axis=1)  # whether one powerline is above its limit, per cont
         # print(res.nonzero())
         # import pdb

lightsim2grid/tests/test_n1contingencyrewards.py

@@ -7,6 +7,7 @@
 # This file is part of LightSim2grid, LightSim2grid a implements a c++ backend targeting the Grid2Op platform.
 import unittest
 import warnings
+import copy
 import numpy as np
 
 import grid2op
@@ -54,7 +55,7 @@ def threshold_margin(self):
         return 1.
 
     def l_ids(self):
-        # return [0]
+        return [0]
         return None
 
     def setUp(self) -> None:
@@ -76,6 +77,10 @@ def setUp(self) -> None:
         params.NB_TIMESTEP_COOLDOWN_LINE = 0
         params.NB_TIMESTEP_COOLDOWN_SUB = 0
         self.env.change_parameters(params)
+        if self.is_dc():
+            params = copy.deepcopy(params)
+            params.ENV_DC = True
+            params.FORECAST_DC = True
         self.env.change_forecast_parameters(params)
         assert (self.env.get_thermal_limit() == TH_LIM_A_REF).all()
         self.my_ids = self.l_ids()
@@ -90,11 +95,33 @@ def tearDown(self) -> None:
 
     def _aux_test_reward(self, obs, reward):
         unsafe_cont = 0
+        if self.is_dc():
+            th_lim = obs._thermal_limit
+            # A = sqrt(p**2 + q**2) / (sqrt3) * v)
+            # A**2 = (p**2 + q**2) / (3. v**2)
+            # p**2 = A**2 * 3. * v**2 - q**2
+            # and now the units...
+            # W**2 = A * 3. * V**2 - VAr**2
+            # MW**2 = 1e12 * A**2 * 3. * v**2 - 1e12 * VAr**2
+            # MW**2 = 1e6*A**2 * 3. * 1e6*(V)**2 *  - MVAr**2
+            # MW**2 = kA**2 * 3. * kV**2 - MVAr**2
+            p_square = 3. * (1e-3*th_lim)**2 * (obs.v_or)**2 - (obs.q_or)**2
+            p_square[p_square <= 0.] = 0.
+            th_lim_p = np.sqrt(p_square)
+
+        # print("test:")
         for l_id in self.my_ids:
             sim_obs, sim_r, sim_d, sim_i = obs.simulate(self.env.action_space({"set_line_status": [(l_id, -1)]}),
                                                         time_step=0)
-            if np.any(sim_obs.a_or > obs._thermal_limit) or sim_d:
-                unsafe_cont += 1       
+            if not self.is_dc():
+                if np.any(sim_obs.a_or > obs._thermal_limit) or sim_d:
+                    unsafe_cont += 1       
+            else:
+                # print(sim_obs.p_or)
+                # print(th_lim_p)
+                if np.any(np.abs(sim_obs.p_or) > th_lim_p) or sim_d:
+                    unsafe_cont += 1       
+
         assert reward == (len(self.my_ids) - unsafe_cont), f"{reward} vs {(len(self.my_ids) - unsafe_cont)}"
 
     def test_do_nothing(self):
@@ -154,6 +181,10 @@ def test_copy(self):
         assert reward != reward_cpy
 
 
+class TestN1ContingencyReward_DC(TestN1ContingencyReward_Base):
+    def is_dc(self):
+        return True
+
 # TODO test with only a subset of powerlines
 # TODO test with the "margin"
 # TODO test with pandapower and lightsim as base backend