# Copyright (c) 2019-2020, 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 Grid2Op, Grid2Op a testbed platform to model sequential decision making in power systems.
import copy
import numpy as np
from typing import Tuple, Union
try:
from typing import Self
except ImportError:
from typing_extensions import Self
from grid2op.Action.baseAction import BaseAction
from grid2op.dtypes import dt_int, dt_bool, dt_float
from grid2op.Space import GridObjects
# TODO see if it can be done in c++ easily
[docs]class ValueStore:
"""
USE ONLY IF YOU WANT TO CODE A NEW BACKEND
.. warning:: /!\\\\ Internal, do not modify, alter, change, override the implementation unless you know what you are doing /!\\\\
If you override them you might even notice some extremely weird behaviour. It's not "on purpose", we are aware of
it but we won't change it (for now at least)
.. warning::
Objects from this class should never be created by anyone except by objects of the :class:`grid2op.Action._backendAction._BackendAction`
when they are created or when instances of `_BackendAction` are process *eg* with :func:`_BackendAction.__call__` or
:func:`_BackendAction.get_loads_bus` etc.
There are two correct uses for this class:
#. by iterating manually with the `for xxx in value_stor_instance: `
#. by checking which objects have been changed (with :attr:`ValueStore.changed`) and then check the
new value of the elements **changed** with :attr:`ValueStore.values` [el_id]
.. danger::
You should never trust the values in :attr:`ValueStore.values` [el_id] if :attr:`ValueStore.changed` [el_id] is `False`.
Access data (values) only when the corresponding "mask" (:attr:`ValueStore.changed`) is `True`.
This is, of course, ensured by default if you use the practical way of iterating through them with:
.. code-block:: python
load_p: ValueStore # a ValueStore object named "load_p"
for load_id, new_p in load_p:
# do something
In this case only "new_p" will be given if corresponding `changed` mask is true.
Attributes
----------
Examples
---------
Say you have a "ValueStore" `val_sto` (in :class:`grid2op.Action._backendAction._BackendAction` you will end up manipulating
pretty much all the time `ValueStore` if you use it correctly, with :func:`_BackendAction.__call__` but also is you call
:func:`_BackendAction.get_loads_bus`, :func:`_BackendAction.get_loads_bus_global`, :func:`_BackendAction.get_gens_bus`, ...)
Basically, the "variables" named `prod_p`, `prod_v`, `load_p`, `load_q`, `storage_p`,
`topo__`, `shunt_p`, `shunt_q`, `shunt_bus`, `backendAction.get_lines_or_bus()`,
`backendAction.get_lines_or_bus_global()`, etc in the doc of :class:`grid2op.Action._backendAction._BackendAction`
are all :class:`ValueStore`.
Recommended usage:
.. code-block:: python
val_sto: ValueStore # a ValueStore object named "val_sto"
for el_id, new_val in val_sto:
# do something
# less abstractly, say `load_p` is a ValueStore:
# for load_id, new_p in load_p:
# do the real changes of load active value in self._grid
# load_id => id of loads for which the active consumption changed
# new_p => new load active consumption for `load_id`
# self._grid.change_load_active_value(load_id, new_p) # fictive example of course...
More advanced / vectorized usage (only do that if you found out your backend was
slow because of the iteration in python above, this is error-prone and in general
might not be worth it...):
.. code-block:: python
val_sto: ValueStore # a ValueStore object named "val_sto"
# less abstractly, say `load_p` is a ValueStore:
# self._grid.change_all_loads_active_value(where_changed=load_p.changed,
new_vals=load_p.values[load_p.changed])
# fictive example of couse, I highly doubt the self._grid
# implements a method named exactly `change_all_loads_active_value`
WARNING, DANGER AHEAD:
Never trust the data in load_p.values[~load_p.changed], they might even be un intialized...
"""
def __init__(self, size, dtype):
## TODO at the init it's mandatory to have everything at "1" here
# if topo is not "fully connected" it will not work
#: :class:`np.ndarray`
#: The new target values to be set in `backend._grid` in `apply_action`
#: never use the values if the corresponding mask is set to `False`
#: (it might be non initialized).
self.values = np.empty(size, dtype=dtype)
#: :class:`np.ndarray` (bool)
#: Mask representing which values (stored in :attr:`ValueStore.values` ) are
#: meaningful. The other values (corresponding to `changed=False` ) are meaningless.
self.changed = np.full(size, dtype=dt_bool, fill_value=False)
#: used internally for iteration
self.last_index = 0
self.__size = size
if issubclass(dtype, dt_int):
self.set_val = self._set_val_int
self.change_val = self._change_val_int
elif issubclass(dtype, dt_float):
self.set_val = self._set_val_float
self.change_val = self._change_val_float
def _set_val_float(self, newvals):
changed_ = np.isfinite(newvals)
self.changed[changed_] = True
self.values[changed_] = newvals[changed_]
def _set_val_int(self, newvals):
changed_ = newvals != 0
self.changed[changed_] = True
self.values[changed_] = newvals[changed_]
def _change_val_int(self, newvals):
changed_ = newvals & (self.values > 0)
self.changed[changed_] = True
self.values[changed_] = (1 - self.values[changed_]) + 2
def _change_val_float(self, newvals):
changed_ = np.abs(newvals) >= 1e-7
self.changed[changed_] = True
self.values[changed_] += newvals[changed_]
def reset(self):
self.changed[:] = False
self.last_index = 0
def change_status(self, switch, lineor_id, lineex_id, old_vect):
if not switch.any():
# nothing is modified so i stop here
return
# changed
changed_ = switch
# make it to ids
id_chg_or = lineor_id[changed_]
id_chg_ex = lineex_id[changed_]
self.changed[id_chg_or] = True
self.changed[id_chg_ex] = True
# disconnect the powerlines
me_or_bus = self.values[id_chg_or]
me_ex_bus = self.values[id_chg_ex]
was_connected = (me_or_bus > 0) | (me_ex_bus > 0)
was_disco = ~was_connected
# it was connected, i disconnect it
self.values[id_chg_or[was_connected]] = -1
self.values[id_chg_ex[was_connected]] = -1
# it was disconnected, i reconnect it
reco_or = id_chg_or[was_disco]
reco_ex = id_chg_ex[was_disco]
self.values[reco_or] = old_vect[reco_or]
self.values[reco_ex] = old_vect[reco_ex]
def set_status(self, set_status, lineor_id, lineex_id, old_vect):
id_or = lineor_id
id_ex = lineex_id
# disco
disco_ = set_status == -1
reco_ = set_status == 1
# make it to ids
id_reco_or = id_or[reco_]
id_reco_ex = id_ex[reco_]
id_disco_or = id_or[disco_]
id_disco_ex = id_ex[disco_]
self.changed[id_reco_or] = True
self.changed[id_reco_ex] = True
self.changed[id_disco_or] = True
self.changed[id_disco_ex] = True
# disconnect the powerlines
self.values[id_disco_or] = -1
self.values[id_disco_ex] = -1
# reconnect the powerlines
# don't consider powerlines that have been already changed with topology
# ie reconnect to the old bus only powerline from which we don't know the status
id_reco_or = id_reco_or[self.values[id_reco_or] < 0]
id_reco_ex = id_reco_ex[self.values[id_reco_ex] < 0]
self.values[id_reco_or] = old_vect[id_reco_or]
self.values[id_reco_ex] = old_vect[id_reco_ex]
def get_line_status(self, lineor_id, lineex_id):
return self.values[lineor_id], self.values[lineex_id]
def update_connected(self, current_values):
indx_conn = current_values.values > 0
self.values[indx_conn] = current_values.values[indx_conn]
def all_changed(self):
self.reset()
self.changed[:] = True
def __getitem__(self, item):
return self.values[item]
def __setitem__(self, key, value):
self.values[key] = value
self.changed[key] = value
def __iter__(self):
return self
def __next__(self):
res = None
while self.last_index < self.values.shape[0]:
if self.changed[self.last_index]:
res = (self.last_index, self.values[self.last_index])
self.last_index += 1
if res is not None:
break
if res is not None:
return res
else:
raise StopIteration
def __len__(self):
return self.__size
[docs] def reorder(self, new_order):
"""reorder the element modified, this is use when converting backends only and should not be use
outside of this usecase"""
self.changed[:] = self.changed[new_order]
self.values[:] = self.values[new_order]
def copy_from_index(self, ref, index):
self.reset()
self.changed[:] = ref.changed[index]
self.values[:] = ref.values[index]
def __copy__(self):
res = type(self)(self.values.shape[0], self.values.dtype.type)
res.values[:] = self.values
res.changed[:] = self.changed
res.last_index = self.last_index
res.__size = self.__size
return res
def __deepcopy__(self, memodict={}):
res = type(self)(self.values.shape[0], self.values.dtype.type)
res.values[:] = self.values
res.changed[:] = self.changed
res.last_index = self.last_index
res.__size = self.__size
return res
[docs] def copy(self, other):
"""deepcopy, shallow or deep, without having to initialize everything again"""
self.values[:] = other.values
self.changed[:] = other.changed
self.last_index = other.last_index
self.__size = other.__size
def force_unchanged(self, mask, local_bus):
to_unchanged = local_bus == -1
to_unchanged[~mask] = False
self.changed[to_unchanged] = False
def register_new_topo(self, current_topo: "ValueStore"):
mask_co = current_topo.values >= 1
self.values[mask_co] = current_topo.values[mask_co]
[docs]class _BackendAction(GridObjects):
"""
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Internal class, use at your own risk.
This class "digest" the players / environment / opponent / voltage controlers "actions",
and transform it to one single "state" that can in turn be process by the backend
in the function :func:`grid2op.Backend.Backend.apply_action`.
.. note::
In a :class:`_BackendAction` only the state of the element that have been modified
by an "entity" (agent, environment, opponent, voltage controler etc.) is given.
We expect the backend to "remember somehow" the state of all the rest.
This is to save a lot of computation time for larger grid.
.. note::
You probably don't need to import the `_BackendAction` class (this is why
we "hide" it),
but the `backendAction` you will receive in `apply_action` is indeed
a :class:`_BackendAction`, hence this documentation.
If you want to use grid2op to develop agents or new time series,
this class should behave transparently for you and you don't really
need to spend time reading its documentation.
If you want to develop in grid2op and code a new backend, you might be interested in:
- :func:`_BackendAction.__call__`
- :func:`_BackendAction.get_loads_bus`
- :func:`_BackendAction.get_loads_bus_global`
- :func:`_BackendAction.get_gens_bus`
- :func:`_BackendAction.get_gens_bus_global`
- :func:`_BackendAction.get_lines_or_bus`
- :func:`_BackendAction.get_lines_or_bus_global`
- :func:`_BackendAction.get_lines_ex_bus`
- :func:`_BackendAction.get_lines_ex_bus_global`
- :func:`_BackendAction.get_storages_bus`
- :func:`_BackendAction.get_storages_bus_global`
- :func:`_BackendAction.get_shunts_bus_global`
And in this case, for usage examples, see the examples available in:
- https://github.com/rte-france/Grid2Op/tree/master/examples/backend_integration: a step by step guide to
code a new backend
- :class:`grid2op.Backend.educPandaPowerBackend.EducPandaPowerBackend` and especially the
:func:`grid2op.Backend.educPandaPowerBackend.EducPandaPowerBackend.apply_action`
- :ref:`create-backend-module` page of the documentation, especially the
:ref:`backend-action-create-backend` section
Otherwise, "TL;DR" (only relevant when you want to implement the :func:`grid2op.Backend.Backend.apply_action`
function, rest is not shown):
.. code-block:: python
def apply_action(self, backendAction: Union["grid2op.Action._backendAction._BackendAction", None]) -> None:
if backendAction is None:
return
(
active_bus,
(prod_p, prod_v, load_p, load_q, storage_p),
topo__,
shunts__,
) = backendAction()
# change the active values of the loads
for load_id, new_p in load_p:
# do the real changes in self._grid
# change the reactive values of the loads
for load_id, new_q in load_q:
# do the real changes in self._grid
# change the active value of generators
for gen_id, new_p in prod_p:
# do the real changes in self._grid
# for the voltage magnitude, pandapower expects pu but grid2op provides kV,
# so we need a bit of change
for gen_id, new_v in prod_v:
# do the real changes in self._grid
# process the topology :
# option 1: you can directly set the element of the grid in the "topo_vect"
# order, for example you can modify in your solver the busbar to which
# element 17 of `topo_vect` is computed (this is necessarily a local view of
# the buses )
for el_topo_vect_id, new_el_bus in topo__:
# connect this object to the `new_el_bus` (local) in self._grid
# OR !!! (use either option 1 or option 2.a or option 2.b - exclusive OR)
# option 2: use "per element type" view (this is usefull)
# if your solver has organized its data by "type" and you can
# easily access "all loads" and "all generators" etc.
# option 2.a using "local view":
# new_bus is either -1, 1, 2, ..., backendAction.n_busbar_per_sub
lines_or_bus = backendAction.get_lines_or_bus()
for line_id, new_bus in lines_or_bus:
# connect "or" side of "line_id" to (local) bus `new_bus` in self._grid
# OR !!! (use either option 1 or option 2.a or option 2.b - exclusive OR)
# option 2.b using "global view":
# new_bus is either 0, 1, 2, ..., backendAction.n_busbar_per_sub * backendAction.n_sub
# (this suppose internally that your solver and grid2op have the same
# "ways" of labelling the buses...)
lines_or_bus = backendAction.get_lines_or_bus_global()
for line_id, new_bus in lines_or_bus:
# connect "or" side of "line_id" to (global) bus `new_bus` in self._grid
# now repeat option a OR b calling the right methods
# for each element types (*eg* get_lines_ex_bus, get_loads_bus, get_gens_bus,
# get_storages_bus for "option a-like")
######## end processing of the topology ###############
# now implement the shunts:
if shunts__ is not None:
shunt_p, shunt_q, shunt_bus = shunts__
if (shunt_p.changed).any():
# p has changed for at least a shunt
for shunt_id, new_shunt_p in shunt_p:
# do the real changes in self._grid
if (shunt_q.changed).any():
# q has changed for at least a shunt
for shunt_id, new_shunt_q in shunt_q:
# do the real changes in self._grid
if (shunt_bus.changed).any():
# at least one shunt has been disconnected
# or has changed the buses
# do like for normal topology with:
# option a -like (using local bus):
for shunt_id, new_shunt_bus in shunt_bus:
...
# OR
# option b -like (using global bus):
shunt_global_bus = backendAction.get_shunts_bus_global()
for shunt_id, new_shunt_bus in shunt_global_bus:
# connect shunt_id to (global) bus `new_shunt_bus` in self._grid
.. warning::
The steps shown here are generic and might not be optimised for your backend. This
is why you probably do not see any of them directly in :class:`grid2op.Backend.PandaPowerBackend`
(where everything is vectorized to make things fast **with pandapower**).
It is probably a good idea to first get this first implementation up and running, passing
all the tests, and then to worry about optimization:
The real problem is that programmers have spent far too much
time worrying about efficiency in the wrong places and at the wrong times;
premature optimization is the root of all evil (or at least most of it)
in programming.
Donald Knuth, "*The Art of Computer Programming*"
"""
[docs] def __init__(self):
"""
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
This is handled by the environment !
"""
GridObjects.__init__(self)
cls = type(self)
# last connected registered
self.last_topo_registered: ValueStore = ValueStore(cls.dim_topo, dtype=dt_int)
# topo at time t
self.current_topo: ValueStore = ValueStore(cls.dim_topo, dtype=dt_int)
# by default everything is on busbar 1
self.last_topo_registered.values[:] = 1
self.current_topo.values[:] = 1
# injection at time t
self.prod_p: ValueStore = ValueStore(cls.n_gen, dtype=dt_float)
self.prod_v: ValueStore = ValueStore(cls.n_gen, dtype=dt_float)
self.load_p: ValueStore = ValueStore(cls.n_load, dtype=dt_float)
self.load_q: ValueStore = ValueStore(cls.n_load, dtype=dt_float)
self.storage_power: ValueStore = ValueStore(cls.n_storage, dtype=dt_float)
self.activated_bus = np.full((cls.n_sub, cls.n_busbar_per_sub), dtype=dt_bool, fill_value=False)
self.big_topo_to_subid: np.ndarray = np.repeat(
list(range(cls.n_sub)), repeats=cls.sub_info
)
# shunts
if cls.shunts_data_available:
self.shunt_p: ValueStore = ValueStore(cls.n_shunt, dtype=dt_float)
self.shunt_q: ValueStore = ValueStore(cls.n_shunt, dtype=dt_float)
self.shunt_bus: ValueStore = ValueStore(cls.n_shunt, dtype=dt_int)
self.current_shunt_bus: ValueStore = ValueStore(cls.n_shunt, dtype=dt_int)
self.current_shunt_bus.values[:] = 1
self._status_or_before: np.ndarray = np.ones(cls.n_line, dtype=dt_int)
self._status_ex_before: np.ndarray = np.ones(cls.n_line, dtype=dt_int)
self._status_or: np.ndarray = np.ones(cls.n_line, dtype=dt_int)
self._status_ex: np.ndarray = np.ones(cls.n_line, dtype=dt_int)
self._loads_bus = None
self._gens_bus = None
self._lines_or_bus = None
self._lines_ex_bus = None
self._storage_bus = None
[docs] def __deepcopy__(self, memodict={}) -> Self:
"""
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
"""
res = type(self)()
# last connected registered
res.last_topo_registered.copy(self.last_topo_registered)
res.current_topo.copy(self.current_topo)
res.prod_p.copy(self.prod_p)
res.prod_v.copy(self.prod_v)
res.load_p.copy(self.load_p)
res.load_q.copy(self.load_q)
res.storage_power.copy(self.storage_power)
res.activated_bus[:, :] = self.activated_bus
# res.big_topo_to_subid[:] = self.big_topo_to_subid # cste
cls = type(self)
if cls.shunts_data_available:
res.shunt_p.copy(self.shunt_p)
res.shunt_q.copy(self.shunt_q)
res.shunt_bus.copy(self.shunt_bus)
res.current_shunt_bus.copy(self.current_shunt_bus)
res._status_or_before[:] = self._status_or_before
res._status_ex_before[:] = self._status_ex_before
res._status_or[:] = self._status_or
res._status_ex[:] = self._status_ex
res._loads_bus = copy.deepcopy(self._loads_bus)
res._gens_bus = copy.deepcopy(self._gens_bus)
res._lines_or_bus = copy.deepcopy(self._lines_or_bus)
res._lines_ex_bus = copy.deepcopy(self._lines_ex_bus)
res._storage_bus = copy.deepcopy(self._storage_bus)
return res
[docs] def __copy__(self) -> Self:
"""
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
"""
res = self.__deepcopy__() # nothing less to do
return res
[docs] def reorder(self, no_load, no_gen, no_topo, no_storage, no_shunt) -> None:
"""
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
This is handled by BackendConverter, do not alter
Reorder the element modified, this is use when converting backends only and should not be use
outside of this usecase
no_* stands for "new order"
"""
self.last_topo_registered.reorder(no_topo)
self.current_topo.reorder(no_topo)
self.prod_p.reorder(no_gen)
self.prod_v.reorder(no_gen)
self.load_p.reorder(no_load)
self.load_q.reorder(no_load)
self.storage_power.reorder(no_storage)
cls = type(self)
if cls.shunts_data_available:
self.shunt_p.reorder(no_shunt)
self.shunt_q.reorder(no_shunt)
self.shunt_bus.reorder(no_shunt)
self.current_shunt_bus.reorder(no_shunt)
[docs] def reset(self) -> None:
"""
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
This is called by the environment, do not alter.
"""
# last known topo
self.last_topo_registered.reset()
# topo at time t
self.current_topo.reset()
# injection at time t
self.prod_p.reset()
self.prod_v.reset()
self.load_p.reset()
self.load_q.reset()
self.storage_power.reset()
# storage unit have their power reset to 0. each step
self.storage_power.changed[:] = True
self.storage_power.values[:] = 0.0
# shunts
cls = type(self)
if cls.shunts_data_available:
self.shunt_p.reset()
self.shunt_q.reset()
self.shunt_bus.reset()
self.current_shunt_bus.reset()
self.last_topo_registered.register_new_topo(self.current_topo)
[docs] def all_changed(self) -> None:
"""
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
This is called by the environment, do not alter.
"""
# last topo
self.last_topo_registered.all_changed()
# topo at time t
self.current_topo.all_changed()
# injection at time t
self.prod_p.all_changed()
self.prod_v.all_changed()
self.load_p.all_changed()
self.load_q.all_changed()
self.storage_power.all_changed()
# TODO handle shunts
# shunts
# if self.shunts_data_available:
# self.shunt_p.all_changed()
# self.shunt_q.all_changed()
# self.shunt_bus.all_changed()
[docs] def set_redispatch(self, new_redispatching):
"""
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
This is called by the environment, do not alter.
"""
self.prod_p.change_val(new_redispatching)
[docs] def _aux_iadd_inj(self, dict_injection):
"""
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Internal implementation of +=
"""
if "load_p" in dict_injection:
tmp = dict_injection["load_p"]
self.load_p.set_val(tmp)
if "load_q" in dict_injection:
tmp = dict_injection["load_q"]
self.load_q.set_val(tmp)
if "prod_p" in dict_injection:
tmp = dict_injection["prod_p"]
self.prod_p.set_val(tmp)
if "prod_v" in dict_injection:
tmp = dict_injection["prod_v"]
self.prod_v.set_val(tmp)
[docs] def _aux_iadd_shunt(self, other):
"""
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Internal implementation of +=
"""
shunts = {}
if type(other).shunts_data_available:
shunts["shunt_p"] = other.shunt_p
shunts["shunt_q"] = other.shunt_q
shunts["shunt_bus"] = other.shunt_bus
arr_ = shunts["shunt_p"]
self.shunt_p.set_val(arr_)
arr_ = shunts["shunt_q"]
self.shunt_q.set_val(arr_)
arr_ = shunts["shunt_bus"]
self.shunt_bus.set_val(arr_)
self.current_shunt_bus.values[self.shunt_bus.changed] = self.shunt_bus.values[self.shunt_bus.changed]
[docs] def _aux_iadd_reconcile_disco_reco(self):
"""
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Internal implementation of +=
"""
disco_or = (self._status_or_before == -1) | (self._status_or == -1)
disco_ex = (self._status_ex_before == -1) | (self._status_ex == -1)
disco_now = (
disco_or | disco_ex
) # a powerline is disconnected if at least one of its extremity is
# added
reco_or = (self._status_or_before == -1) & (self._status_or >= 1)
reco_ex = (self._status_or_before == -1) & (self._status_ex >= 1)
reco_now = reco_or | reco_ex
# Set nothing
set_now = np.zeros_like(self._status_or)
# Force some disconnections
set_now[disco_now] = -1
set_now[reco_now] = 1
self.current_topo.set_status(
set_now,
self.line_or_pos_topo_vect,
self.line_ex_pos_topo_vect,
self.last_topo_registered,
)
[docs] def __iadd__(self, other : BaseAction) -> Self:
"""
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
This is called by the environment, do not alter.
The goal of this function is to "fused" together all the different types
of modifications handled by:
- the Agent
- the opponent
- the time series (part of the environment)
- the voltage controler
It might be called multiple times per step.
Parameters
----------
other: :class:`grid2op.Action.BaseAction`
Returns
-------
The updated state of `self` after the new action `other` has been added to it.
"""
set_status = other._set_line_status
switch_status = other._switch_line_status
set_topo_vect = other._set_topo_vect
switcth_topo_vect = other._change_bus_vect
redispatching = other._redispatch
storage_power = other._storage_power
# I deal with injections
# Ia set the injection
if other._modif_inj:
self._aux_iadd_inj(other._dict_inj)
# Ib change the injection aka redispatching
if other._modif_redispatch:
self.prod_p.change_val(redispatching)
# Ic storage unit
if other._modif_storage:
self.storage_power.set_val(storage_power)
# II shunts
if type(self).shunts_data_available:
self._aux_iadd_shunt(other)
# III line status
# this need to be done BEFORE the topology, as a connected powerline will be connected to their old bus.
# regardless if the status is changed in the action or not.
if other._modif_change_status:
self.current_topo.change_status(
switch_status,
self.line_or_pos_topo_vect,
self.line_ex_pos_topo_vect,
self.last_topo_registered,
)
if other._modif_set_status:
self.current_topo.set_status(
set_status,
self.line_or_pos_topo_vect,
self.line_ex_pos_topo_vect,
self.last_topo_registered,
)
# if other._modif_change_status or other._modif_set_status:
(
self._status_or_before[:],
self._status_ex_before[:],
) = self.current_topo.get_line_status(
self.line_or_pos_topo_vect, self.line_ex_pos_topo_vect
)
# IV topo
if other._modif_change_bus:
self.current_topo.change_val(switcth_topo_vect)
if other._modif_set_bus:
self.current_topo.set_val(set_topo_vect)
# V Force disconnected status
# of disconnected powerlines extremities
self._status_or[:], self._status_ex[:] = self.current_topo.get_line_status(
self.line_or_pos_topo_vect, self.line_ex_pos_topo_vect
)
# At least one disconnected extremity
if other._modif_change_bus or other._modif_set_bus:
self._aux_iadd_reconcile_disco_reco()
return self
[docs] def _assign_0_to_disco_el(self) -> None:
"""
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
This is handled by the environment, do not alter.
Do not consider disconnected elements are modified for there active / reactive / voltage values
"""
cls = type(self)
gen_changed = self.current_topo.changed[cls.gen_pos_topo_vect]
gen_bus = self.current_topo.values[cls.gen_pos_topo_vect]
self.prod_p.force_unchanged(gen_changed, gen_bus)
self.prod_v.force_unchanged(gen_changed, gen_bus)
load_changed = self.current_topo.changed[cls.load_pos_topo_vect]
load_bus = self.current_topo.values[cls.load_pos_topo_vect]
self.load_p.force_unchanged(load_changed, load_bus)
self.load_q.force_unchanged(load_changed, load_bus)
sto_changed = self.current_topo.changed[cls.storage_pos_topo_vect]
sto_bus = self.current_topo.values[cls.storage_pos_topo_vect]
self.storage_power.force_unchanged(sto_changed, sto_bus)
[docs] def __call__(self) -> Tuple[np.ndarray,
Tuple[ValueStore, ValueStore, ValueStore, ValueStore, ValueStore],
ValueStore,
Union[Tuple[ValueStore, ValueStore, ValueStore], None]]:
"""
This function should be called at the top of the :func:`grid2op.Backend.Backend.apply_action`
implementation when you decide to code a new backend.
It processes the state of the backend into a form "easy to use" in the `apply_action` method.
.. danger::
It is mandatory to call it, otherwise some features might not work.
.. warning:: /!\\\\ Do not alter / modify / change / override this implementation /!\\\\
Examples
-----------
A typical implementation of `apply_action` will start with:
.. code-block:: python
def apply_action(self, backendAction: Union["grid2op.Action._backendAction._BackendAction", None]) -> None:
if backendAction is None:
return
(
active_bus,
(prod_p, prod_v, load_p, load_q, storage),
topo__,
shunts__,
) = backendAction()
# process the backend action by updating `self._grid`
Returns
-------
- `active_bus`: matrix with `type(self).n_sub` rows and `type(self).n_busbar_per_bus` columns. Each elements
represents a busbars of the grid. ``False`` indicates that nothing is connected to this busbar and ``True``
means that at least an element is connected to this busbar
- (prod_p, prod_v, load_p, load_q, storage): 5-tuple of Iterable to set the new values of generators, loads and storage units.
- topo: iterable representing the target topology (in local bus, elements are ordered with their
position in the `topo_vect` vector)
"""
self._assign_0_to_disco_el()
injections = (
self.prod_p,
self.prod_v,
self.load_p,
self.load_q,
self.storage_power,
)
topo = self.current_topo
shunts = None
if type(self).shunts_data_available:
shunts = self.shunt_p, self.shunt_q, self.shunt_bus
self._get_active_bus()
return self.activated_bus, injections, topo, shunts
[docs] def get_loads_bus(self) -> ValueStore:
"""
This function might be called in the implementation of :func:`grid2op.Backend.Backend.apply_action`.
It is relevant when your solver expose API by "element types" for example
you get the possibility to set and access all loads at once, all generators at
once and your solver can easily move element from different busbar in a given
substation.
This corresponds to option 2a described (shortly) in :class:`_BackendAction`.
In this setting, this function will give you the "local bus" id for each loads that
have been changed by the agent / time series / voltage controlers / opponent / etc.
.. warning:: /!\\\\ Do not alter / modify / change / override this implementation /!\\\\
.. seealso::
The other related functions:
- :func:`_BackendAction.get_loads_bus`
- :func:`_BackendAction.get_gens_bus`
- :func:`_BackendAction.get_lines_or_bus`
- :func:`_BackendAction.get_lines_ex_bus`
- :func:`_BackendAction.get_storages_bus`
Examples
-----------
A typical use of `get_loads_bus` in `apply_action` is:
.. code-block:: python
def apply_action(self, backendAction: Union["grid2op.Action._backendAction._BackendAction", None]) -> None:
if backendAction is None:
return
(
active_bus,
(prod_p, prod_v, load_p, load_q, storage),
_,
shunts__,
) = backendAction()
# process the backend action by updating `self._grid`
...
# now process the topology (called option 2.a in the doc):
lines_or_bus = backendAction.get_lines_or_bus()
for line_id, new_bus in lines_or_bus:
# connect "or" side of "line_id" to (local) bus `new_bus` in self._grid
self._grid.something(...)
# or
self._grid.something = ...
lines_ex_bus = backendAction.get_lines_ex_bus()
for line_id, new_bus in lines_ex_bus:
# connect "ex" side of "line_id" to (local) bus `new_bus` in self._grid
self._grid.something(...)
# or
self._grid.something = ...
storages_bus = backendAction.get_storages_bus()
for el_id, new_bus in storages_bus:
# connect storage id `el_id` to (local) bus `new_bus` in self._grid
self._grid.something(...)
# or
self._grid.something = ...
gens_bus = backendAction.get_gens_bus()
for el_id, new_bus in gens_bus:
# connect generator id `el_id` to (local) bus `new_bus` in self._grid
self._grid.something(...)
# or
self._grid.something = ...
loads_bus = backendAction.get_loads_bus()
for el_id, new_bus in loads_bus:
# connect generator id `el_id` to (local) bus `new_bus` in self._grid
self._grid.something(...)
# or
self._grid.something = ...
# continue implementation of `apply_action`
"""
if self._loads_bus is None:
self._loads_bus = ValueStore(type(self).n_load, dtype=dt_int)
self._loads_bus.copy_from_index(self.current_topo, type(self).load_pos_topo_vect)
return self._loads_bus
def _aux_to_global(self, value_store, to_subid) -> ValueStore:
value_store = copy.deepcopy(value_store)
value_store.values = type(self).local_bus_to_global(value_store.values, to_subid)
return value_store
[docs] def get_loads_bus_global(self) -> ValueStore:
"""
This function might be called in the implementation of :func:`grid2op.Backend.Backend.apply_action`.
It is relevant when your solver expose API by "element types" for example
you get the possibility to set and access all loads at once, all generators at
once AND you can easily switch element from one "busbars" to another in
the whole grid handled by your solver.
This corresponds to situation 2b described in :class:`_BackendAction`.
In this setting, this function will give you the "local bus" id for each loads that
have been changed by the agent / time series / voltage controlers / opponent / etc.
.. warning:: /!\\\\ Do not alter / modify / change / override this implementation /!\\\\
.. seealso::
The other related functions:
- :func:`_BackendAction.get_loads_bus_global`
- :func:`_BackendAction.get_gens_bus_global`
- :func:`_BackendAction.get_lines_or_bus_global`
- :func:`_BackendAction.get_lines_ex_bus_global`
- :func:`_BackendAction.get_storages_bus_global`
Examples
-----------
A typical use of `get_loads_bus_global` in `apply_action` is:
.. code-block:: python
def apply_action(self, backendAction: Union["grid2op.Action._backendAction._BackendAction", None]) -> None:
if backendAction is None:
return
(
active_bus,
(prod_p, prod_v, load_p, load_q, storage),
_,
shunts__,
) = backendAction()
# process the backend action by updating `self._grid`
...
# now process the topology (called option 2.a in the doc):
lines_or_bus = backendAction.get_lines_or_bus_global()
for line_id, new_bus in lines_or_bus:
# connect "or" side of "line_id" to (global) bus `new_bus` in self._grid
self._grid.something(...)
# or
self._grid.something = ...
lines_ex_bus = backendAction.get_lines_ex_bus_global()
for line_id, new_bus in lines_ex_bus:
# connect "ex" side of "line_id" to (global) bus `new_bus` in self._grid
self._grid.something(...)
# or
self._grid.something = ...
storages_bus = backendAction.get_storages_bus_global()
for el_id, new_bus in storages_bus:
# connect storage id `el_id` to (global) bus `new_bus` in self._grid
self._grid.something(...)
# or
self._grid.something = ...
gens_bus = backendAction.get_gens_bus_global()
for el_id, new_bus in gens_bus:
# connect generator id `el_id` to (global) bus `new_bus` in self._grid
self._grid.something(...)
# or
self._grid.something = ...
loads_bus = backendAction.get_loads_bus_global()
for el_id, new_bus in loads_bus:
# connect generator id `el_id` to (global) bus `new_bus` in self._grid
self._grid.something(...)
# or
self._grid.something = ...
# continue implementation of `apply_action`
"""
tmp_ = self.get_loads_bus()
return self._aux_to_global(tmp_, type(self).load_to_subid)
[docs] def get_gens_bus(self) -> ValueStore:
"""
This function might be called in the implementation of :func:`grid2op.Backend.Backend.apply_action`.
It is relevant when your solver expose API by "element types" for example
you get the possibility to set and access all loads at once, all generators at
once and your solver can easily move element from different busbar in a given
substation.
This corresponds to option 2a described (shortly) in :class:`_BackendAction`.
In this setting, this function will give you the "local bus" id for each generators that
have been changed by the agent / time series / voltage controlers / opponent / etc.
.. warning:: /!\\\\ Do not alter / modify / change / override this implementation /!\\\\
.. seealso::
The other related functions:
- :func:`_BackendAction.get_loads_bus`
- :func:`_BackendAction.get_gens_bus`
- :func:`_BackendAction.get_lines_or_bus`
- :func:`_BackendAction.get_lines_ex_bus`
- :func:`_BackendAction.get_storages_bus`
Examples
---------
Some examples are given in the documentation of :func:`_BackendAction.get_loads_bus`
"""
if self._gens_bus is None:
self._gens_bus = ValueStore(type(self).n_gen, dtype=dt_int)
self._gens_bus.copy_from_index(self.current_topo, type(self).gen_pos_topo_vect)
return self._gens_bus
[docs] def get_gens_bus_global(self) -> ValueStore:
"""
This function might be called in the implementation of :func:`grid2op.Backend.Backend.apply_action`.
It is relevant when your solver expose API by "element types" for example
you get the possibility to set and access all loads at once, all generators at
once AND you can easily switch element from one "busbars" to another in
the whole grid handled by your solver.
This corresponds to situation 2b described in :class:`_BackendAction`.
In this setting, this function will give you the "local bus" id for each loads that
have been changed by the agent / time series / voltage controlers / opponent / etc.
.. warning:: /!\\\\ Do not alter / modify / change / override this implementation /!\\\\
.. seealso::
The other related functions:
- :func:`_BackendAction.get_loads_bus_global`
- :func:`_BackendAction.get_gens_bus_global`
- :func:`_BackendAction.get_lines_or_bus_global`
- :func:`_BackendAction.get_lines_ex_bus_global`
- :func:`_BackendAction.get_storages_bus_global`
Examples
---------
Some examples are given in the documentation of :func:`_BackendAction.get_loads_bus_global`
"""
tmp_ = copy.deepcopy(self.get_gens_bus())
return self._aux_to_global(tmp_, type(self).gen_to_subid)
[docs] def get_lines_or_bus(self) -> ValueStore:
"""
This function might be called in the implementation of :func:`grid2op.Backend.Backend.apply_action`.
It is relevant when your solver expose API by "element types" for example
you get the possibility to set and access all loads at once, all generators at
once and your solver can easily move element from different busbar in a given
substation.
This corresponds to option 2a described (shortly) in :class:`_BackendAction`.
In this setting, this function will give you the "local bus" id for each line (or side) that
have been changed by the agent / time series / voltage controlers / opponent / etc.
.. warning:: /!\\\\ Do not alter / modify / change / override this implementation /!\\\\
.. seealso::
The other related functions:
- :func:`_BackendAction.get_loads_bus`
- :func:`_BackendAction.get_gens_bus`
- :func:`_BackendAction.get_lines_or_bus`
- :func:`_BackendAction.get_lines_ex_bus`
- :func:`_BackendAction.get_storages_bus`
Examples
---------
Some examples are given in the documentation of :func:`_BackendAction.get_loads_bus`
"""
if self._lines_or_bus is None:
self._lines_or_bus = ValueStore(type(self).n_line, dtype=dt_int)
self._lines_or_bus.copy_from_index(
self.current_topo, type(self).line_or_pos_topo_vect
)
return self._lines_or_bus
[docs] def get_lines_or_bus_global(self) -> ValueStore:
"""
This function might be called in the implementation of :func:`grid2op.Backend.Backend.apply_action`.
It is relevant when your solver expose API by "element types" for example
you get the possibility to set and access all loads at once, all generators at
once AND you can easily switch element from one "busbars" to another in
the whole grid handled by your solver.
This corresponds to situation 2b described in :class:`_BackendAction`.
In this setting, this function will give you the "local bus" id for each loads that
have been changed by the agent / time series / voltage controlers / opponent / etc.
.. warning:: /!\\\\ Do not alter / modify / change / override this implementation /!\\\\
.. seealso::
The other related functions:
- :func:`_BackendAction.get_loads_bus_global`
- :func:`_BackendAction.get_gens_bus_global`
- :func:`_BackendAction.get_lines_or_bus_global`
- :func:`_BackendAction.get_lines_ex_bus_global`
- :func:`_BackendAction.get_storages_bus_global`
Examples
---------
Some examples are given in the documentation of :func:`_BackendAction.get_loads_bus_global`
"""
tmp_ = self.get_lines_or_bus()
return self._aux_to_global(tmp_, type(self).line_or_to_subid)
[docs] def get_lines_ex_bus(self) -> ValueStore:
"""
This function might be called in the implementation of :func:`grid2op.Backend.Backend.apply_action`.
It is relevant when your solver expose API by "element types" for example
you get the possibility to set and access all loads at once, all generators at
once and your solver can easily move element from different busbar in a given
substation.
This corresponds to option 2a described (shortly) in :class:`_BackendAction`.
In this setting, this function will give you the "local bus" id for each line (ex side) that
have been changed by the agent / time series / voltage controlers / opponent / etc.
.. warning:: /!\\\\ Do not alter / modify / change / override this implementation /!\\\\
.. seealso::
The other related functions:
- :func:`_BackendAction.get_loads_bus`
- :func:`_BackendAction.get_gens_bus`
- :func:`_BackendAction.get_lines_or_bus`
- :func:`_BackendAction.get_lines_ex_bus`
- :func:`_BackendAction.get_storages_bus`
Examples
---------
Some examples are given in the documentation of :func:`_BackendAction.get_loads_bus`
"""
if self._lines_ex_bus is None:
self._lines_ex_bus = ValueStore(type(self).n_line, dtype=dt_int)
self._lines_ex_bus.copy_from_index(
self.current_topo, type(self).line_ex_pos_topo_vect
)
return self._lines_ex_bus
[docs] def get_lines_ex_bus_global(self) -> ValueStore:
"""
This function might be called in the implementation of :func:`grid2op.Backend.Backend.apply_action`.
It is relevant when your solver expose API by "element types" for example
you get the possibility to set and access all loads at once, all generators at
once AND you can easily switch element from one "busbars" to another in
the whole grid handled by your solver.
This corresponds to situation 2b described in :class:`_BackendAction`.
In this setting, this function will give you the "local bus" id for each loads that
have been changed by the agent / time series / voltage controlers / opponent / etc.
.. warning:: /!\\\\ Do not alter / modify / change / override this implementation /!\\\\
.. seealso::
The other related functions:
- :func:`_BackendAction.get_loads_bus_global`
- :func:`_BackendAction.get_gens_bus_global`
- :func:`_BackendAction.get_lines_or_bus_global`
- :func:`_BackendAction.get_lines_ex_bus_global`
- :func:`_BackendAction.get_storages_bus_global`
Examples
---------
Some examples are given in the documentation of :func:`_BackendAction.get_loads_bus_global`
"""
tmp_ = self.get_lines_ex_bus()
return self._aux_to_global(tmp_, type(self).line_ex_to_subid)
[docs] def get_storages_bus(self) -> ValueStore:
"""
This function might be called in the implementation of :func:`grid2op.Backend.Backend.apply_action`.
It is relevant when your solver expose API by "element types" for example
you get the possibility to set and access all loads at once, all generators at
once and your solver can easily move element from different busbar in a given
substation.
This corresponds to option 2a described (shortly) in :class:`_BackendAction`.
In this setting, this function will give you the "local bus" id for each storage that
have been changed by the agent / time series / voltage controlers / opponent / etc.
.. warning:: /!\\\\ Do not alter / modify / change / override this implementation /!\\\\
.. seealso::
The other related functions:
- :func:`_BackendAction.get_loads_bus`
- :func:`_BackendAction.get_gens_bus`
- :func:`_BackendAction.get_lines_or_bus`
- :func:`_BackendAction.get_lines_ex_bus`
- :func:`_BackendAction.get_storages_bus`
Examples
---------
Some examples are given in the documentation of :func:`_BackendAction.get_loads_bus`
"""
if self._storage_bus is None:
self._storage_bus = ValueStore(type(self).n_storage, dtype=dt_int)
self._storage_bus.copy_from_index(self.current_topo, type(self).storage_pos_topo_vect)
return self._storage_bus
[docs] def get_storages_bus_global(self) -> ValueStore:
"""
This function might be called in the implementation of :func:`grid2op.Backend.Backend.apply_action`.
It is relevant when your solver expose API by "element types" for example
you get the possibility to set and access all loads at once, all generators at
once AND you can easily switch element from one "busbars" to another in
the whole grid handled by your solver.
This corresponds to situation 2b described in :class:`_BackendAction`.
In this setting, this function will give you the "local bus" id for each loads that
have been changed by the agent / time series / voltage controlers / opponent / etc.
.. warning:: /!\\\\ Do not alter / modify / change / override this implementation /!\\\\
.. seealso::
The other related functions:
- :func:`_BackendAction.get_loads_bus_global`
- :func:`_BackendAction.get_gens_bus_global`
- :func:`_BackendAction.get_lines_or_bus_global`
- :func:`_BackendAction.get_lines_ex_bus_global`
- :func:`_BackendAction.get_storages_bus_global`
Examples
---------
Some examples are given in the documentation of :func:`_BackendAction.get_loads_bus_global`
"""
tmp_ = self.get_storages_bus()
return self._aux_to_global(tmp_, type(self).storage_to_subid)
[docs] def get_shunts_bus_global(self) -> ValueStore:
"""
This function might be called in the implementation of :func:`grid2op.Backend.Backend.apply_action`.
It is relevant when your solver expose API by "element types" for example
you get the possibility to set and access all loads at once, all generators at
once AND you can easily switch element from one "busbars" to another in
the whole grid handled by your solver.
This corresponds to situation 2b described in :class:`_BackendAction`.
In this setting, this function will give you the "local bus" id for each loads that
have been changed by the agent / time series / voltage controlers / opponent / etc.
.. warning:: /!\\\\ Do not alter / modify / change / override this implementation /!\\\\
.. seealso::
The other related functions:
- :func:`_BackendAction.get_loads_bus_global`
- :func:`_BackendAction.get_gens_bus_global`
- :func:`_BackendAction.get_lines_or_bus_global`
- :func:`_BackendAction.get_lines_ex_bus_global`
- :func:`_BackendAction.get_storages_bus_global`
Examples
---------
Some examples are given in the documentation of :func:`_BackendAction.get_loads_bus_global`
"""
tmp_ = self.shunt_bus
return self._aux_to_global(tmp_, type(self).shunt_to_subid)
[docs] def _get_active_bus(self) -> None:
"""
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
"""
self.activated_bus[:, :] = False
tmp = self.current_topo.values - 1
is_el_conn = tmp >= 0
self.activated_bus[self.big_topo_to_subid[is_el_conn], tmp[is_el_conn]] = True
if type(self).shunts_data_available:
is_el_conn = self.current_shunt_bus.values >= 0
tmp = self.current_shunt_bus.values - 1
self.activated_bus[type(self).shunt_to_subid[is_el_conn], tmp[is_el_conn]] = True
[docs] def update_state(self, powerline_disconnected) -> None:
"""
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
This is handled by the environment !
Update the internal state. Should be called after the cascading failures.
"""
if (powerline_disconnected >= 0).any():
arr_ = np.zeros(powerline_disconnected.shape, dtype=dt_int)
arr_[powerline_disconnected >= 0] = -1
self.current_topo.set_status(
arr_,
self.line_or_pos_topo_vect,
self.line_ex_pos_topo_vect,
self.last_topo_registered,
)
self.last_topo_registered.update_connected(self.current_topo)
self.current_topo.reset()