Skip to content

Commit

Permalink
Adding type hints for backend classes and examples (#532)
Browse files Browse the repository at this point in the history
  • Loading branch information
BDonnot authored Sep 21, 2023
1 parent 808adcb commit f448a09
Show file tree
Hide file tree
Showing 11 changed files with 276 additions and 196 deletions.
5 changes: 5 additions & 0 deletions CHANGELOG.rst
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,11 @@ Change Log
- [???] "asynch" multienv
- [???] properly model interconnecting powerlines

[1.9.6] - 2023-xx-yy
----------------------
- [ADDED] now depends on the `typing_extensions` package
- [IMPROVED] type hints for Backend and PandapowerBackend

[1.9.5] - 2023-09-18
---------------------
- [FIXED] issue https://github.com/rte-france/Grid2Op/issues/518
Expand Down
20 changes: 12 additions & 8 deletions examples/backend_integration/Step1_loading.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,14 +23,18 @@
import os
import numpy as np
import grid2op
from typing import Optional, Tuple, Union

from grid2op.Backend import Backend # required

# to serve as an example
import pandapower as pp


class CustomBackend_Step1(Backend):
def load_grid(self, path, filename=None):
def load_grid(self,
path : Union[os.PathLike, str],
filename : Optional[Union[os.PathLike, str]]=None) -> None:
# first load the grid from the file
full_path = path
if filename is not None:
Expand Down Expand Up @@ -92,25 +96,25 @@ def load_grid(self, path, filename=None):

self._compute_pos_big_topo()

def apply_action(self, action):
def apply_action(self, backendAction: Union["grid2op.Action._backendAction._BackendAction", None]) -> None:
raise NotImplementedError("Will be detailed in another example script")

def runpf(self, is_dc=False):
def runpf(self, is_dc : bool=False) -> Tuple[bool, Union[Exception, None]]:
raise NotImplementedError("Will be detailed in another example script")

def get_topo_vect(self):
def get_topo_vect(self) -> np.ndarray:
raise NotImplementedError("Will be detailed in another example script")

def generators_info(self):
def generators_info(self)-> Tuple[np.ndarray, np.ndarray, np.ndarray]:
raise NotImplementedError("Will be detailed in another example script")

def loads_info(self):
def loads_info(self)-> Tuple[np.ndarray, np.ndarray, np.ndarray]:
raise NotImplementedError("Will be detailed in another example script")

def lines_or_info(self):
def lines_or_info(self)-> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
raise NotImplementedError("Will be detailed in another example script")

def lines_ex_info(self):
def lines_ex_info(self)-> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
raise NotImplementedError("Will be detailed in another example script")


Expand Down
12 changes: 7 additions & 5 deletions examples/backend_integration/Step2_modify_load.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,21 +15,23 @@
"""
import numpy as np
import pandapower as pp
from typing import Optional, Tuple, Union

from Step1_loading import CustomBackend_Step1


class CustomBackend_Step2(CustomBackend_Step1):
def apply_action(self, action):
def apply_action(self, backendAction: Union["grid2op.Action._backendAction._BackendAction", None]) -> None:
# the following few lines are highly recommended
if action is None:
if backendAction is None:
return

(
active_bus,
(prod_p, prod_v, load_p, load_q, storage),
_,
shunts__,
) = action()
) = backendAction()

# change the active values of the loads
for load_id, new_p in load_p:
Expand All @@ -38,7 +40,7 @@ def apply_action(self, action):
for load_id, new_q in load_q:
self._grid.load["q_mvar"].iloc[load_id] = new_q

def runpf(self, is_dc=False):
def runpf(self, is_dc : bool=False) -> Tuple[bool, Union[Exception, None]]:
# possible implementation of the runpf function
try:
if is_dc:
Expand All @@ -50,7 +52,7 @@ def runpf(self, is_dc=False):
# of the powerflow has not converged, results are Nan
return False, exc_

def loads_info(self):
def loads_info(self)-> Tuple[np.ndarray, np.ndarray, np.ndarray]:
# carefull with copy / deep copy
load_p = self._grid.res_load["p_mw"].values # in MW
load_q = self._grid.res_load["q_mvar"].values # in MVAr
Expand Down
12 changes: 7 additions & 5 deletions examples/backend_integration/Step3_modify_gen.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,24 +18,26 @@
"""
import numpy as np
from typing import Optional, Tuple, Union

from Step2_modify_load import CustomBackend_Step2


class CustomBackend_Step3(CustomBackend_Step2):
def apply_action(self, action):
def apply_action(self, backendAction: Union["grid2op.Action._backendAction._BackendAction", None]) -> None:
# the following few lines are highly recommended
if action is None:
if backendAction is None:
return

# loads are modified in the previous script
super().apply_action(action)
super().apply_action(backendAction)

(
active_bus,
(prod_p, prod_v, load_p, load_q, storage),
_,
shunts__,
) = action()
) = backendAction()

# change the active value of generators
for gen_id, new_p in prod_p:
Expand All @@ -49,7 +51,7 @@ def apply_action(self, action):
self.gen_to_subid[gen_id]
] # now it is :-)

def generators_info(self):
def generators_info(self)-> Tuple[np.ndarray, np.ndarray, np.ndarray]:
prod_p = self._grid.res_gen["p_mw"].values # in MW
prod_q = self._grid.res_gen["q_mvar"].values # in MVAr

Expand Down
15 changes: 8 additions & 7 deletions examples/backend_integration/Step4_modify_line_status.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,18 +18,19 @@
"""
import numpy as np
from typing import Optional, Tuple, Union

from Step3_modify_gen import CustomBackend_Step3


class CustomBackend_Step4(CustomBackend_Step3):
def apply_action(self, action):
def apply_action(self, backendAction: Union["grid2op.Action._backendAction._BackendAction", None]) -> None:
# the following few lines are highly recommended
if action is None:
if backendAction is None:
return

# loads and generators are modified in the previous script
super().apply_action(action)
super().apply_action(backendAction)

# disconnected powerlines are indicated because they are
# connected to bus "-1" in the `get_lines_or_bus()` and
Expand All @@ -46,7 +47,7 @@ def apply_action(self, action):
n_line_pp = self._grid.line.shape[0]

# handle the disconnection on "or" side
lines_or_bus = action.get_lines_or_bus()
lines_or_bus = backendAction.get_lines_or_bus()
for line_id, new_bus in lines_or_bus:
if line_id < n_line_pp:
# a pandapower powerline has bee disconnected in grid2op
Expand All @@ -64,7 +65,7 @@ def apply_action(self, action):
# element was connected
dt["in_service"].iloc[line_id_db] = True

lines_ex_bus = action.get_lines_ex_bus()
lines_ex_bus = backendAction.get_lines_ex_bus()
for line_id, new_bus in lines_ex_bus:
if line_id < n_line_pp:
# a pandapower powerline has bee disconnected in grid2op
Expand Down Expand Up @@ -95,7 +96,7 @@ def _aux_get_line_info(self, colname_powerline, colname_trafo):
)
return res

def lines_or_info(self):
def lines_or_info(self)-> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
"""
Main method to retrieve the information at the "origin" side of the powerlines and transformers.
Expand Down Expand Up @@ -134,7 +135,7 @@ def lines_or_info(self):
v_or[~status] = 0.
return p_or, q_or, v_or, a_or

def lines_ex_info(self):
def lines_ex_info(self)-> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
"""
Main method to retrieve the information at the "extremity" side of the powerlines and transformers.
Expand Down
17 changes: 9 additions & 8 deletions examples/backend_integration/Step5_modify_topology.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import copy
import pandas as pd
import numpy as np
from typing import Optional, Tuple, Union

from Step4_modify_line_status import CustomBackend_Step4

Expand Down Expand Up @@ -55,17 +56,17 @@ def _aux_change_bus_or_disconnect(self, new_bus, dt, key, el_id):
# grid2op built-in "***_global()" functions that allows to retrieve the global id
# (from 0 to n_total_bus-1) directly (instead of manipulating local bus id that
# are either 1 or 2)
def apply_action(self, action):
def apply_action(self, backendAction: Union["grid2op.Action._backendAction._BackendAction", None]) -> None:
# the following few lines are highly recommended
if action is None:
return

# loads and generators are modified in the previous script
super().apply_action(action)
super().apply_action(backendAction)

# handle the load (see the comment above the definition of this
# function as to why it's possible to use the get_loads_bus_global)
loads_bus = action.get_loads_bus_global()
loads_bus = backendAction.get_loads_bus_global()
for load_id, new_bus in loads_bus:
self._aux_change_bus_or_disconnect(new_bus,
self._grid.load,
Expand All @@ -74,7 +75,7 @@ def apply_action(self, action):

# handle the generators (see the comment above the definition of this
# function as to why it's possible to use the get_loads_bus_global)
gens_bus = action.get_gens_bus_global()
gens_bus = backendAction.get_gens_bus_global()
for gen_id, new_bus in gens_bus:
self._aux_change_bus_or_disconnect(new_bus,
self._grid.gen,
Expand All @@ -85,7 +86,7 @@ def apply_action(self, action):
# function as to why it's possible to use the get_lines_or_bus_global)
n_line_pp = self._grid.line.shape[0]

lines_or_bus = action.get_lines_or_bus_global()
lines_or_bus = backendAction.get_lines_or_bus_global()
for line_id, new_bus in lines_or_bus:
if line_id < n_line_pp:
dt = self._grid.line
Expand All @@ -101,7 +102,7 @@ def apply_action(self, action):
key,
line_id_pp)

lines_ex_bus = action.get_lines_ex_bus_global()
lines_ex_bus = backendAction.get_lines_ex_bus_global()
for line_id, new_bus in lines_ex_bus:
if line_id < n_line_pp:
dt = self._grid.line
Expand All @@ -126,7 +127,7 @@ def apply_action(self, action):
(prod_p, prod_v, load_p, load_q, storage),
_,
shunts__,
) = action()
) = backendAction()
bus_is = self._grid.bus["in_service"]
for i, (bus1_status, bus2_status) in enumerate(active_bus):
bus_is[i] = bus1_status
Expand All @@ -149,7 +150,7 @@ def _aux_get_topo_vect(self, res, dt, key, pos_topo_vect, add_id=0):
el_id += 1

# it should return, in the correct order, on which bus each element is connected
def get_topo_vect(self):
def get_topo_vect(self) -> np.ndarray:
res = np.full(self.dim_topo, fill_value=-2, dtype=int)
# read results for load
self._aux_get_topo_vect(res, self._grid.load, "bus", self.load_pos_topo_vect)
Expand Down
Loading

0 comments on commit f448a09

Please sign in to comment.