# Copyright (c) 2020-2024, RTE (https://www.rte-france.com)
# See AUTHORS.txt
# 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 LightSim2grid, LightSim2grid implements a c++ backend targeting the Grid2Op platform.
import os
import copy
from typing import Any, Dict, Tuple, Optional, Union
from packaging import version
# NOT `from lightsim2grid.solver import SolverType`: lightsim2grid.solver is itself
# deprecated and warns unconditionally on import, which would make every
# `import lightsim2grid` (which imports this module) print that warning regardless of
# whether solver_type / SolverType is ever used. SolverType itself is not deprecated to
# reference for back-compat bridging, so it lives in this private module instead. It is
# deliberately a plain top-level module and NOT inside lightsim2grid._utils: see
# lightsim2grid/_solver_type.py for why (importing it eagerly, from here, while nested
# inside that subpackage broke grid2op.make(..., backend=LightSimBackend()) entirely by
# reentering lightsim2grid._utils's own grid2op.Backend import mid-circular-import).
from lightsim2grid._solver_type import SolverType
try:
from typing import Self
except ImportError:
# python version is probably bellow 3.11
from typing_extensions import Self
import warnings
import numpy as np
import pandas as pd
import time
import grid2op
from grid2op.Action import CompleteAction
from grid2op.Backend import Backend
from grid2op.Exceptions import BackendError, Grid2OpException
from grid2op.dtypes import dt_float, dt_int, dt_bool
try:
from grid2op.Action._backendAction import _BackendAction
except ImportError as exc_: # noqa: F841
from grid2op.Action._BackendAction import _BackendAction
try:
from grid2op.Space import DEFAULT_N_BUSBAR_PER_SUB
except ImportError:
# for backward compatibility with grid2op <= 1.9.8
DEFAULT_N_BUSBAR_PER_SUB = 2
try:
from grid2op.Space import DEFAULT_ALLOW_DETACHMENT
except ImportError:
# for backward compatibility with grid2op < 1.11.0
DEFAULT_ALLOW_DETACHMENT = False
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal
from lightsim2grid.algorithm import AlgorithmType, AlgoConfig
LOADER_KWARGS_TYPING = Dict[str, Any] # TODO improve this
grid2op_min_cls_attr_ver = version.parse("1.6.4")
grid2op_min_shunt_cls_properly_handled = version.parse("1.9.7")
[docs]
class LightSimBackend(Backend):
"""
This is a specialization of the grid2op Backend class to use the lightsim2grid solver,
coded in c++, aiming at speeding up the computations.
"""
shunts_data_available = True
glop_version = Backend.glop_version if hasattr(Backend, "glop_version") else grid2op.__version__
if not hasattr(Backend, "n_busbar_per_sub"):
# for legacy grid2op
n_busbar_per_sub = DEFAULT_N_BUSBAR_PER_SUB
if not hasattr(Backend, "detachment_is_allowed"):
# for legacy grid2op (< 1.11.0)
detachment_is_allowed = DEFAULT_ALLOW_DETACHMENT
KEYS_PYPOWSYBL_LOADER = {
# "gen_slack_id",
"pypowsybl_load_kwargs",
"grid",
"use_buses_for_sub",
"sort_index",
"init_vm_pu",
"sn_mva",
"double_bus_per_sub",
"use_grid2op_default_names",
"reconnect_disco_gen",
"reconnect_disco_load",
"n_busbar_per_sub"
}
KEYS_PANDAPOWER_LOADER = {
"pp_orig_file",
# "grid"
}
KEYS_MATPOWER_LOADER = {
"grid",
"n_busbar_per_sub",
}
def __init__(self,
detailed_infos_for_cascading_failures: bool=False,
can_be_copied: bool=True,
max_iter: int=10,
tol: float=1e-8,
solver_type: Optional[AlgorithmType]=None,
algo_type: Optional[Union[AlgorithmType, str]]=None,
turned_off_pv : bool=True, # are gen turned off (or with p=0) contributing to voltage or not
dist_slack_non_renew: bool=False, # distribute the slack on non renewable turned on (and with P>0) generators
use_static_gen: bool=False, # add the static generators as generator gri2dop side
loader_method: Literal["pandapower", "pypowsybl", "matpower"] = "pandapower",
loader_kwargs : LOADER_KWARGS_TYPING= None,
stop_if_load_disco : Optional[bool] = None,
stop_if_gen_disco : Optional[bool] = None,
stop_if_storage_disco : Optional[bool] = None,
automatically_disconnect : bool = False,
gen_slack_id=None,
):
#: ``int`` maximum number of iteration allowed for the solver
#: if the solver has not converge after this, it will
#: send a "divergence error"
self.max_it = max_iter
#: ``float`` tolerance of the solver
self.tol = tol # tolerance for the solver
algo_type = self._aux_merge_solver_type_algo_type(solver_type, algo_type)
real_algo_type = self._check_suitable_solver_type(algo_type, check_in_avail_solver=False)
self.__current_algo_type = real_algo_type
#: :class:`~lightsim2grid.algorithm.AlgoConfig` set through :func:`set_ac_algo_config`
#: / :func:`set_dc_algo_config` (``None`` if never customized), re-applied after
#: every `env.reset()` and preserved by `backend.copy()`.
self.__current_ac_algo_config = None
self.__current_dc_algo_config = None
#: does the "turned off" generators (including when p=0)
#: are pv buses
self._turned_off_pv = turned_off_pv
#: distributed slack, on non renewable gen with P > 0
self._dist_slack_non_renew = dist_slack_non_renew
# add the static gen to the list of controlable gen in grid2Op
self._use_static_gen = use_static_gen # TODO implement it
#: For now, you can initialize a "lightsim2grid" LightsimBackend
#: from pandapower, from pypowsybl or from a MATPOWER case file.
#: Use with `LightsimBackend(..., loader_method="pypowsybl")`,
#: `LightsimBackend(..., loader_method="matpower")`
#: or `LightsimBackend(..., loader_method="pandapower")` (default)
#:
#: .. versionchanged:: 1.0.1
#: added the `"matpower"` loader (reads the environment `grid.m`
#: or `grid.mat` file directly)
self._loader_method = loader_method
#: Which key-word arguments will be used to initialize the Gridmodel
#: either from pandapower or lightsim2grid.
#: It is not currently used for pandapower.
#:
#: For pypowsybl it can contain the following keys:
#:
#: - `grid` (``pypowsybl.network.Network``): the pypowsybl
#: grid that will be used (thus skipping the call to pypowsybl.network.load).
#: So the "path", "filename" of grid2op.make and "pypowsybl_load_kwargs"
#: will be ignored if this is set.
#: - `pypowsybl_load_kwargs` (``dict``): addition keywords arguments
#: passed to ``pypowsybl.network.load(grid, **pypowsybl_load_kwargs)``
#: Not used when grid is provided.
#: - `n_busbar_per_sub` (``int``): number of independant buses for
#: each substation in the GridModel.
#: - `use_buses_for_sub` (``bool``): whether to use the buses (in the
#: pypowsybl Network) to initialize the "substation" in the lightsim2grid
#: Gridmodel (if ``True``). If ``False`` it will use the `voltage_levels`
#: of the pypowsybl Network.
#: - `use_grid2op_default_names` (``bool``): whether to use the "default names"
#: assigne by grid2op or to read them from the the iidm grid.
#: - `reconnect_disco_gen` (``bool``): whether to reconnect the disconnected
#: generators in the iidm grid. If set to ``True`` then the generators will be
#: reconnected with a p setpoint of 0. MW and a voltage setpoint of 1 pu
#: - `reconnect_disco_load` (``bool``): whether to reconnec the disconnected
#: load from in the iidm grid. If reconnected, load will have a target
#: p set to 0. and a target q set to 0.
#: - `sort_index`
#: - `init_vm_pu`
#: - `sn_mva`
#:
#: For matpower it can contain the following keys:
#:
#: - `grid` (``dict``): an already parsed matpower case (anything
#: :func:`lightsim2grid.network.init_from_matpower` accepts as its
#: `source`), used instead of reading the "path" / "filename" given
#: to `grid2op.make`.
#: - `n_busbar_per_sub` (``int``): number of independant buses for
#: each substation in the GridModel (same meaning as for pypowsybl).
#: There is no `double_bus_per_sub` here: that key predates grid2op
#: supporting any number of busbars per substation, and only the
#: pypowsybl loader keeps it for backward compatibility.
#:
self._loader_kwargs :LOADER_KWARGS_TYPING = loader_kwargs
#: .. versionadded:: 0.8.0
#:
#: if set to ``True`` (default) then the backend will raise a
#: BackendError in case of disconnected load
#:
#: .. note:: When using grid2op >= 1.11.0 and lightsim2grid >= 0.10.0 this
#: is automatically set-up with the call to grid2op.make,
#: to match the behaviour expected by `allow_detachment`
self._stop_if_load_disco = stop_if_load_disco
#: .. versionadded:: 0.8.0
#:
#: if set to ``True`` (default) then the backend will raise a
#: BackendError in case of disconnected generator
#:
#: .. note:: When using grid2op >= 1.11.0 and lightsim2grid >= 0.10.0 this
#: is automatically set-up with the call to grid2op.make,
#: to match the behaviour expected by `allow_detachment`
self._stop_if_gen_disco = stop_if_gen_disco
#: .. versionadded:: 0.10.0
#:
#: if set to ``True`` (default) then the backend will raise a
#: BackendError in case of disconnected storage that are
#: asked to produce / absorb something
#:
#: .. note:: When using grid2op >= 1.11.0 and lightsim2grid >= 0.10.0 this
#: is automatically set-up with the call to grid2op.make,
#: to match the behaviour expected by `allow_detachment`
self._stop_if_storage_disco = stop_if_storage_disco
#: .. versionadded:: 0.11.0
#:
#: if set to ``True`` the backend will automatically
#: disconnect any load / generator not in the main
#: connected component. The default is ``False``
#: which means it will NOT disconnect them
#: automatically and raise a "non connected grid"
#: error (this should only be used if grid2op allows detachment
#: and the detachment is allowed in the env configuration)
self._automatically_disconnect = automatically_disconnect
self._aux_init_super(detailed_infos_for_cascading_failures,
can_be_copied,
algo_type,
max_iter,
tol,
turned_off_pv,
dist_slack_non_renew,
use_static_gen,
loader_method,
loader_kwargs,
stop_if_load_disco,
stop_if_gen_disco,
stop_if_storage_disco,
automatically_disconnect,
gen_slack_id)
#: whether or not to run a routine to detect the islanding
#: before computing the powerflow (this routine should
#: automatically switch off element not in the main CC)
#: by default it's False and only set to True if
#: automatically_disconnect is True (and if the environment
#: supports detachment)
self._need_islanding_detection = False
# backward compat: need to define it if not done by grid2op
if not hasattr(self, "_can_be_copied"):
self._can_be_copied = can_be_copied
#: I can output the voltage angle
self.can_output_theta = True
#: .. versionadded:: 0.8.0
#:
#: Which type of grid format can be read by your backend.
#: It is "json" if loaded from pandapower,
#: "xiidm" if loaded from pypowsybl or
#: "m" / "mat" if loaded from matpower.
self.supported_grid_format = None
if loader_method == "pandapower":
self.supported_grid_format = ("json", ) # new in 0.8.0
elif loader_method == "pypowsybl":
self.supported_grid_format = ("xiidm", ) # new in 0.8.0
elif loader_method == "matpower":
# a matpower case comes either as the ".m" script it is distributed as,
# or as the ".mat" binary matpower itself saves. `init_from_matpower`
# reads both, so an environment can ship either one as its "grid" file.
self.supported_grid_format = ("m", "mat", ) # new in 1.0.1
else:
raise BackendError(f"Uknown loader_method : '{loader_method}'")
# lazy loading because it crashes...
from grid2op.Space import GridObjects # lazy import
self.__has_storage = hasattr(GridObjects, "n_storage")
if not self.__has_storage:
warnings.warn("Please upgrade your grid2op to >= 1.5.0. You are using a backward compatibility "
"feature that will be removed in further lightsim2grid version.")
if version.parse(grid2op.__version__) < grid2op_min_shunt_cls_properly_handled:
warnings.warn(f"You are using a legacy grid2op version. It is not possible to deactivate the shunts in lightsim2grid. "
f"Please upgrade to grid2op >= {grid2op_min_shunt_cls_properly_handled}")
self.shunts_data_available = True # needs to be self and not type(self) here
type(self).shunts_data_available = True
self.nb_bus_total = None
self.initdc = True # does not really hurt computation time
self.__nb_powerline = None
self.__nb_bus_before = None
self._init_bus_load = None
self._init_bus_gen = None
self._init_bus_lor = None
self._init_bus_lex = None
self._big_topo_to_obj = None
self.nb_obj_per_bus = None
self._timer_preproc = 0.
self._timer_postproc = 0.
self._timer_solver = 0.
self._timer_read_data_back = 0.
self._timer_fetch_data_cpp = 0.
self._timer_apply_act = 0.
self.next_prod_p = None # this vector is updated with the action that will modify the environment
# it is done to keep track of the redispatching
self.topo_vect = None
self.shunt_topo_vect = None
#: the initial pandapower backend (if loaded from pandapower)
self._init_pp_backend = None
#: the initial pypowsybl grid (if loaded from pypowsybl)
self._orig_grid_pypowsybl = None
#: .. versionadded:: 1.0.1
#:
#: Whether the loader in charge already told the :class:`lightsim2grid.network.LSGrid`
#: how its buses are grouped into substations / voltage levels.
#:
#: `init_from_pypowsybl` and `init_from_matpower` both build the grid substation
#: by substation (``init_bus(n_sub, n_busbar_per_sub, ...)``), so the substation
#: count and the number of busbar sections per substation are already right and
#: must not be overwritten by the backend. The pandapower converter, on the other
#: hand, is handed a bus table and no substation notion at all: grid2op's own
#: pandapower backend is what says how many substations there are.
self._grid_laid_out_by_loader = False
self.V = None
self.prod_pu_to_kv = None
self.load_pu_to_kv = None
self.lines_or_pu_to_kv = None
self.lines_ex_pu_to_kv = None
self.storage_pu_to_kv = None
self.p_or = None
self.q_or = None
self.v_or = None
self.a_or = None
self.p_ex = None
self.q_ex = None
self.v_ex = None
self.a_ex = None
self.load_p = None
self.load_q = None
self.load_v = None
self.prod_p = None
self.prod_q = None
self.prod_v = None
self.storage_p = None
self.storage_q = None
self.storage_v = None
# shunts
self.sh_p = None
self.sh_q = None
self.sh_v = None
self.sh_theta = None
self.sh_bus = None
# voltage angle
self.line_or_theta = None
self.line_ex_theta = None
self.load_theta = None
self.gen_theta = None
self.storage_theta = None
self.thermal_limit_a = None
self.dim_topo = -1
# self._init_action_to_set = None
self._backend_action_class = None
self.cst_1 = dt_float(1.0)
self.__me_at_init = None
self.__init_topo_vect = None
self.__init_shunt_bus = None
# available solver in lightsim
self.available_default_algorithms = []
# computation time of just the powerflow (when the grid is formatted
# by the gridmodel already)
# it takes only into account the time spend in the powerflow algorithm
self.comp_time = 0.
#: computation time of the powerflow
#: it takes into account everything in the gridmodel, including the mapping
#: to the solver, building of Ybus and Sbus AND the time to solve the powerflow
self.timer_gridmodel_xx_pf = 0.
self._timer_postproc = 0.
self._timer_preproc = 0.
self._timer_solver = 0.
self._timer_read_data_back = 0.
self._timer_fetch_data_cpp = 0.
self._timer_apply_act = 0.
# hack for the storage unit:
# in grid2op, for simplicity, I suppose that if a storage is alone on a busbar, and
# that it produces / absorbs nothing, then that's fine
# this behaviour in lightsim (c++ side) would be detected as a non connex grid and raise
# a diverging powerflow
# i "fake" to disconnect storage with these properties
# TODO hummm we need to clarify that ! pandapower automatically disconnect this stuff too ! This is super weird
# TODO and should rather be handled in pandapower backend
# backend SHOULD not do these kind of stuff
self._idx_hack_storage = []
#: ..versionadded: 0.9.3
#: sometimes some actions will make the grid fails
#: but grid2op expect it to fail not on "apply_action"
#: but rather when calling `runpf`
#: this flags remembers it
self._next_pf_fails : Optional[BackendError] = None
#: .. versionadded:: 0.11.1
#:
#: generator to assign to the slack
#: for now only used for pypowsybl
self._gen_slack_id = gen_slack_id
if self._gen_slack_id is not None and self._dist_slack_non_renew:
raise RuntimeError("You choose the generators participating to the slack (using `gen_slack_id` kwargs), "
"and at the same time you told lightsim2grid that slack is on "
"on non-renewable generators (using `dist_slack_non_renew` kwargs). "
"Chose one or the other, but not both.")
if self._gen_slack_id is not None and self._loader_method != "pypowsybl":
raise RuntimeError("Choosing the generators in the slak is only possible when "
"loading a grid from pypowsybl at the moment. Please fill a feature request "
"(github issue) if this feature interest you.")
# speed optimization
self._lineor_res = None
self._lineex_res = None
self._load_res = None
self._gen_res = None
self._shunt_res = None
self._trafo_hv_res = None
self._trafo_lv_res = None
self._storage_res = None
self._reset_res_pointers()
self._debug_Vdc = None # use only for debug !
def _aux_assign_init_pp_backend(self):
from lightsim2grid._utils import _DoNotUseAnywherePandaPowerBackend
_DoNotUseAnywherePandaPowerBackend._clear_grid_dependant_class_attributes()
try:
self._init_pp_backend = _DoNotUseAnywherePandaPowerBackend(with_numba=False)
except TypeError as exc_:
# oldest version of grid2op do not support the kwargs "with_numba"
# (before 1.9.1)
self._init_pp_backend = _DoNotUseAnywherePandaPowerBackend()
def _aux_init_super(self,
detailed_infos_for_cascading_failures,
can_be_copied,
algo_type,
max_iter,
tol,
turned_off_pv,
dist_slack_non_renew,
use_static_gen,
loader_method,
loader_kwargs,
stop_if_load_disco,
stop_if_gen_disco,
stop_if_storage_disco,
automatically_disconnect,
gen_slack_id):
try:
# for grid2Op >= 1.7.1
Backend.__init__(self,
detailed_infos_for_cascading_failures=detailed_infos_for_cascading_failures,
can_be_copied=can_be_copied,
algo_type=algo_type,
max_iter=max_iter,
tol=tol,
turned_off_pv=turned_off_pv,
dist_slack_non_renew=dist_slack_non_renew,
use_static_gen=use_static_gen,
loader_method=loader_method,
loader_kwargs=loader_kwargs,
stop_if_load_disco=stop_if_load_disco,
stop_if_gen_disco=stop_if_gen_disco,
stop_if_storage_disco=stop_if_storage_disco,
automatically_disconnect=automatically_disconnect,
gen_slack_id=gen_slack_id,
)
except TypeError as exc_:
warnings.warn("Please use grid2op >= 1.7.1: with older grid2op versions, "
"you cannot set max_iter, tol nor algo_type arguments.")
Backend.__init__(self,
detailed_infos_for_cascading_failures=detailed_infos_for_cascading_failures)
cls = type(self)
if hasattr(cls, "can_handle_more_than_2_busbar"):
# do not forget to propagate this if needed
self.can_handle_more_than_2_busbar()
if hasattr(cls, "can_handle_detachment"):
# do not forget to propagate this if needed
self.can_handle_detachment()
if automatically_disconnect:
# authorize disconnection of elements in the backend
self._prevent_automatic_disconnection = False
def turnedoff_no_pv(self):
self._turned_off_pv = False
self._grid.turnedoff_no_pv()
def turnedoff_pv(self):
self._turned_off_pv = True
self._grid.turnedoff_pv()
[docs]
def get_theta(self) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
"""
Returns
-------
line_or_theta: ``numpy.ndarray``
For each orgin side of powerline, gives the voltage angle
line_ex_theta: ``numpy.ndarray``
For each extremity side of powerline, gives the voltage angle
load_theta: ``numpy.ndarray``
Gives the voltage angle to the bus at which each load is connected
gen_theta: ``numpy.ndarray``
Gives the voltage angle to the bus at which each generator is connected
storage_theta: ``numpy.ndarray``
Gives the voltage angle to the bus at which each storage unit is connected
"""
return self.cst_1 * self.line_or_theta, \
self.cst_1 * self.line_ex_theta, \
self.cst_1 * self.load_theta, \
self.cst_1 * self.gen_theta, \
self.cst_1 * self.storage_theta
[docs]
def get_algo_types(self) -> Tuple[AlgorithmType, AlgorithmType]:
"""Return the types of solver that are used in the form a tuple with 2 elements.
The first one is the solver used for AC computation, the second one for DC computation (and also for
initialization of AC computations)
You can use it to check what is used:
.. code-block:: python
import grid2op
import lightsim2grid
from lightsim2grid import LightSimBackend
env_name = ...
env = grid2op.make(env_name, backend=LightSimBackend())
print(env.backend.get_algo_types())
# >>> (<AlgorithmType.NRSing_KLU: 7>, <AlgorithmType.DC_KLU: 9>) [can depend on your installation of lightsim2grid]
env2 = grid2op.make(env_name, backend=LightSimBackend(algo_type=lightsim2grid.algorithm.AlgorithmType.NR_SparseLU))
print(env2.backend.get_algo_types())
# >>> (<AlgorithmType.NR_SparseLU: 0>, <AlgorithmType.DC_KLU: 9>) [can depend on your installation of lightsim2grid]
"""
return self._grid.get_algo_type(), self._grid.get_dc_algo_type()
[docs]
def set_solver_type(self, algo_type: AlgorithmType) -> None:
"""DEPRECATED use :func:`set_algo_type` instead"""
self.set_algo_type(algo_type)
[docs]
def set_algo_type(self, algo_type: Union[AlgorithmType, str]) -> None:
"""
Change the type of solver you want to use, and remember the choice so it is
applied again after every `env.reset()` and preserved by `backend.copy()`.
Note that a powergrid should have been loaded for this function to work.
This function does not modify :attr:`LightSimBackend.max_iter` nor :attr:`LightSimBackend.tol`. You might want
to modify these values depending on the solver you are using.
Notes
------
By default, the fastest AC solver is used for your platform. This means that if KLU is available, then it is used
otherwise it's SparseLU.
This has to be set for every backend that you want to use. For example, you have to set it
in the backend of the `_obs_env` of the observation and if you are using "grid2op.MultMixEnv` you
have to set it in all mixes!
Parameters
----------
algo_type: Union[lightsim2grid.AlgorithmType, str]
The new algorithm you want to use, either as an :class:`~lightsim2grid.algorithm.AlgorithmType`
enum value, or (since lightsim2grid 1.0) as a plain ``str``: the registry name of a
string-only built-in algorithm with no ``AlgorithmType`` enum value (eg
``"NRRefactorRetry_KLU"``) or of a plugin registered through
``lightsim2grid.load_algorithm_plugin`` (see :ref:`solver_plugin`). See
``env.backend._grid.available_algorithm_names()`` for the list of valid names.
"""
if algo_type is None:
raise BackendError("Impossible to change the algorithm type to None. Please enter a valid algorithm type.")
real_algo_type = self._check_suitable_solver_type(algo_type)
self.__current_algo_type = copy.deepcopy(real_algo_type)
self._grid.change_algorithm(self.__current_algo_type)
@staticmethod
def _aux_algo_config_to_state(config: AlgoConfig) -> Tuple[list, list]:
# AlgoConfig (pybind11) supports neither pickling nor copy.deepcopy (it would
# break pickling the whole backend, see test_save_load in test_pickleable.py,
# if stored as an attribute directly). Its `int_params` / `real_params` are
# plain python lists though (returned by value), so this plain (picklable,
# deepcopy-able) tuple of lists is what is actually kept on the backend;
# `_aux_algo_config_from_state` rebuilds a real AlgoConfig from it whenever
# one needs to be re-applied (reset(), copy()).
return (list(config.int_params), list(config.real_params))
@staticmethod
def _aux_algo_config_from_state(state: Tuple[list, list]) -> AlgoConfig:
int_params, real_params = state
config = AlgoConfig()
config.int_params = list(int_params)
config.real_params = list(real_params)
return config
[docs]
def set_ac_algo_config(self, config: AlgoConfig) -> None:
"""
Change the :class:`~lightsim2grid.algorithm.AlgoConfig` (scaling / refactor
policy and their per-policy parameters) used by the AC algorithm, and remember
it so it is applied again after every `env.reset()` and preserved by
`backend.copy()`.
Unlike calling ``env.backend._grid.set_ac_algo_config(...)`` directly, which is
silently reverted on the next `env.reset()`, this is the persistent, supported
way to customize the AC :class:`~lightsim2grid.algorithm.AlgoConfig`. Note that
a powergrid should have been loaded for this function to work.
Parameters
----------
config: lightsim2grid.algorithm.AlgoConfig
The new AlgoConfig to use for the AC algorithm.
"""
if self._grid is None:
raise BackendError("Impossible to set an AlgoConfig before a powergrid has been loaded.")
self.__current_ac_algo_config = self._aux_algo_config_to_state(config)
self._grid.set_ac_algo_config(config)
[docs]
def get_ac_algo_config(self) -> AlgoConfig:
"""Return the :class:`~lightsim2grid.algorithm.AlgoConfig` currently used by the AC algorithm."""
return self._grid.get_ac_algo_config()
[docs]
def set_dc_algo_config(self, config: AlgoConfig) -> None:
"""Same as :func:`LightSimBackend.set_ac_algo_config`, for the DC algorithm."""
if self._grid is None:
raise BackendError("Impossible to set an AlgoConfig before a powergrid has been loaded.")
self.__current_dc_algo_config = self._aux_algo_config_to_state(config)
self._grid.set_dc_algo_config(config)
[docs]
def get_dc_algo_config(self) -> AlgoConfig:
"""Same as :func:`LightSimBackend.get_ac_algo_config`, for the DC algorithm."""
return self._grid.get_dc_algo_config()
[docs]
def _aux_merge_solver_type_algo_type(
self,
solver_type: Optional[Union[AlgorithmType, SolverType]],
algo_type: Optional[AlgorithmType]) -> Optional[AlgorithmType]:
"""Merge the deprecated `solver_type` kwarg into the canonical `algo_type` one.
"solver" now refers specifically to the *linear* solver (KLU, SparseLU, NICSLU,
CKTSO), not the powerflow algorithm nor the combination of both that
`AlgorithmType` enumerates -- hence the rename. `solver_type` is kept only so
existing code keeps working.
"""
if solver_type is None:
return algo_type
warnings.warn("The `solver_type` kwarg is deprecated: \"solver\" now refers to the "
"linear solver only (KLU, SparseLU, NICSLU, CKTSO), not the powerflow "
"algorithm. Use `algo_type` instead.",
DeprecationWarning,
3)
# normalize only for the comparison: a `SolverType` and the `AlgorithmType` it
# aliases are not `==` to one another (different Enum classes), even though they
# designate the same algorithm.
solver_type_normalized = solver_type.value if isinstance(solver_type, SolverType) else solver_type
if algo_type is not None and algo_type != solver_type_normalized:
raise BackendError("Both `solver_type` (deprecated) and `algo_type` were provided "
"with different values. Pass only `algo_type`.")
return solver_type
def _check_suitable_solver_type(
self,
algo_type: Union[AlgorithmType, SolverType, str], check_in_avail_solver=True) -> Union[AlgorithmType, str]:
if algo_type is None:
return
if isinstance(algo_type, SolverType):
warnings.warn("Passing a SolverType is deprecated. Please use lightsim2grid.AlgorithmType "
"(via the `algo_type` kwarg / `set_algo_type`) instead.",
DeprecationWarning,
2)
algo_type = algo_type.value
if isinstance(algo_type, str):
# a registry name: either a string-only built-in (eg "NRRefactorRetry_KLU",
# which has no AlgorithmType enum value) or a plugin registered through
# load_algorithm_plugin(). Can only be checked once a grid is loaded.
if check_in_avail_solver and getattr(self, "_grid", None) is not None:
avail_names = self._grid.available_algorithm_names()
if algo_type not in avail_names:
raise BackendError(f"The algorithm name \"{algo_type}\" is not available on your system "
f"(nor registered). Available algorithm names are {avail_names}")
return algo_type
if not isinstance(algo_type, AlgorithmType):
raise BackendError(f"The algorithm type must be from type \"lightsim2grid.AlgorithmType\", a plain "
f"`str` (a registered algorithm name), and not {type(algo_type)}")
if check_in_avail_solver and algo_type not in self.available_default_algorithms:
raise BackendError(f"The algorithm type provided \"{algo_type}\" is not available on your system. Available"
f"algorithms are {self.available_default_algorithms}")
return algo_type
[docs]
def set_solver_max_iter(self, max_iter: int) -> None:
"""
Set the maximum number of iteration the solver is allowed to perform.
We do not recommend to modify the default value (10), unless you are using the GaussSeidel powerflow.
This powerflow being slower, we do not recommend to use it.
Recommendation, for medium sized grid (**eg** based on the ieee 118):
- for AlgorithmType.NR_SparseLU: 10
- for AlgorithmType.GaussSeidel: 10000
- for AlgorithmType.NR_KLU: 10
Parameters
----------
max_iter: ``int``
Maximum number of iteration the powerflow can run. It should be number >= 1
Notes
-------
This has to be set for every backend that you want to use. For example, you have to set it
in the backend of the `_obs_env` of the observation and if you are using "grid2op.MultMixEnv` you
have to set it in all mixes!
"""
try:
max_iter = int(max_iter)
except Exception as exc_:
raise BackendError(f"Impossible to convert \"max_iter={max_iter}\" to an integer with exception \"{exc_}\"")
if max_iter < 1:
raise BackendError("max_iter should be a strictly positive integer (integer >= 1)")
self.max_it = max_iter
[docs]
def set_tol(self, new_tol: float) -> None:
"""
Set the tolerance of the powerflow. This means that the powerflow will stop when the Kirchhoff's Circuit Laws
are met up to a tolerance of "new_tol".
Decrease the tolerance might speed up the computation of the powerflow but will decrease the accuracy. We do
not recommend to modify the default value of 1e-8.
Parameters
----------
new_tol: ``float``
The new tolerance to use (should be a float > 0)
Notes
-------
This has to be set for every backend that you want to use. For example, you have to set it
in the backend of the `_obs_env` of the observation and if you are using "grid2op.MultMixEnv` you
have to set it in all mixes!
"""
try:
new_tol = float(new_tol)
except Exception as exc_:
raise BackendError(f"Impossible to convert \"new_tol={new_tol}\" to an float with error \"{exc_}\"") from exc_
if new_tol <= 0:
raise BackendError("new_tol should be a strictly positive float (float > 0)")
self.tol = new_tol
self._idx_hack_storage = np.zeros(0, dtype=dt_int)
def _handle_turnedoff_pv(self):
if self._turned_off_pv:
self._grid.turnedoff_pv()
else:
self._grid.turnedoff_no_pv()
def _assign_right_solver(self):
# storage units share the distributed slack with the generators: a battery
# taking part in it makes the slack distributed even beside a single slack generator
slack_weights = np.array([el.slack_weight for el in self._grid.get_generators()] +
[el.slack_weight for el in self._grid.get_storages()])
nb_slack_nonzero = (np.abs(slack_weights) > 1e-5).sum()
has_single_slack = nb_slack_nonzero == 1
if has_single_slack and not self._dist_slack_non_renew:
if AlgorithmType.NRSing_KLU in self.available_default_algorithms:
# use the faster KLU if available
self._grid.change_algorithm(AlgorithmType.NRSing_KLU)
else:
self._grid.change_algorithm(AlgorithmType.NRSing_SparseLU)
else:
# grid has multiple slacks
if AlgorithmType.NR_KLU in self.available_default_algorithms:
# use the faster KLU if available
self._grid.change_algorithm(AlgorithmType.NR_KLU)
else:
self._grid.change_algorithm(AlgorithmType.NR_SparseLU)
def _aux_set_correct_detach_flags_d_allowed(self):
# user allowed detachment, I check the correct flags
if self._stop_if_gen_disco is None:
# user did not specify anything
self._stop_if_gen_disco = False
elif not self._stop_if_gen_disco:
# force conversion to proper type
self._stop_if_gen_disco = False
elif self._stop_if_gen_disco:
# erase default values and continue like the grid2op call specifies
warnings.warn("Call to `grid2op.make(..., allow_detachment=True)` will erase the lightsim2grid kwargs `stop_if_gen_disco=True`")
self._stop_if_gen_disco = False
if self._stop_if_load_disco is None:
# user did not specify anything
self._stop_if_load_disco = False
elif not self._stop_if_load_disco:
# force conversion to the proper type
self._stop_if_load_disco = False
elif self._stop_if_load_disco:
# erase default values and continue like the grid2op call specifies
warnings.warn("Call to `grid2op.make(..., allow_detachment=True)` will erase the lightsim2grid kwargs `stop_if_load_disco=True`")
self._stop_if_load_disco = False
if self._stop_if_storage_disco is None:
# user did not specify anything
self._stop_if_storage_disco = False
elif not self._stop_if_storage_disco:
# force conversion to the proper type
self._stop_if_storage_disco = False
elif self._stop_if_storage_disco:
# erase default values and continue like the grid2op call specifies
warnings.warn("Call to `grid2op.make(..., allow_detachment=True)` will erase the lightsim2grid kwargs `stop_if_storage_disco=True`")
self._stop_if_storage_disco = False
def _aux_set_correct_detach_flags_d_not_allowed(self):# user did not allow detachment (or it's a legacy grid2op version), I check the correct flags
if self._stop_if_gen_disco is None:
# user did not specify anything
self._stop_if_gen_disco = True
elif self._stop_if_gen_disco:
# force conversion to proper type
self._stop_if_gen_disco = True
elif not self._stop_if_gen_disco:
# erase default values and continue like the grid2op call specifies
warnings.warn("Call to `grid2op.make(..., allow_detachment=False)` will erase the lightsim2grid kwargs `stop_if_gen_disco=False`")
self._stop_if_gen_disco = True
if self._stop_if_load_disco is None:
# user did not specify anything
self._stop_if_load_disco = True
elif self._stop_if_load_disco:
# force conversion to proper type
self._stop_if_load_disco = True
elif not self._stop_if_load_disco:
# erase default values and continue like the grid2op call specifies
warnings.warn("Call to `grid2op.make(..., allow_detachment=False)` will erase the lightsim2grid kwargs `stop_if_load_disco=False`")
self._stop_if_load_disco = True
if self._stop_if_storage_disco is None:
# user did not specify anything
self._stop_if_storage_disco = True
elif self._stop_if_storage_disco:
# force conversion to proper type
self._stop_if_storage_disco = True
elif not self._stop_if_storage_disco:
# erase default values and continue like the grid2op call specifies
warnings.warn("Call to `grid2op.make(..., allow_detachment=False)` will erase the lightsim2grid kwargs `stop_if_storage_disco=False`")
self._stop_if_storage_disco = True
def _aux_set_correct_detach_flags(self):
if self.detachment_is_allowed:
self._aux_set_correct_detach_flags_d_allowed()
else:
self._aux_set_correct_detach_flags_d_not_allowed()
[docs]
def load_grid(self,
path : Union[os.PathLike, str],
filename : Optional[Union[os.PathLike, str]]=None) -> None:
cls = type(self)
if hasattr(cls, "can_handle_more_than_2_busbar"):
# grid2op version >= 1.10.0 then we use this
self.can_handle_more_than_2_busbar()
if hasattr(cls, "can_handle_detachment"):
# grid2op version >= 1.11.0 then we use this
self.can_handle_detachment()
self._aux_set_correct_detach_flags()
if self._automatically_disconnect and not self.detachment_is_allowed:
warnings.warn(("It is not possible to automatically disconnect element if grid2op "
"if grid2op does not allow detachment (make sure to upgrade to grid2op >= 1.11 "
"or if the environment is not configure to use it (add the kwargs grid2op.make(..., allow_detachment=True))")
)
self._automatically_disconnect = False
if self._loader_method == "pandapower":
self._load_grid_pandapower(path, filename)
elif self._loader_method == "pypowsybl":
self._load_grid_pypowsybl(path, filename)
elif self._loader_method == "matpower":
self._load_grid_matpower(path, filename)
else:
raise BackendError(f"Impossible to initialize the backend with '{self._loader_method}'")
self._grid.tell_solver_need_reset()
self._reset_res_pointers() # force the re reading of the accessors at next powerflow
self.V = np.ones(self.nb_bus_total, dtype=complex)
if self._automatically_disconnect:
self._need_islanding_detection = True
def _should_not_have_to_do_this(self, path=None, filename=None):
# included in grid2op now !
# but before `make_complete_path` was introduced we need to still
# be able to use lightsim2grid
if path is None and filename is None:
raise Grid2OpException(
"You must provide at least one of path or file to load a powergrid."
)
if path is None:
full_path = filename
elif filename is None:
full_path = path
else:
full_path = os.path.join(path, filename)
if not os.path.exists(full_path):
raise Grid2OpException('There is no powergrid at "{}"'.format(full_path))
return full_path
def _aux_get_substation_handling_from_loader_kwargs(self, loader_kwargs) -> Union[int, None]:
res = None
if "double_bus_per_sub" in loader_kwargs and loader_kwargs["double_bus_per_sub"]:
res = DEFAULT_N_BUSBAR_PER_SUB
if "n_busbar_per_sub" in loader_kwargs and loader_kwargs["n_busbar_per_sub"]:
if res is not None:
raise BackendError(f"When intializing a grid from {self._loader_method}, you cannot "
"set both `double_bus_per_sub` and `n_busbar_per_sub` "
"in the `loader_kwargs`. "
"You can only set `n_busbar_per_sub` in this case.")
res = int(loader_kwargs["n_busbar_per_sub"])
if loader_kwargs["n_busbar_per_sub"] != res:
raise BackendError(f"When initializing a grid from {self._loader_method}, the `n_busbar_per_sub` "
"loader kwargs should be properly convertible to an int "
"giving the default number of busbar sections per substation.")
if res is not None:
if self.n_busbar_per_sub != res:
if self.n_busbar_per_sub != DEFAULT_N_BUSBAR_PER_SUB:
warnings.warn(f"You specified `n_busbar={self.n_busbar_per_sub}` when calling `grid2op.make(...)` "
f"but also specified `n_busbar_per_sub` (or `double_bus_per_sub`) in the "
f"`loader_kwargs` when building the `LightSimBackend` "
f"(**eg** `LighSimBackend(..., loader_kwargs={{'n_busbar_per_sub': {res} }})`). "
f"IN THIS CASE {self.n_busbar_per_sub} WILL BE USED (the things specified in the "
f"`grid2op.make(...)`)")
res = self.n_busbar_per_sub
elif res != DEFAULT_N_BUSBAR_PER_SUB:
warnings.warn(f"You specified `n_busbar_per_sub={res}` (or `double_bus_per_sub`) "
f"in the `load_kwargs` when building the LightSimBackend "
f"(**eg** `LighSimBackend(..., loader_kwargs={{'n_busbar_per_sub': {res} }})`) "
f"without modifying the `n_busbar=...` when calling `grid2op.make(...)`. "
f"LightSimBackend WILL OVERRIDE THE MAXIMUM NUMBER OF BUSES PER SUBSTATION "
f"and set it to {res}. You can silence this warning by creating the env with "
f"`grid2op.make(..., n_busbar={res})`")
self.n_busbar_per_sub = res
return res
def _get_subid_from_buses_legacy(self, buses_sub_id, el_sub_df):
# buses_sub_id is the first element as returned by from_pypowsybl / init function
# el_sub_df is an element dataframe returned by the same function
tmp = pd.merge(el_sub_df.reset_index(), buses_sub_id, how="left", right_on="sub_id", left_on="sub_id")
res = tmp.drop_duplicates(subset='id').set_index("id").sort_index()["bus_global_id"].values
return res
def _load_grid_pypowsybl(self, path=None, filename=None):
from lightsim2grid.network.from_pypowsybl import init as init_from_pypowsybl
import pypowsybl.network as pypow_net
loader_kwargs = {}
if self._loader_kwargs is not None:
loader_kwargs = self._loader_kwargs
self._aux_check_loader_kwargs(loader_kwargs, "pypowsybl", type(self).KEYS_PYPOWSYBL_LOADER)
try:
full_path = self.make_complete_path(path, filename)
except AttributeError as _:
warnings.warn("Please upgrade your grid2op version")
full_path = self._should_not_have_to_do_this(path, filename)
if "grid" in loader_kwargs:
if not isinstance(loader_kwargs["grid"], pypow_net.Network):
raise Grid2OpException("The grid you provided in the 'grid' loader_kwargs "
"is not a valid pypowsybl grid. It is of type "
f"{type(loader_kwargs['grid'])}.")
# use the grid provided as input
grid_tmp = loader_kwargs["grid"]
else:
if "pypowsybl_load_kwargs" in loader_kwargs:
pypowsybl_load_kwargs = loader_kwargs["pypowsybl_load_kwargs"]
else:
pypowsybl_load_kwargs = {}
grid_tmp = pypow_net.load(full_path, **pypowsybl_load_kwargs)
self._orig_grid_pypowsybl = grid_tmp
self._grid_laid_out_by_loader = True
buses_for_sub = False
if "use_buses_for_sub" in loader_kwargs and loader_kwargs["use_buses_for_sub"]:
df = grid_tmp.get_buses()
buses_for_sub = True
self.n_sub = df.shape[0]
else:
df = grid_tmp.get_voltage_levels()
self.n_sub = df.shape[0]
sort_index = True
if "sort_index" in loader_kwargs and (bool(loader_kwargs["sort_index"]) == loader_kwargs["sort_index"]):
sort_index = bool(loader_kwargs["sort_index"])
init_vm_pu = 1.06
if "init_vm_pu" in loader_kwargs and float(loader_kwargs["init_vm_pu"]) != loader_kwargs["init_vm_pu"]:
init_vm_pu = float(loader_kwargs["init_vm_pu"])
sn_mva = None
if "sn_mva" in loader_kwargs and float(loader_kwargs["sn_mva"]) != loader_kwargs["sn_mva"]:
sn_mva = float(loader_kwargs["sn_mva"])
n_busbar_per_sub = self._aux_get_substation_handling_from_loader_kwargs(loader_kwargs)
if n_busbar_per_sub is None:
n_busbar_per_sub = self.n_busbar_per_sub
self._grid, subs_id = init_from_pypowsybl(
grid_tmp,
gen_slack_id=self._gen_slack_id,
slack_bus_id=None,
sn_mva=sn_mva,
sort_index=sort_index,
only_main_component=self._automatically_disconnect,
return_sub_id=True,
n_busbar_per_sub=n_busbar_per_sub,
buses_for_sub=buses_for_sub,
init_vm_pu=init_vm_pu,
)
self.__nb_powerline = len(self._grid.get_lines())
(gen_sub, load_sub, (lor_sub, tor_sub), (lex_sub, tex_sub),
batt_sub, sh_sub, hvdc_sub_from_id, hvdc_sub_to_id) = subs_id
if self._gen_slack_id is not None:
self._gen_slack_id = gen_sub["desired_slack"].values.nonzero()[0]
if not buses_for_sub:
self.__nb_bus_before = self._grid.get_n_sub()
else:
self.__nb_bus_before = grid_tmp.get_buses().shape[0]
self._aux_setup_right_after_grid_init()
# the names
use_grid2op_default_names = True
if "use_grid2op_default_names" in loader_kwargs and not loader_kwargs["use_grid2op_default_names"]:
use_grid2op_default_names = False
if use_grid2op_default_names:
self.name_sub = ["sub_{}".format(i) for i, _ in enumerate(range(self.n_sub))]
self._grid.set_substation_names(self.name_sub)
else:
self.name_sub = self._grid.get_substation_names()
if self.n_sub != len(self.name_sub):
raise RuntimeError("LightSimBackend failed to initialize the grid.")
# mandatory for the backend
self.n_line = len(self._grid.get_lines()) + len(self._grid.get_trafos())
self.n_gen = len(self._grid.get_generators())
self.n_load = len(self._grid.get_loads())
self.n_storage = len(self._grid.get_storages())
if type(self).shunts_data_available:
self.n_shunt = len(self._grid.get_shunts())
else:
self.n_shunt = None
# assign substation to each element (grid2op side). `init_from_pypowsybl` has
# already put this on the grid (the same values it returns in the dataframes
# above), so read it back from there rather than deriving it twice
self._aux_read_subid_from_grid()
# handle the names
if use_grid2op_default_names:
self.name_load = np.array([f"load_{el.sub_id}_{id_obj}" for id_obj, el in enumerate(self._grid.get_loads())]).astype(str)
self.name_gen = np.array([f"gen_{el.sub_id}_{id_obj}" for id_obj, el in enumerate(self._grid.get_generators())]).astype(str)
self.name_line = np.array(
[f"{el.sub1_id}_{el.sub2_id}_{id_obj}" for id_obj, el in enumerate(self._grid.get_lines())] +
[f"{el.sub1_id}_{el.sub2_id}_{id_obj + self.__nb_powerline}" for id_obj, el in enumerate(self._grid.get_trafos())]).astype(str)
self.name_storage = np.array([f"storage_{el.sub_id}_{id_obj}" for id_obj, el in enumerate(self._grid.get_storages())]).astype(str)
self.name_shunt = np.array([f"shunt_{el.sub_id}_{id_obj}" for id_obj, el in enumerate(self._grid.get_shunts())]).astype(str)
# synch name of the grid model
self._grid.set_gen_names(self.name_gen)
self._grid.set_load_names(self.name_load)
self._grid.set_line_names(self.name_line[:self.__nb_powerline])
self._grid.set_trafo_names(self.name_line[self.__nb_powerline:])
self._grid.set_storage_names(self.name_storage)
self._grid.set_shunt_names(self.name_shunt)
else:
self.name_load = np.array(load_sub.index.astype(str))
self.name_gen = np.array(gen_sub.index.astype(str))
self.name_line = np.concatenate((lor_sub.index.astype(str), tor_sub.index.astype(str)))
self.name_storage = np.array(batt_sub.index.astype(str))
self.name_shunt = np.array(sh_sub.index.astype(str))
self.name_sub = self._grid.get_substation_names().copy()
# and now things needed by the backend (legacy)
self.prod_pu_to_kv = 1.0 * self._grid.get_bus_vn_kv()[[el.sub_id for el in self._grid.get_generators()]]
self.prod_pu_to_kv = self.prod_pu_to_kv.astype(dt_float)
if "reconnect_disco_gen" in loader_kwargs and loader_kwargs["reconnect_disco_gen"]:
for el in self._grid.get_generators():
if not el.connected:
self._grid.reactivate_gen(el.id)
self._grid.change_bus_gen(el.id, self.gen_to_subid[el.id])
self._grid.change_p_gen(el.id, 0.)
self._grid.change_v_gen(el.id, self.prod_pu_to_kv[el.id])
if "reconnect_disco_load" in loader_kwargs and loader_kwargs["reconnect_disco_load"]:
for el in self._grid.get_loads():
if not el.connected:
self._grid.reactivate_load(el.id)
self._grid.change_bus_load(el.id, self.load_to_subid[el.id])
self._grid.change_p_load(el.id, 0.)
self._grid.change_q_load(el.id, 0.)
# complete the other vectors
self._compute_pos_big_topo()
self.__nb_powerline = len(self._grid.get_lines())
# init this
self.prod_p = np.array([el.target_p_mw for el in self._grid.get_generators()], dtype=dt_float)
self.next_prod_p = np.array([el.target_p_mw for el in self._grid.get_generators()], dtype=dt_float)
self.nb_bus_total = len(self._grid.get_bus_vn_kv())
# and now things needed by the backend (legacy)
self._big_topo_to_obj = [(None, None) for _ in range(type(self).dim_topo)]
self.prod_pu_to_kv = 1.0 * self._grid.get_bus_vn_kv()[[el.bus_id for el in self._grid.get_generators()]]
self.prod_pu_to_kv = self.prod_pu_to_kv.astype(dt_float)
# TODO
max_not_too_max = (np.finfo(dt_float).max * 0.5 - 1.)
self.thermal_limit_a = max_not_too_max * np.ones(self.n_line, dtype=dt_float)
bus_vn_kv = np.array(self._grid.get_bus_vn_kv())
self._sh_vnkv = bus_vn_kv[self.shunt_to_subid]
self._aux_finish_setup_after_reading()
[docs]
def _aux_read_subid_from_grid(self) -> None:
"""
Read back, from the :class:`lightsim2grid.network.LSGrid` itself, the
substation / voltage level each element belongs to.
Which substation an element sits in is a property of the file the grid was read
from, so it is the loader (`init_from_pypowsybl`, `init_from_matpower` and the
shared `init_from_powermodels` engine behind it) that knows it, and the loader
sets it on the grid it returns -- see the `set_*_to_subid` block at the end of
`lightsim2grid/network/from_powermodels/initLSGrid.py`. Rather than deriving the
same information a second time here from whatever intermediate the loader
happened to hand back, read it off the grid: one source of truth, and one place
to fix when a source format numbers its substations in its own way.
Note that this is why the answer is read from `el.sub_id` and never from
`el.bus_id`: an element the source file declares out of service reports
``bus_id == -1``, while the substation it belongs to is a property of the grid
rather than of its status.
The two "line" arrays are grid2op's, so they hold the powerlines then the
transformers, split at `self.__nb_powerline` -- which is the number of
powerlines of the very same grid, so the concatenation below and that split
point cannot disagree.
"""
cls = type(self)
self.load_to_subid = np.array([el.sub_id for el in self._grid.get_loads()], dtype=dt_int)
self.gen_to_subid = np.array([el.sub_id for el in self._grid.get_generators()], dtype=dt_int)
self.line_or_to_subid = np.array([el.sub1_id for el in self._grid.get_lines()] +
[el.sub1_id for el in self._grid.get_trafos()], dtype=dt_int)
self.line_ex_to_subid = np.array([el.sub2_id for el in self._grid.get_lines()] +
[el.sub2_id for el in self._grid.get_trafos()], dtype=dt_int)
if self.__has_storage:
self.storage_to_subid = np.array([el.sub_id for el in self._grid.get_storages()], dtype=dt_int)
if cls.shunts_data_available:
self.shunt_to_subid = np.array([el.sub_id for el in self._grid.get_shunts()], dtype=dt_int)
[docs]
def _load_grid_matpower(self, path=None, filename=None):
"""
Initialize the backend from a MATPOWER case (an environment shipping a `grid.m`
or a `grid.mat` file), through :func:`lightsim2grid.network.init_from_matpower`.
MATPOWER knows nothing about substations, busbar sections, element names or
thermal limits: a case is a set of numbered buses and the branches / generators
/ loads hanging off them. So, exactly like a pypowsybl grid loaded with
`use_buses_for_sub=True`, there is one grid2op substation per MATPOWER bus, the
elements get grid2op's default names, and the thermal limits are left
"infinite" for the environment's `config.py` to set (the standard grid2op place
for them).
"""
from lightsim2grid.network import init_from_matpower
loader_kwargs = {}
if self._loader_kwargs is not None:
loader_kwargs = self._loader_kwargs
self._aux_check_loader_kwargs(loader_kwargs, "matpower", type(self).KEYS_MATPOWER_LOADER)
n_busbar_per_sub = self._aux_get_substation_handling_from_loader_kwargs(loader_kwargs)
if n_busbar_per_sub is None:
n_busbar_per_sub = self.n_busbar_per_sub
if "grid" in loader_kwargs:
# an already parsed matpower case was given: use it and ignore the
# "path" / "filename" of `grid2op.make` (same contract as the pypowsybl
# loader's own "grid" kwarg)
source = loader_kwargs["grid"]
else:
try:
source = self.make_complete_path(path, filename)
except AttributeError as _:
warnings.warn("Please upgrade your grid2op version")
source = self._should_not_have_to_do_this(path, filename)
self._grid = init_from_matpower(source, n_busbar_per_sub=n_busbar_per_sub)
# `init_from_matpower` laid the substations out itself (one per matpower bus,
# with `n_busbar_per_sub` busbar sections each): the backend must not redo it
self._grid_laid_out_by_loader = True
self.__nb_powerline = len(self._grid.get_lines())
self.n_sub = self._grid.get_n_sub()
self.__nb_bus_before = self.n_sub
self._aux_setup_right_after_grid_init()
# mandatory for the backend
self.n_line = len(self._grid.get_lines()) + len(self._grid.get_trafos())
self.n_gen = len(self._grid.get_generators())
self.n_load = len(self._grid.get_loads())
self.n_storage = len(self._grid.get_storages())
if type(self).shunts_data_available:
self.n_shunt = len(self._grid.get_shunts())
else:
self.n_shunt = None
# assign substation to each element (grid2op side)
self._aux_read_subid_from_grid()
# the names. A matpower case names nothing -- it numbers buses -- so grid2op's
# own default names are used, and it is grid2op that makes them: leave them
# unset and let `Backend._fill_names_obj` fill them in from the substation ids
# just read. An environment loaded from a matpower case is then named exactly
# like any other grid2op grid whose backend leaves the names out, by the one
# piece of code that decides what those names look like.
if not hasattr(self, "_fill_names_obj"):
raise BackendError("Loading a grid from a matpower case needs a grid2op recent "
"enough to provide `Backend._fill_names_obj`: matpower names "
"nothing, so grid2op's default names are used and grid2op is "
"what makes them. Please upgrade grid2op.")
self.name_sub = None
self.name_load = None
self.name_gen = None
self.name_line = None
self.name_storage = None
self.name_shunt = None
with warnings.catch_warnings():
# `_fill_names_obj` warns once per name vector, because a backend reaching
# it usually forgot to fill them. Here there is nothing to forget: the
# source format HAS no names, which the docstring above says and the
# documentation repeats.
warnings.simplefilter("ignore")
self._fill_names_obj()
# and give them back to the grid, so an LSGrid error message names the same
# element grid2op does
self._grid.set_substation_names(self.name_sub)
self._grid.set_gen_names(self.name_gen)
self._grid.set_load_names(self.name_load)
self._grid.set_line_names(self.name_line[:self.__nb_powerline])
self._grid.set_trafo_names(self.name_line[self.__nb_powerline:])
if self.__has_storage:
self._grid.set_storage_names(self.name_storage)
if self.n_shunt is not None:
self._grid.set_shunt_names(self.name_shunt)
# complete the other vectors
self._compute_pos_big_topo()
# and now things needed by the backend (legacy)
bus_vn_kv = np.array(self._grid.get_bus_vn_kv())
self.prod_pu_to_kv = bus_vn_kv[self.gen_to_subid].astype(dt_float)
self.load_pu_to_kv = bus_vn_kv[self.load_to_subid].astype(dt_float)
self.lines_or_pu_to_kv = bus_vn_kv[self.line_or_to_subid].astype(dt_float)
self.lines_ex_pu_to_kv = bus_vn_kv[self.line_ex_to_subid].astype(dt_float)
if self.__has_storage:
self.storage_pu_to_kv = bus_vn_kv[self.storage_to_subid].astype(dt_float)
if self.n_shunt is not None:
self._sh_vnkv = bus_vn_kv[self.shunt_to_subid]
# matpower's RATE_A is a branch MVA rating, not the ampere limit grid2op wants,
# and is 0 ("unlimited") in a fair share of the published cases: leave the
# limits open here and let the environment's `config.py` set them, which is
# where a grid2op environment declares them anyway.
max_not_too_max = (np.finfo(dt_float).max * 0.5 - 1.)
self.thermal_limit_a = max_not_too_max * np.ones(self.n_line, dtype=dt_float)
self.prod_p = np.array([el.target_p_mw for el in self._grid.get_generators()], dtype=dt_float)
self.next_prod_p = 1.0 * self.prod_p
self.nb_bus_total = len(bus_vn_kv)
self._big_topo_to_obj = [(None, None) for _ in range(type(self).dim_topo)]
self._aux_finish_setup_after_reading()
@property
def available_solvers(self):
warnings.warn(
"deprecated, please use :attr:`available_default_algorithms` instead",
category=DeprecationWarning,
stacklevel=2)
return self.available_default_algorithms
def _aux_setup_right_after_grid_init(self):
if not self._grid_laid_out_by_loader:
self._grid.set_n_sub(self.__nb_bus_before)
self._handle_turnedoff_pv()
self.available_default_algorithms = self._grid.available_default_algorithms()
if self.__current_algo_type is None:
# previous default behaviour (< 0.7)
# by default it builds the backend with the fastest solver
# automatically found
self._assign_right_solver()
if AlgorithmType.DC_KLU in self.available_default_algorithms:
# use the faster KLU if available even for DC approximation
self._grid.change_algorithm(AlgorithmType.DC_KLU)
self.__current_algo_type = copy.deepcopy(self._grid.get_algo_type())
else:
# check that the solver type provided is installed with lightsim2grid
self._check_suitable_solver_type(self.__current_algo_type)
self._grid.change_algorithm(self.__current_algo_type)
# handle multiple busbar per substations
if hasattr(type(self), "can_handle_more_than_2_busbar") and not self._grid_laid_out_by_loader:
# grid2op version >= 1.10.0 then we use this
self._grid._max_nb_bus_per_sub = self.n_busbar_per_sub
self._grid.tell_solver_need_reset()
def init_from_loaded_pandapower(self, pp_net):
if hasattr(type(self), "can_handle_more_than_2_busbar") and self._init_pp_backend is not None:
# TODO I don't really understand why I need the check "self._init_pp_backend is not None"
# it's needed at least for `test_n1contingencyrewards.py`
type(self._init_pp_backend).n_busbar_per_sub = self.n_busbar_per_sub
self._init_pp_backend = pp_net.copy()
self._aux_init_pandapower()
# handles redispatching
if type(pp_net).redispatching_unit_commitment_availble:
self.redispatching_unit_commitment_availble = True
for attr_nm in ["gen_type", "gen_pmin", "gen_pmax",
"gen_redispatchable", "gen_max_ramp_up",
"gen_max_ramp_down", "gen_min_uptime",
"gen_min_downtime", "gen_cost_per_MW",
"gen_startup_cost", "gen_shutdown_cost",
"gen_renewable"
]:
setattr(self, attr_nm, copy.deepcopy(getattr( type(pp_net), attr_nm)))
# handles storages
for attr_nm in ["storage_type",
"storage_Emax",
"storage_Emin",
"storage_max_p_prod" ,
"storage_max_p_absorb",
"storage_marginal_cost",
"storage_loss",
"storage_charging_efficiency",
"storage_discharging_efficiency",
]:
setattr(self, attr_nm, copy.deepcopy(getattr( type(pp_net), attr_nm)))
# handle the "_public" type of attributes
# ie attributes added when methods like load_grid_public were introduced in
# grid2op
for attr_nm in ["_load_bus_target", "_gen_bus_target", "_storage_bus_target", "_shunt_bus_target"]:
if not hasattr(pp_net, attr_nm):
# in grid2op compat mode
continue
setattr(self, attr_nm, copy.deepcopy(getattr(pp_net, attr_nm)))
def _load_grid_pandapower(self, path=None, filename=None):
self._aux_assign_init_pp_backend()
if hasattr(Backend, "can_handle_more_than_2_busbar"):
type(self._init_pp_backend).n_busbar_per_sub = type(self).n_busbar_per_sub
if hasattr(Backend, "detachment_is_allowed"):
type(self._init_pp_backend).detachment_is_allowed = type(self).detachment_is_allowed
if hasattr(Backend, "shunts_data_available"):
type(self._init_pp_backend).shunts_data_available = type(self).shunts_data_available
type(self._init_pp_backend).set_env_name(type(self).env_name)
if type(self).glop_version is not None:
type(self._init_pp_backend).glop_version = type(self).glop_version
self._init_pp_backend.load_grid(path, filename)
self._aux_init_pandapower()
def _aux_check_loader_kwargs(
self,
loader_kwargs,
loader_name,
allowed_loader_kwargs):
for el in loader_kwargs:
if el not in allowed_loader_kwargs:
raise RuntimeError(f"Invalid 'loader_kwargs' {el} provided when loading a "
f"LightSimBackend with {loader_name} grids. Supported keys are: "
f"{sorted(allowed_loader_kwargs)}.")
def _aux_init_pandapower(self):
from lightsim2grid.network import init_from_pandapower
pp_orig_file = "pandapower_v2"
loader_kwargs = {}
if self._loader_kwargs is not None:
loader_kwargs = self._loader_kwargs
self._aux_check_loader_kwargs(loader_kwargs, "pandapower", type(self).KEYS_PANDAPOWER_LOADER)
if "pp_orig_file" in loader_kwargs and str(loader_kwargs["pp_orig_file"]) == loader_kwargs["pp_orig_file"]:
pp_orig_file = str(loader_kwargs["pp_orig_file"])
self._grid = init_from_pandapower(self._init_pp_backend._grid,
self._init_pp_backend.n_sub,
self.n_busbar_per_sub,
pp_orig_file=pp_orig_file,
# grid2op's own pandapower backend is what says
# what a substation is on this path, and has been
# for as long as it has existed:
# `_aux_finish_setup_after_reading` copies its
# answer onto the grid. Let the loader work out
# its own and the two could disagree silently.
init_subid=False)
self.__nb_bus_before = self._init_pp_backend.get_nb_active_bus()
self._aux_setup_right_after_grid_init()
self.__nb_powerline = self._init_pp_backend._grid.line.shape[0]
self.__nb_bus_before = self._init_pp_backend.get_nb_active_bus()
# deactive the buses that have been added
for bus_id, bus_status in enumerate(self._init_pp_backend._grid.bus["in_service"].values):
if bus_status:
self._grid.reactivate_bus(bus_id)
else:
self._grid.deactivate_bus(bus_id)
pp_cls = type(self._init_pp_backend)
if pp_cls.n_line <= -1:
warnings.warn("You are using a legacy (quite old now) grid2op version. Please upgrade it.")
pp_cls = self._init_pp_backend
self.n_line = pp_cls.n_line
self.n_gen = pp_cls.n_gen
self.n_load = pp_cls.n_load
self.n_sub = pp_cls.n_sub
self.sub_info = pp_cls.sub_info
self.dim_topo = pp_cls.dim_topo
self.load_to_subid = pp_cls.load_to_subid
self.gen_to_subid = pp_cls.gen_to_subid
self.line_or_to_subid = pp_cls.line_or_to_subid
self.line_ex_to_subid = pp_cls.line_ex_to_subid
self.load_to_sub_pos = pp_cls.load_to_sub_pos
self.gen_to_sub_pos = pp_cls.gen_to_sub_pos
self.line_or_to_sub_pos = pp_cls.line_or_to_sub_pos
self.line_ex_to_sub_pos = pp_cls.line_ex_to_sub_pos
# synch names with pandapower ones
self.name_gen = pp_cls.name_gen
self._grid.set_gen_names(self.name_gen)
self.name_load = pp_cls.name_load
self._grid.set_load_names(self.name_load)
self.name_line = pp_cls.name_line
self._grid.set_line_names(self.name_line[:self.__nb_powerline])
self._grid.set_trafo_names(self.name_line[self.__nb_powerline:])
self.name_sub = pp_cls.name_sub
self._grid.set_substation_names(self.name_sub)
self.prod_pu_to_kv = self._init_pp_backend.prod_pu_to_kv.copy()
self.load_pu_to_kv = self._init_pp_backend.load_pu_to_kv.copy()
self.lines_or_pu_to_kv = self._init_pp_backend.lines_or_pu_to_kv.copy()
self.lines_ex_pu_to_kv = self._init_pp_backend.lines_ex_pu_to_kv.copy()
# TODO storage check grid2op version and see if storage is available !
if self.__has_storage:
self.n_storage = pp_cls.n_storage
self.storage_to_subid = pp_cls.storage_to_subid
self.storage_pu_to_kv = self._init_pp_backend.storage_pu_to_kv
self.name_storage = pp_cls.name_storage
self._grid.set_storage_names(self.name_storage)
self.storage_to_sub_pos = pp_cls.storage_to_sub_pos
self.storage_type = pp_cls.storage_type
self.storage_Emin = pp_cls.storage_Emin
self.storage_Emax = pp_cls.storage_Emax
self.storage_max_p_prod = pp_cls.storage_max_p_prod
self.storage_max_p_absorb = pp_cls.storage_max_p_absorb
self.storage_marginal_cost = pp_cls.storage_marginal_cost
self.storage_loss = pp_cls.storage_loss
self.storage_discharging_efficiency = pp_cls.storage_discharging_efficiency
self.storage_charging_efficiency = pp_cls.storage_charging_efficiency
self.nb_bus_total = self._init_pp_backend._grid.bus.shape[0]
self.thermal_limit_a = copy.deepcopy(self._init_pp_backend.thermal_limit_a)
self._init_bus_load = 1.0 * self._init_pp_backend._grid.load["bus"].values
self._init_bus_gen = 1.0 * self._init_pp_backend._grid.gen["bus"].values
self._init_bus_lor = 1.0 * self._init_pp_backend._grid.line["from_bus"].values
self._init_bus_lex = 1.0 * self._init_pp_backend._grid.line["to_bus"].values
t_for = 1.0 * self._init_pp_backend._grid.trafo["hv_bus"].values
t_fex = 1.0 * self._init_pp_backend._grid.trafo["lv_bus"].values
self._init_bus_lor = np.concatenate((self._init_bus_lor, t_for)).astype(int)
self._init_bus_lex = np.concatenate((self._init_bus_lex, t_fex)).astype(int)
self._init_bus_load = self._init_bus_load.astype(int)
self._init_bus_gen = self._init_bus_gen.astype(int)
tmp = self._init_bus_lor + self.__nb_bus_before
self._init_bus_lor = np.concatenate((self._init_bus_lor.reshape(-1, 1),
tmp.reshape(-1, 1)), axis=-1)
tmp = self._init_bus_lex + self.__nb_bus_before
self._init_bus_lex = np.concatenate((self._init_bus_lex.reshape(-1, 1),
tmp.reshape(-1, 1)), axis=-1)
tmp = self._init_bus_load + self.__nb_bus_before
self._init_bus_load = np.concatenate((self._init_bus_load.reshape(-1, 1),
tmp.reshape(-1, 1)), axis=-1)
tmp = self._init_bus_gen + self.__nb_bus_before
self._init_bus_gen = np.concatenate((self._init_bus_gen.reshape(-1, 1),
tmp.reshape(-1, 1)), axis=-1)
self._big_topo_to_obj = [(None, None) for _ in range(self.dim_topo)]
self.prod_p = 1.0 * self._init_pp_backend._grid.gen["p_mw"].values
self.next_prod_p = 1.0 * self._init_pp_backend._grid.gen["p_mw"].values
if type(self).shunts_data_available:
if pp_cls.n_shunt is not None:
# modern grid2op version
self.n_shunt = pp_cls.n_shunt
self.shunt_to_subid = pp_cls.shunt_to_subid
self.name_shunt = pp_cls.name_shunt
else:
# legacy grid2op version...
warnings.warn("You are using a legacy grid2op version, please upgrade grid2op.")
self.n_shunt = self._init_pp_backend.n_shunt
self.shunt_to_subid = self._init_pp_backend.shunt_to_subid
self.name_shunt = self._init_pp_backend.name_shunt
else:
self.n_shunt = None
self._compute_pos_big_topo()
if hasattr(self._init_pp_backend, "_sh_vnkv"):
# attribute has been added in grid2op ~1.3 or 1.4
self._sh_vnkv = self._init_pp_backend._sh_vnkv
self._aux_finish_setup_after_reading()
def _aux_finish_setup_after_reading(self):
# set up the "lightsim grid" accordingly
my_cls = type(self)
if my_cls.n_line <= -1:
warnings.warn("You are using a legacy (quite old now) grid2op version. Please upgrade it.")
cls = self
else:
cls = my_cls
self._grid.set_load_pos_topo_vect(cls.load_pos_topo_vect)
self._grid.set_gen_pos_topo_vect(cls.gen_pos_topo_vect)
self._grid.set_line_pos1_topo_vect(cls.line_or_pos_topo_vect[:self.__nb_powerline])
self._grid.set_line_pos2_topo_vect(cls.line_ex_pos_topo_vect[:self.__nb_powerline])
self._grid.set_trafo_pos1_topo_vect(cls.line_or_pos_topo_vect[self.__nb_powerline:])
self._grid.set_trafo_pos2_topo_vect(cls.line_ex_pos_topo_vect[self.__nb_powerline:])
self._grid.set_load_to_subid(cls.load_to_subid)
self._grid.set_gen_to_subid(cls.gen_to_subid)
self._grid.set_line_to_sub1_id(cls.line_or_to_subid[:self.__nb_powerline])
self._grid.set_line_to_sub2_id(cls.line_ex_to_subid[:self.__nb_powerline])
self._grid.set_trafo_to_sub1_id(cls.line_or_to_subid[self.__nb_powerline:])
self._grid.set_trafo_to_sub2_id(cls.line_ex_to_subid[self.__nb_powerline:])
# TODO storage check grid2op version and see if storage is available !
if self.__has_storage:
self._grid.set_storage_to_subid(cls.storage_to_subid)
self._grid.set_storage_pos_topo_vect(cls.storage_pos_topo_vect)
if type(self).shunts_data_available:
self._grid.set_shunt_to_subid(cls.shunt_to_subid)
nm_ = "load"
for load_id, pos_big_topo in enumerate(cls.load_pos_topo_vect):
self._big_topo_to_obj[pos_big_topo] = (load_id, nm_)
nm_ = "gen"
for gen_id, pos_big_topo in enumerate(cls.gen_pos_topo_vect):
self._big_topo_to_obj[pos_big_topo] = (gen_id, nm_)
nm_ = "lineor"
for l_id, pos_big_topo in enumerate(cls.line_or_pos_topo_vect):
self._big_topo_to_obj[pos_big_topo] = (l_id, nm_)
nm_ = "lineex"
for l_id, pos_big_topo in enumerate(cls.line_ex_pos_topo_vect):
self._big_topo_to_obj[pos_big_topo] = (l_id, nm_)
# TODO storage check grid2op version and see if storage is available !
if self.__has_storage:
nm_ = "storage"
for l_id, pos_big_topo in enumerate(cls.storage_pos_topo_vect):
self._big_topo_to_obj[pos_big_topo] = (l_id, nm_)
# number of object per bus, to activate, deactivate them
self.nb_obj_per_bus = np.zeros(2 * self.__nb_bus_before, dtype=dt_int).reshape(-1)
self.topo_vect = np.ones(cls.dim_topo, dtype=dt_int).reshape(-1)
if cls.shunts_data_available:
self.shunt_topo_vect = np.ones(cls.n_shunt, dtype=dt_int)
# shunts
self.sh_p = np.full(cls.n_shunt, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.sh_q = np.full(cls.n_shunt, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.sh_v = np.full(cls.n_shunt, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.sh_theta = np.full(cls.n_shunt, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.sh_bus = np.full(cls.n_shunt, dtype=dt_int, fill_value=-1).reshape(-1)
self.p_or = np.full(cls.n_line, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.q_or = np.full(cls.n_line, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.v_or = np.full(cls.n_line, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.a_or = np.full(cls.n_line, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.p_ex = np.full(cls.n_line, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.q_ex = np.full(cls.n_line, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.v_ex = np.full(cls.n_line, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.a_ex = np.full(cls.n_line, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.load_p = np.full(cls.n_load, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.load_q = np.full(cls.n_load, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.load_v = np.full(cls.n_load, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.prod_p = np.full(cls.n_gen, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.prod_q = np.full(cls.n_gen, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.prod_v = np.full(cls.n_gen, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.line_or_theta = np.full(cls.n_line, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.line_ex_theta = np.full(cls.n_line, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.load_theta = np.full(cls.n_load, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.gen_theta = np.full(cls.n_gen, dtype=dt_float, fill_value=np.nan).reshape(-1)
# TODO storage check grid2op version and see if storage is available !
if self.__has_storage:
self.storage_p = np.full(cls.n_storage, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.storage_q = np.full(cls.n_storage, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.storage_v = np.full(cls.n_storage, dtype=dt_float, fill_value=np.nan).reshape(-1)
self.storage_theta = np.full(cls.n_storage, dtype=dt_float, fill_value=np.nan).reshape(-1)
self._count_object_per_bus()
# set the initial topology vector
n_sub_cls_orig = cls.n_sub
n_sub_ls_orig = LightSimBackend.n_sub
try:
cls.n_sub = self.n_sub
LightSimBackend.n_sub = self.n_sub
if my_cls.load_pos_topo_vect is None:
# old grid2op version
self.topo_vect[:] = 1
self.sh_bus[:] = 1
else:
self.topo_vect.flags.writeable = True # this is set to False in _read_topo_vect
self._read_topo_vect(self.topo_vect) # this also reads shunts information
self._grid.tell_solver_need_reset()
self.__me_at_init = self._grid.copy()
self.__init_topo_vect = np.ones(cls.dim_topo, dtype=dt_int)
self.__init_topo_vect[:] = self.topo_vect
self.__init_topo_vect.flags.writeable = False
if my_cls.shunts_data_available:
self.__init_shunt_bus = self.sh_bus.copy()
self.__init_shunt_bus.flags.writeable = False
finally:
cls.n_sub = n_sub_cls_orig
LightSimBackend.n_sub = n_sub_ls_orig
[docs]
def _read_topo_vect(self, res):
"""the input vector "res" is modified !"""
# TODO speed optimization here to read it "better"
cls = type(self)
# handle object in topo vect
res[cls.load_pos_topo_vect] = cls.global_bus_to_local(
np.array([el.bus_id for el in self._grid.get_loads()]), cls.load_to_subid)
res[cls.gen_pos_topo_vect] = cls.global_bus_to_local(
np.array([el.bus_id for el in self._grid.get_generators()]), cls.gen_to_subid)
if self.__has_storage:
res[cls.storage_pos_topo_vect] = cls.global_bus_to_local(
np.array([el.bus_id for el in self._grid.get_storages()]), cls.storage_to_subid)
lor_glob_bus = np.concatenate((np.array([el.bus1_id for el in self._grid.get_lines()]),
np.array([el.bus1_id for el in self._grid.get_trafos()])))
res[cls.line_or_pos_topo_vect] = cls.global_bus_to_local(lor_glob_bus, cls.line_or_to_subid)
lex_glob_bus = np.concatenate((np.array([el.bus2_id for el in self._grid.get_lines()]),
np.array([el.bus2_id for el in self._grid.get_trafos()])))
res[cls.line_ex_pos_topo_vect] = cls.global_bus_to_local(lex_glob_bus, cls.line_ex_to_subid)
res.flags.writeable = False
# handle shunts (not in topo vect)
if cls.shunts_data_available:
self.sh_bus.flags.writeable = True
sh_bus_global = np.asarray([el.bus_id for el in self._grid.get_shunts()])
self.sh_bus[:] = cls.global_bus_to_local(sh_bus_global,
cls.shunt_to_subid).copy()
self.sh_bus.flags.writeable = False
[docs]
def assert_grid_correct_after_powerflow(self) -> None:
"""
This method is called by the environment. It ensure that the backend remains consistent even after a powerflow
has be run with :func:`Backend.runpf` method.
:return: ``None``
:raise: :class:`grid2op.Exceptions.EnvError` and possibly all of its derived class.
"""
# test the results gives the proper size
super().assert_grid_correct_after_powerflow()
if self._init_pp_backend is not None:
from lightsim2grid._utils import _DoNotUseAnywherePandaPowerBackend # lazy import
_DoNotUseAnywherePandaPowerBackend._clear_grid_dependant_class_attributes()
self._init_pp_backend.__class__ = type(self._init_pp_backend).init_grid(type(self))
self._backend_action_class = _BackendAction.init_grid(type(self))
if self.prod_pu_to_kv is not None:
assert np.isfinite(self.prod_pu_to_kv).all()
if self.load_pu_to_kv is not None:
assert np.isfinite(self.load_pu_to_kv).all()
if self.lines_or_pu_to_kv is not None:
assert np.isfinite(self.lines_or_pu_to_kv).all()
if self.lines_ex_pu_to_kv is not None:
assert np.isfinite(self.lines_ex_pu_to_kv).all()
if self.__has_storage and self.n_storage > 0 and self.storage_pu_to_kv is not None:
assert np.isfinite(self.storage_pu_to_kv).all()
if self._dist_slack_non_renew and not type(self).redispatching_unit_commitment_availble:
raise Grid2OpException("You asked to distribute the slack on renewable generators, "
"yet the environment does not seem to know which generators are "
"renewables or not. Make sure env.redispatching_unit_commitment_availble is True "
"(this include having a 'prods_charac.csv' properly formatted and "
"at the right location)")
def _count_object_per_bus(self):
# should be called only when self.topo_vect and self.shunt_topo_vect are set
# todo factor that more properly to update it when it's modified, and not each time
self.nb_obj_per_bus = np.zeros(2 * self.__nb_bus_before, dtype=dt_int)
arr_ = self.topo_vect[self.load_pos_topo_vect] - 1
# TODO handle -1 here, eventually
arr_ = self.load_to_subid + self.__nb_bus_before * arr_
self.nb_obj_per_bus[arr_] += 1
arr_ = self.topo_vect[self.gen_pos_topo_vect] - 1
# TODO handle -1 here, eventually
arr_ = self.gen_to_subid + self.__nb_bus_before * arr_
self.nb_obj_per_bus[arr_] += 1
arr_ = self.topo_vect[self.line_or_pos_topo_vect]
is_connected = arr_ > 0 # powerline is disconnected
arr_ = self.line_or_to_subid[is_connected] + self.__nb_bus_before * (arr_[is_connected] -1)
self.nb_obj_per_bus[arr_] += 1
arr_ = self.topo_vect[self.line_ex_pos_topo_vect]
is_connected = arr_ > 0 # powerline is disconnected
arr_ = self.line_ex_to_subid[is_connected] + self.__nb_bus_before * (arr_[is_connected] -1)
self.nb_obj_per_bus[arr_] += 1
if type(self).shunts_data_available:
arr_ = self.shunt_topo_vect
is_connected = arr_ > 0
arr_ = self.shunt_to_subid[is_connected] + self.__nb_bus_before * (arr_[is_connected] - 1)
self.nb_obj_per_bus[arr_] += 1
[docs]
def close(self) -> None:
if self._init_pp_backend is not None:
# self.init_pp_backend.close() # should not close it, the same init_pp_backend is used when copied
self._init_pp_backend = None
self._reset_res_pointers()
try:
self._fill_nans()
except AttributeError:
# some attributes are not completely filled
pass
self._grid = None
self.__me_at_init = None
[docs]
def apply_action(self, backendAction: Union["grid2op.Action._backendAction._BackendAction", None]) -> None:
"""
Specific implementation of the method to apply an action modifying a powergrid in the pandapower format.
"""
tick = time.perf_counter()
self._next_pf_fails = None
active_bus, *_, topo__, shunts__ = backendAction()
# change the overall topology
chgt = backendAction.current_topo.changed
self._grid.update_topo(chgt, backendAction.current_topo.values)
if not self._automatically_disconnect:
# the backend cannot change the topology, so
# this holds
self.topo_vect.flags.writeable = True
self.topo_vect[chgt] = backendAction.current_topo.values[chgt]
self.topo_vect.flags.writeable = False
else:
if chgt.sum():
# there has been topological changes, I need
# to perform a BFS to detect possible islanding
self._need_islanding_detection = True
# update the injections
gen_v = backendAction.prod_v.values.copy()
gen_v[~backendAction.prod_v.changed] = 0. # sometimes there were nan here, which lead to a warnings
gen_v /= self.prod_pu_to_kv
try:
self._grid.update_gens_p(backendAction.prod_p.changed,
backendAction.prod_p.values)
self._grid.update_gens_v(backendAction.prod_v.changed,
gen_v)
self._grid.update_loads_p(backendAction.load_p.changed,
backendAction.load_p.values)
self._grid.update_loads_q(backendAction.load_q.changed,
backendAction.load_q.values)
except RuntimeError as exc_:
# see https://github.com/Grid2Op/lightsim2grid/issues/66 (even though it's not a "bug" and has not been replicated)
raise BackendError(f"{exc_}") from exc_
if self.__has_storage:
# print(f"\t (in backend) storage_power {backendAction.storage_power.values[backendAction.storage_power.changed]}") # TODO DEBUG WINDOWS
try:
self._grid.update_storages_p(backendAction.storage_power.changed,
backendAction.storage_power.values)
except RuntimeError as exc_:
# modification of power of disconnected storage has no effect in lightsim2grid
if self.detachment_is_allowed:
# a storage units is allowed to be disconnected in this case
pass
else:
self._next_pf_fails = BackendError("Some storage units would be disconnected")
# handle shunts
if type(self).shunts_data_available:
shunt_p, shunt_q, shunt_bus = backendAction.shunt_p, backendAction.shunt_q, backendAction.shunt_bus
# remember the topology not to need to read it back from the grid
self.sh_bus.flags.writeable = True
self.sh_bus[shunt_bus.changed] = shunt_bus.values[shunt_bus.changed]
self.sh_bus.flags.writeable = False
# shunt topology
# (need to be done before to avoid error like "impossible to set reactive value of a disconnected shunt")
for sh_id, new_bus in shunt_bus:
if new_bus == -1:
self._grid.deactivate_shunt(sh_id)
else:
self._grid.reactivate_shunt(sh_id)
if hasattr(type(self), "local_bus_to_global_int"):
self._grid.change_bus_shunt(sh_id, type(self).local_bus_to_global_int(new_bus, self.shunt_to_subid[sh_id]))
else:
self._grid.change_bus_shunt(sh_id, self.shunt_to_subid[sh_id] + (new_bus == 2) * type(self).n_sub)
for sh_id, new_p in shunt_p:
self._grid.change_p_shunt(sh_id, new_p)
for sh_id, new_q in shunt_q:
self._grid.change_q_shunt(sh_id, new_q)
self._handle_dist_slack()
self._timer_apply_act += time.perf_counter() - tick
# self._grid.tell_solver_need_reset() # bug 128 added
def _handle_dist_slack(self):
if self._dist_slack_non_renew:
# tell lightsim2grid to distribute the slack on non renewable generators
# for this step
self._grid.update_slack_weights(type(self).gen_redispatchable)
if self._gen_slack_id is not None:
# tell lightsim2grid to distribute the slack on all connected generators
# that are tagged as "possible slack" by the user input
# (set when creating the LightSimBackend through `gen_slack_id=XXX`)
self._grid.update_slack_weights_by_id(self._gen_slack_id)
def _fetch_grid_data(self):
beg_test = time.perf_counter()
if self._lineor_res is None:
self._lineor_res = self._grid.get_line_res1_full()
if self._lineex_res is None:
self._lineex_res = self._grid.get_line_res2_full()
if self._load_res is None:
self._load_res = self._grid.get_loads_res_full()
if self._gen_res is None:
self._gen_res = self._grid.get_gen_res_full()
if self._trafo_hv_res is None:
self._trafo_hv_res = self._grid.get_trafo_res1_full()
if self._trafo_lv_res is None:
self._trafo_lv_res = self._grid.get_trafo_res2_full()
if self._storage_res is None:
self._storage_res = self._grid.get_storages_res_full()
if self._shunt_res is None:
self._shunt_res = self._grid.get_shunts_res_full()
self._timer_fetch_data_cpp += time.perf_counter() - beg_test
[docs]
def runpf(self, is_dc : bool=False) -> Tuple[bool, Union[Exception, None]]:
my_exc_ = None
res = False
try:
if self._next_pf_fails is not None:
raise self._next_pf_fails
beg_preproc = time.perf_counter()
if self._need_islanding_detection:
# no OLF-style redistribution of the islanded power here: the set-points
# stay the environment's, the distributed slack absorbs the difference
self._grid.consider_only_main_component(False)
if is_dc:
# somehow, when asked to do a powerflow in DC, pandapower assign Vm to be
# one everywhere...
# But not when it initializes in DC mode... (see below)
self.V = np.ones(self.nb_bus_total, dtype=complex) # * self._grid.get_init_vm_pu()
tick = time.perf_counter()
self._timer_preproc += tick - beg_preproc
V = self._grid.dc_pf(self.V, self.max_it, self.tol)
self._timer_solver += time.perf_counter() - tick
if V.shape[0] == 0:
raise BackendError(f"Divergence of DC powerflow (non connected grid). Detailed error: {self._grid.get_dc_solver().get_error()}")
else:
if (self.V is None) or (self.V.shape[0] == 0):
# create the vector V as it is not created
self.V = np.ones(self.nb_bus_total, dtype=complex) * self._grid.get_init_vm_pu()
if self.initdc:
self._grid.deactivate_result_computation()
# if I init with dc values, it should depends on previous state
self.V[:] = 1. # self._grid.get_init_vm_pu() # see issue 30
# apparently pandapower run a "real" dc powerflow with vm_pu = 1
# when it initialize the AC powerflow,
# print(f"\tLightSimBackend: self.V.shape = {self.V.shape}") # TODO DEBUG WINDOWS
self._debug_Vdc = self._grid.dc_pf(copy.deepcopy(self.V), self.max_it, self.tol)
self._grid.reactivate_result_computation()
if self._debug_Vdc.shape[0] == 0:
raise BackendError(f"Divergence of DC powerflow (non connected grid) at the "
f"initialization of AC powerflow. Detailed error: "
f"{self._grid.get_dc_solver().get_error()}")
V_init = self._debug_Vdc.copy()
else:
V_init = copy.deepcopy(self.V)
tick = time.perf_counter()
self._timer_preproc += tick - beg_preproc
V = self._grid.ac_pf(V_init, self.max_it, self.tol)
self._timer_solver += time.perf_counter() - tick
if V.shape[0] == 0:
raise BackendError(f"Divergence of AC powerflow. Detailed error: {self._grid.get_solver().get_error()}")
beg_postroc = time.perf_counter()
if is_dc:
self.comp_time += self._grid.get_dc_computation_time()
self.timer_gridmodel_xx_pf += self._grid.timer_last_dc_pf
else:
self.comp_time += self._grid.get_computation_time()
# NB get_computation_time returns "time_total_nr", which is
# defined in the powerflow algorithm and not on the linear solver.
# it takes into account everything needed to solve the powerflow
# once everything is passed to the solver.
# It does not take into account the time to format the data in the
# from the GridModel
self.timer_gridmodel_xx_pf += self._grid.timer_last_ac_pf
# timer_gridmodel_xx_pf takes all the time within the gridmodel "ac_pf"
beg_readback = time.perf_counter()
self.V[:] = V
self._fetch_grid_data()
self._allow_modif()
(self.p_or[:self.__nb_powerline],
self.q_or[:self.__nb_powerline],
self.v_or[:self.__nb_powerline],
self.a_or[:self.__nb_powerline],
self.line_or_theta[:self.__nb_powerline]) = self._lineor_res
(self.p_or[self.__nb_powerline:],
self.q_or[self.__nb_powerline:],
self.v_or[self.__nb_powerline:],
self.a_or[self.__nb_powerline:],
self.line_or_theta[self.__nb_powerline:]) = self._trafo_hv_res
(self.p_ex[:self.__nb_powerline],
self.q_ex[:self.__nb_powerline],
self.v_ex[:self.__nb_powerline],
self.a_ex[:self.__nb_powerline],
self.line_ex_theta[:self.__nb_powerline]) = self._lineex_res
(self.p_ex[self.__nb_powerline:],
self.q_ex[self.__nb_powerline:],
self.v_ex[self.__nb_powerline:],
self.a_ex[self.__nb_powerline:],
self.line_ex_theta[self.__nb_powerline:]) = self._trafo_lv_res
self.a_or *= 1000. # kA in lightsim, A expected in grid2op
self.a_ex *= 1000. # kA in lightsim, A expected in grid2op
self.a_or[~np.isfinite(self.a_or)] = 0.
self.v_or[~np.isfinite(self.v_or)] = 0.
self.a_ex[~np.isfinite(self.a_ex)] = 0.
self.v_ex[~np.isfinite(self.v_ex)] = 0.
self.load_p[:], self.load_q[:], self.load_v[:], self.load_theta[:] = self._load_res
self.prod_p[:], self.prod_q[:], self.prod_v[:], self.gen_theta[:] = self._gen_res
if self.__has_storage:
self.storage_p[:], self.storage_q[:], self.storage_v[:], self.storage_theta[:] = self._storage_res
self.storage_v[self.storage_v == -1.] = 0. # voltage is 0. for disconnected elements in grid2op
self._timer_read_data_back += time.perf_counter() - beg_readback
self.next_prod_p[:] = self.prod_p
if self._stop_if_load_disco and ((~np.isfinite(self.load_v)).any() or (self.load_v <= 0.).any()):
disco = (~np.isfinite(self.load_v)) | (self.load_v <= 0.)
load_disco = disco.nonzero()[0]
self._timer_postproc += time.perf_counter() - beg_postroc
raise BackendError(f"At least one load is disconnected (check loads {load_disco})")
if self._stop_if_gen_disco and ((~np.isfinite(self.prod_v)).any() or (self.prod_v <= 0.).any()):
disco = (~np.isfinite(self.prod_v)) | (self.prod_v <= 0.)
gen_disco = disco.nonzero()[0]
self._timer_postproc += time.perf_counter() - beg_postroc
raise BackendError(f"At least one generator is disconnected (check gen {gen_disco})")
if self.__has_storage:
# TODO speed optim here: self._grid.get_storage_target_p() is not "cached"
sto_active = (np.abs(self._grid.get_storage_target_p()) > 0.)
sto_act_disco = (((~np.isfinite(self.storage_v)) & sto_active).any() or
((self.storage_v <= 0.) & sto_active).any()
)
if self._stop_if_storage_disco and sto_act_disco:
disco = ((~np.isfinite(self.storage_v)) | (self.storage_v <= 0.)) & sto_active
sto_disco = disco.nonzero()[0]
self._timer_postproc += time.perf_counter() - beg_postroc
raise BackendError(f"At least one storage unit is disconnected (check storage {sto_disco})")
# TODO storage case of divergence !
if type(self).shunts_data_available:
self._set_shunt_info()
if (np.abs(self.line_or_theta) >= 1e6).any() or (np.abs(self.line_ex_theta) >= 1e6).any():
raise BackendError("Some theta are above 1e6 which should not be happening !")
res = True
# `unset_changes()` is a no-op since lightsim2grid 1.0.0: the grid marks
# its own solver cache in sync after every powerflow. Kept so a backend
# instance whose grid had `allow_cache_reuse(False)` still behaves the
# way it used to.
self._grid.unset_changes()
self._disallow_modif()
if self._need_islanding_detection:
# topology might have changed in the
# powerflow computation
self.topo_vect.flags.writeable = True
self._read_topo_vect(self.topo_vect)
self.topo_vect.flags.writeable = False
self._need_islanding_detection = False
self._timer_postproc += time.perf_counter() - beg_postroc
except Exception as exc_:
# of the powerflow has not converged, results are Nan
self._fill_nans()
self._reset_res_pointers()
res = False
my_exc_ = exc_
if not isinstance(my_exc_, BackendError):
my_exc_ = BackendError(f"Converted the error of type {type(my_exc_)}, message was: {my_exc_}")
if is_dc:
# set back the solver to its previous state
self._grid.change_algorithm(self.__current_algo_type)
self._grid.tell_solver_need_reset()
self._grid.reactivate_result_computation()
if self._automatically_disconnect:
# trigger a check for the connected component only
# if relevant attribute is set
self._need_islanding_detection = True
# TODO grid2op compatibility ! (was a single returned element before storage were introduced)
if self.__has_storage:
res = res, my_exc_
return res
def _aux_get_attr_for_modif_flag(self):
li_attr = [
self.p_or,
self.q_or,
self.v_or,
self.a_or,
self.p_ex,
self.q_ex,
self.v_ex,
self.a_ex,
self.load_p,
self.load_q,
self.load_v,
self.prod_p,
self.next_prod_p,
self.prod_q,
self.prod_v,
self.line_or_theta,
self.line_ex_theta,
self.load_theta,
self.gen_theta]
if type(self).shunts_data_available:
li_attr += [
self.sh_p,
self.sh_q,
self.sh_v,
self.sh_theta,
self.sh_bus,
]
if self.__has_storage:
li_attr += [
self.storage_p,
self.storage_q,
self.storage_v,
self.storage_theta,
]
return li_attr
def _aux_modif_flags(self, li_attr, flag):
for el in li_attr:
el.flags.writeable = flag
def _allow_modif(self):
li_attr = self._aux_get_attr_for_modif_flag()
self._aux_modif_flags(li_attr, True)
def _disallow_modif(self):
li_attr = self._aux_get_attr_for_modif_flag()
self._aux_modif_flags(li_attr, False)
[docs]
def _fill_nans(self):
"""fill the results vectors with nans"""
self._next_pf_fails = None
self._allow_modif()
self.p_or[:] = np.nan
self.q_or[:] = np.nan
self.v_or[:] = np.nan
self.a_or[:] = np.nan
self.p_ex[:] = np.nan
self.q_ex[:] = np.nan
self.v_ex[:] = np.nan
self.a_ex[:] = np.nan
self.load_p[:] = np.nan
self.load_q[:] = np.nan
self.load_v[:] = np.nan
self.prod_p[:] = np.nan
self.next_prod_p[:] = np.nan
self.prod_q[:] = np.nan
self.prod_v[:] = np.nan
self.line_or_theta[:] = np.nan
self.line_ex_theta[:] = np.nan
self.load_theta[:] = np.nan
self.gen_theta[:] = np.nan
if type(self).shunts_data_available:
self.sh_p[:] = np.nan
self.sh_q[:] = np.nan
self.sh_v[:] = np.nan
self.sh_theta[:] = np.nan
self.sh_bus[:] = -1
if self.__has_storage:
self.storage_p[:] = np.nan
self.storage_q[:] = np.nan
self.storage_v[:] = np.nan
self.storage_theta[:] = np.nan
if self._grid is not None:
self.V[:] = self._grid.get_init_vm_pu() # reset the V to its "original" value (see issue 30)
self._reset_res_pointers()
self._disallow_modif()
def _reset_res_pointers(self):
self._lineor_res = None
self._lineex_res = None
self._load_res = None
self._gen_res = None
self._trafo_hv_res = None
self._trafo_lv_res = None
self._storage_res = None
self._shunt_res = None
def __deepcopy__(self, memo):
if hasattr(type(self), "copy_public"):
result = self.copy_public()
else:
result = self.copy()
result._reset_res_pointers()
memo[id(self)] = result
return result
[docs]
def copy(self, orig_grid=None) -> Self:
# i can perform a regular copy, everything has been initialized
mygrid = self._grid
_me_at_init = self.__me_at_init
inippbackend = self._init_pp_backend
if _me_at_init is None and self._grid is not None:
# __me_at_init is defined as being the copy of the grid,
# if it's not defined then i can define it here.
# TODO error in deepcopy, this is why I added "self._grid is not None"
_me_at_init = self._grid.copy()
self._grid = None
self.__me_at_init = None
self._init_pp_backend = None
####################
# res = copy.deepcopy(self) # super slow
res = type(self).__new__(type(self))
# make sure to init the "base class"
# in particular with "new" attributes in future grid2op Backend
res._aux_init_super(self.detailed_infos_for_cascading_failures,
self._can_be_copied,
self.__current_algo_type,
self.max_it,
self.tol,
self._turned_off_pv,
self._dist_slack_non_renew,
self._use_static_gen,
self._loader_method,
self._loader_kwargs,
self._stop_if_load_disco,
self._stop_if_gen_disco,
self._stop_if_storage_disco,
self._automatically_disconnect,
self._gen_slack_id)
# for backward compat (attribute was not necessarily present in early grid2op)
if not hasattr(res, "_can_be_copied"):
res._can_be_copied = self._can_be_copied
res.comp_time = self.comp_time
res.timer_gridmodel_xx_pf = self.timer_gridmodel_xx_pf
# copy the regular attribute
res.__has_storage = self.__has_storage
res.__current_algo_type = self.__current_algo_type # forced here because of special `__`
res.__current_ac_algo_config = copy.deepcopy(self.__current_ac_algo_config)
res.__current_dc_algo_config = copy.deepcopy(self.__current_dc_algo_config)
res.__nb_powerline = self.__nb_powerline
res.__nb_bus_before = self.__nb_bus_before
res.cst_1 = dt_float(1.0)
li_regular_attr = ["comp_time", "can_output_theta", "_is_loaded",
"nb_bus_total", "initdc",
"_big_topo_to_obj", "dim_topo",
"_idx_hack_storage",
"_timer_preproc", "_timer_postproc", "_timer_solver",
"_timer_read_data_back", "_timer_apply_act",
"supported_grid_format",
"max_it", "tol", "_turned_off_pv", "_dist_slack_non_renew",
"_use_static_gen", "_loader_method", "_loader_kwargs",
"_stop_if_load_disco", "_stop_if_gen_disco", "_stop_if_storage_disco",
"_timer_fetch_data_cpp", "_next_pf_fails", "_automatically_disconnect",
"_need_islanding_detection",
"_gen_slack_id", "_grid_laid_out_by_loader",
]
for attr_nm in li_regular_attr:
if hasattr(self, attr_nm):
# this test is needed for backward compatibility with older grid2op version
setattr(res, attr_nm, copy.deepcopy(getattr(self, attr_nm)))
# copy the numpy array
li_attr_npy = ["thermal_limit_a", "_sh_vnkv", "_init_bus_load", "_init_bus_gen",
"_init_bus_lor", "_init_bus_lex", "nb_obj_per_bus", "next_prod_p", "topo_vect",
"shunt_topo_vect", "V", "prod_pu_to_kv", "load_pu_to_kv", "lines_or_pu_to_kv",
"lines_ex_pu_to_kv", "storage_pu_to_kv",
"p_or", "q_or", "v_or", "a_or",
"p_ex", "q_ex", "v_ex", "a_ex",
"load_p", "load_q", "load_v",
"prod_p", "prod_q", "prod_v",
"storage_p", "storage_q", "storage_v",
"sh_p", "sh_q", "sh_v", "sh_bus", "sh_theta",
"line_or_theta", "line_ex_theta", "load_theta", "gen_theta", "storage_theta",
"_debug_Vdc",
]
for attr_nm in li_attr_npy:
if hasattr(self, attr_nm):
# this test is needed for backward compatibility with older grid2op version
setattr(res, attr_nm, copy.deepcopy(getattr(self, attr_nm)))
# copy class attribute for older grid2op version (did not use the class attribute)
if version.parse(grid2op.__version__) <= grid2op_min_cls_attr_ver:
cls_attr = ["env_name",
"n_line", "n_gen", "n_load", "n_sub",
"name_gen", "name_load", "name_line", "name_sub", "name_storage",
"sub_info", "dim_topo",
"load_to_subid", "gen_to_subid", "line_or_to_subid",
"line_ex_to_subid", "load_to_sub_pos", "line_or_to_sub_pos",
"line_ex_to_sub_pos", "load_pos_topo_vect", "gen_pos_topo_vect",
"line_or_pos_topo_vect", "line_ex_pos_topo_vect", "storage_pos_topo_vect",
"shunts_data_available", "shunt_to_subid", "n_shunt", "name_shunt",
"n_storage", "storage_to_subid", "storage_to_sub_pos", "storage_type",
"storage_Emax", "storage_Emin", "storage_max_p_prod", "storage_max_p_absorb",
"storage_marginal_cost", "storage_loss", "storage_charging_efficiency",
"storage_discharging_efficiency",
"alarms_area_names", "alarms_lines_area", "alarms_area_lines"
] + type(self)._li_attr_disp
for attr_nm in cls_attr:
if hasattr(self, attr_nm):
# this test is needed for backward compatibility with other grid2op version
setattr(res, attr_nm, copy.deepcopy(getattr(self, attr_nm)))
###############
# handle the most complicated
if mygrid is not None:
res._grid = mygrid.copy()
# res._fill_nans() # don't do this, data are copied from ref
res._reset_res_pointers()
res._fetch_grid_data()
if orig_grid is not None:
if mygrid is not orig_grid:
raise Grid2OpException("Error in the copy of a LightSimBackend: my_grid is not orig_grid")
else:
res._grid = None
if _me_at_init is not None:
# TODO error in deepcopy, this is why I added "_me_at_init is not None"
res.__me_at_init = _me_at_init.copy() # this is const (but I copy it for "safety")
res._init_pp_backend = inippbackend # this is const
res._backend_action_class = self._backend_action_class # this is const
res.__init_topo_vect = self.__init_topo_vect # this is const
res.__init_shunt_bus = self.__init_shunt_bus # this is const
res.available_default_algorithms = self.available_default_algorithms
res._orig_grid_pypowsybl = self._orig_grid_pypowsybl
# assign back "self" attributes
self._grid = mygrid
self._init_pp_backend = inippbackend
self.__me_at_init = _me_at_init
return res
[docs]
def get_line_status(self) -> np.ndarray:
l_s = self._grid.get_lines_status()
t_s = self._grid.get_trafo_status()
return np.concatenate((l_s, t_s)).astype(dt_bool)
[docs]
def get_line_flow(self) -> np.ndarray:
return self.a_or.copy()
[docs]
def get_topo_vect(self) -> np.ndarray:
return self.topo_vect.copy()
[docs]
def generators_info(self) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
return self.prod_p.copy(), self.prod_q.copy(), self.prod_v.copy()
[docs]
def loads_info(self) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
return self.load_p.copy(), self.load_q.copy(), self.load_v.copy()
[docs]
def lines_or_info(self) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
return self.p_or.copy(), self.q_or.copy(), self.v_or.copy(), self.a_or.copy()
[docs]
def lines_ex_info(self) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
return self.p_ex.copy(), self.q_ex.copy(), self.v_ex.copy(), self.a_ex.copy()
[docs]
def storages_info(self) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
if not self.__has_storage:
raise RuntimeError("Storage units are not supported with your grid2op version. Please upgrade to "
"grid2op >1.5")
return self.storage_p.copy(), self.storage_q.copy(), self.storage_v.copy()
[docs]
def shunt_info(self) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
return self.sh_p.copy(), self.sh_q.copy(), self.sh_v.copy(), self.sh_bus.copy()
def _set_shunt_info(self):
tick = time.perf_counter()
self.sh_p[:], self.sh_q[:], self.sh_v[:], self.sh_theta[:] = self._shunt_res
self.sh_v[self.sh_v == -1.] = 0. # in grid2op disco element have voltage of 0. and -1.
self._timer_read_data_back += time.perf_counter() - tick
# self._timer_fetch_data_cpp += time.perf_counter() - tick
[docs]
def _disconnect_line(self, id_):
if id_ < self.__nb_powerline:
self._grid.deactivate_powerline(id_)
else:
self._grid.deactivate_trafo(id_ - self.__nb_powerline)
self.topo_vect.flags.writeable = True
self.topo_vect[self.line_ex_pos_topo_vect[id_]] = -1
self.topo_vect[self.line_or_pos_topo_vect[id_]] = -1
self.topo_vect.flags.writeable = False
def get_current_algo_type(self) -> AlgorithmType:
return self.__current_algo_type
[docs]
def reset(self,
path : Union[os.PathLike, str],
grid_filename : Optional[Union[os.PathLike, str]]=None) -> None:
self._reset_res_pointers()
self._fill_nans()
self._grid = self.__me_at_init.copy()
self._grid.change_algorithm(self.__current_algo_type)
if self.__current_ac_algo_config is not None:
self._grid.set_ac_algo_config(self._aux_algo_config_from_state(self.__current_ac_algo_config))
if self.__current_dc_algo_config is not None:
self._grid.set_dc_algo_config(self._aux_algo_config_from_state(self.__current_dc_algo_config))
self._handle_turnedoff_pv()
self._grid.tell_solver_need_reset()
self.comp_time = 0.
self.timer_gridmodel_xx_pf = 0.
self._timer_postproc = 0.
self._timer_preproc = 0.
self._timer_solver = 0.
self._timer_read_data_back = 0.
self._timer_fetch_data_cpp = 0.
self._timer_apply_act = 0.
# handle initial topology
if type(self).shunts_data_available:
self.sh_bus.flags.writeable = True
self.sh_bus[:] = self.__init_shunt_bus
self.sh_bus.flags.writeable = False
self.topo_vect.flags.writeable = True
self.topo_vect[:] = self.__init_topo_vect
self.topo_vect.flags.writeable = False
if self._automatically_disconnect:
# automatically disconnect things if needed (and if the option is properly set)
self._need_islanding_detection = True
self._reset_res_pointers()
def _dont_use_global_bus_to_local_legacy(cls, global_bus: np.ndarray, to_sub_id: np.ndarray) -> np.ndarray:
res = (1 * global_bus).astype(dt_int) # make a copy
assert cls.n_busbar_per_sub >= 1, f"cls.n_busbar_per_sub should be >=1, found {cls.n_busbar_per_sub}"
assert cls.n_sub >= 1, f"cls.n_sub should be >=1, found {cls.n_sub}"
for i in range(cls.n_busbar_per_sub):
res[(i * cls.n_sub <= global_bus) & (global_bus < (i+1) * cls.n_sub)] = i + 1
res[global_bus == -1] = -1
return res
def _dont_use_get_action_to_set_legacy(self, *arg, **kwargs):
warnings.warn("DEPRECATION: grid2op <=1.4 is not well supported with lightsim2grid. Lots of bugs have been"
"fixed since then. Please upgrade to grid2op >= 1.10",
DeprecationWarning)
line_status = self.get_line_status()
line_status = 2 * line_status - 1
line_status = line_status.astype(dt_int)
topo_vect = self.get_topo_vect()
prod_p, _, prod_v = self.generators_info()
load_p, load_q, _ = self.loads_info()
complete_action_class = CompleteAction.init_grid(self)
set_me = complete_action_class()
set_me.update({"set_line_status": line_status,
"set_bus": topo_vect})
return set_me
if not hasattr(Backend, "global_bus_to_local"):
# for legacy grid2op
setattr(LightSimBackend, "global_bus_to_local", classmethod(_dont_use_global_bus_to_local_legacy))