# Copyright (c) 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 LightSim2grid, LightSim2grid implements a c++ backend targeting the Grid2Op platform.
__all__ = ["TimeSeriesCPP",
# deprecated
"Computers"]
import numpy as np
import warnings
try:
from grid2op.Chronics import Multifolder, GridStateFromFile # type: ignore
from lightsim2grid.lightSimBackend import LightSimBackend
__all__.append("TimeSerie")
GRID2OP_INSTALLED = True
except ImportError as exc_: # noqa: F841
# grid2Op is not installed
GRID2OP_INSTALLED = False
from lightsim2grid.algorithm import AlgorithmType
from .lightsim2grid_cpp import TimeSeriesCPP
# deprecated
Computers = TimeSeriesCPP
[docs]
class TimeSerie:
"""
This helper class, that only works with grid2op when using a LightSimBackend allows to compute
the flows (at the origin side of the powerline / transformers). It is roughly equivalent to the
grid2op code:
.. code-block:: python
import grid2op
import numpy as np
from grid2op.Parameters import Parameters
from lightsim2grid import LightSimBackend
env_name = ...
param = Parameters()
param.NO_OVERFLOW_DISCONNECTION = True
env = grid2op.make(env_name, param=param, backend=LightSimBackend())
done = False
obs = env.reset()
nb_step = obs.max_step
Vs = np.zeros((nb_step, 2 * env.n_sub), dtype=complex)
As = np.zeros((nb_step, env.n_line), dtype=float)
while not done:
obs, reward, done, info = env.step(env.action_space())
Vs[i, :env.n_sub] = env.backend.V
As[i] = obs.a_or
Compare to the previous code, it avoid all grid2op code and can be more than 15 times
faster (on the case 118).
It also allows to use python threading module, as the c++ computation can be done in different python threads (the GIL is not locked
during the c++ computation).
Examples
----------
It can be used as:
.. code-block:: python
from lightsim2grid import TimeSerie
import grid2op
from lightsim2grid import LightSimBackend
env_name = ...
env = grid2op.make(env_name, param=param, backend=LightSimBackend())
time_series = TimeSerie(env)
res_p, res_a, res_v = time_series.get_flows(scenario_id=..., seed=...)
"""
#: the c++ class this wrapper drives. Overridden by
#: :class:`lightsim2grid.injectionSweep.InjectionSweep`, which shares this entire
#: wrapper and only swaps the underlying computer.
_CPP_CLASS = TimeSeriesCPP
def __init__(self, grid2op_env):
if not GRID2OP_INSTALLED:
raise RuntimeError(f"Impossible to use the python wrapper `{type(self).__name__}` "
"when grid2op is not installed. Please fall back to the "
"c++ version (available in python) with:\n"
f"\tfrom lightsim2grid.timeSerie import {type(self)._CPP_CLASS.__name__}\n"
"and refer to the appropriate documentation.")
from grid2op.Environment import Environment # type: ignore # otherwise i got issues...
if not isinstance(grid2op_env.backend, LightSimBackend):
raise RuntimeError("This class only works with LightSimBackend")
if not isinstance(grid2op_env, Environment):
raise RuntimeError("Please an environment of class \"Environment\", "
"and not \"MultimixEnv\" or \"BaseMultiProcessEnv\"")
self.grid2op_env = grid2op_env.copy()
self.computer = type(self)._CPP_CLASS(self.grid2op_env.backend._grid)
self.prod_p = None
self.load_p = None
self.load_q = None
self.__computed = False
self.available_default_algorithms = self.computer.available_default_algorithms()
if AlgorithmType.NR_KLU in self.available_default_algorithms:
# use the faster KLU if available
self.computer.change_algorithm(AlgorithmType.NR_KLU)
@property
def init_from_n_powerflow(self):
"""Whether to initialize the complex voltages of the first step of the batch with
the results of a "n" powerflow (a powerflow at the current state of the grid)
instead of a flat start. Default: ``False``. Must be set before the computation
actually runs (eg before ``compute_V`` is called); it has no effect on a powerflow
that has already been solved.
"""
return self.computer.init_from_n_powerflow
@init_from_n_powerflow.setter
def init_from_n_powerflow(self, val: bool):
if bool(val) != val:
raise ValueError("The `init_from_n_powerflow` attribute must be a boolean.")
self.computer.init_from_n_powerflow = bool(val)
@property
def keep_vinit_at_group_controlled_buses(self):
"""Whether the buses a voltage-control group regulates (regulated from elsewhere, by
an SVC, or by several controllers at least one of which is remote) keep the magnitude
of the starting voltage instead of being set to their set-point before each solve.
The solution is the same; with ``MaxVoltageChange`` damping it can be reached in far
fewer iterations. Defaults to the grid's own setting (see
:func:`lightsim2grid.lightsim2grid_cpp.LSGrid.set_keep_vinit_at_group_controlled_buses`),
``False`` unless set there.
"""
return self.computer.keep_vinit_at_group_controlled_buses
@keep_vinit_at_group_controlled_buses.setter
def keep_vinit_at_group_controlled_buses(self, val: bool):
if bool(val) != val:
raise ValueError("The `keep_vinit_at_group_controlled_buses` attribute must be a boolean.")
self.computer.keep_vinit_at_group_controlled_buses = bool(val)
@property
def compute_physical_violations(self):
"""Whether every converged step reports the PHYSICAL limits its solution leaves --
the ones whose violation means the step is not a state the grid can reach at all
(``ViolationCategory.PHYSICAL``). Default: ``False``. See
:func:`get_physical_violations`.
Each check is a condition a PowSyBl OpenLoadFlow outer loop acts on, and none
enforced here (nothing is switched PV -> PQ or back, nothing is clamped, no machine
leaves the slack distribution, no step is re-solved):
* the **reactive capability** of every bus whose voltage is held by machines
(``LOW_Q`` / ``HIGH_Q`` on the ``BUS``): did it need more reactive power than the
**sum** of what its voltage-regulating generators, storage units, hvdc converter
stations and voltage-mode SVCs can produce? A machine has no reactive setpoint -- its output is
solved for and never clamped -- so a step can converge asking for reactive power
that does not exist. Per bus, not per machine: the split between the machines of
one bus is a sharing convention rather than something the solver decides.
OpenLoadFlow's ``ReactiveLimits``.
* the **release** of every PQ generator flagged as pinned at a reactive limit
(``LOW_VOLTAGE_AT_MIN_Q`` / ``HIGH_VOLTAGE_AT_MAX_Q`` on the ``GENERATOR``, see
:func:`lightsim2grid.network.LSGrid.set_gen_can_be_pv`): does the bus it would
regulate sit below its target while the machine absorbs all it can (or above it
while it produces all it can)? The other direction, PQ -> PV, of the same
``ReactiveLimits`` loop. ``value`` and ``limit`` in kV, compared with
:attr:`physical_violation_tol_vm_pu`; a step that varies that generator's target
(``modify_gen_v``) is checked against its own target. A grid with no flagged
generator reports nothing here.
* the **switch on** of every idle SVC flagged as carrying a standby automaton
(``LOW_VOLTAGE_SVC_STANDBY`` / ``HIGH_VOLTAGE_SVC_STANDBY`` on the ``SVC``, see
:func:`lightsim2grid.network.LSGrid.set_svc_standby`): does the bus it regulates
sit outside the automaton's voltage thresholds? OpenLoadFlow's
``MonitoringVoltageOuterLoop``. ``value`` and ``limit`` in kV, compared with
:attr:`physical_violation_tol_vm_pu`. A grid with no flagged SVC reports nothing
here.
* the **own bus** of every generator regulating a remote bus, when a realistic range
is set (``LOW_VOLTAGE_REMOTE_CONTROL`` / ``HIGH_VOLTAGE_REMOTE_CONTROL`` on the
``GENERATOR``, see
:func:`lightsim2grid.network.LSGrid.set_remote_voltage_control_vm_range`): does
holding the remote target take it outside that range? OpenLoadFlow's
``ReactiveLimits``, in its robust remote voltage control mode. ``value`` and
``limit`` in kV, compared with :attr:`physical_violation_tol_vm_pu`.
* the **active power** of every angle-droop ("AC emulation") hvdc line still in the
linear regime (``HIGH_P`` on the ``HVDC``): did ``p0 + k.(theta1 - theta2)`` leave
``pmax_1to2_mw`` / ``pmax_2to1_mw``? ``status_droop`` is an *input* of the solve,
so nothing saturates the droop on its own. OpenLoadFlow's
``HvdcAcEmulationLimits``.
* the **release** of every angle-droop hvdc line flagged as frozen at its active
power limit (``HVDC_AC_EMULATION_RELEASE`` on the ``HVDC``, see
:func:`lightsim2grid.network.LSGrid.set_hvdc_ac_emulation_frozen`): would its droop
ask for less than that limit? The other half of the same ``HvdcAcEmulationLimits``
loop. ``side`` the direction it is frozen in, ``value`` and ``limit`` in MW.
* the **active power** of every generator and every storage unit carrying the
**distributed slack** (``LOW_P`` / ``HIGH_P`` on the ``GENERATOR`` /
``STORAGE``): the slack is solved inside the
Jacobian by fixed participation factors that know nothing about limits, so
``target_p + its share of the imbalance`` can land beyond ``min_p_mw`` /
``max_p_mw``. Per machine, unlike the reactive check: the active split is not a
convention, it is the participation factors the caller chose. Needs those limits,
which are optional (:func:`lightsim2grid.network.LSGrid.set_gen_p_limits` /
:func:`lightsim2grid.network.LSGrid.set_storage_p_limits`); a grid without them
reports nothing here. A storage unit's are read -- and its violation reported --
in the *generator* convention, unlike its ``target_p_mw``. OpenLoadFlow's
``DistributedSlack``.
The hvdc and active-power checks need only the bus angles and the slack the step
distributed, so they work in DC too; the reactive one and the voltage ones need an AC
algorithm that publishes its per-bus mismatch (every built-in AC algorithm does) and
``compute`` raises for one that does not. A DC batch reports the active-power checks
alone -- a DC powerflow has no reactive power at all, and no voltage magnitude for the
release, standby SVC and remote voltage control checks, so nothing is hidden by that.
Changing this flag invalidates any previously-computed results, but not the
injections already given to ``modify_*``.
"""
return self.computer.compute_physical_violations
@compute_physical_violations.setter
def compute_physical_violations(self, val: bool):
if bool(val) != val:
raise ValueError("The `compute_physical_violations` attribute must be a boolean.")
val = bool(val)
if val == self.computer.compute_physical_violations:
return # no-op, matches the C++ side (which also no-ops and does not clear)
# the C++ setter drops this batch's base case and results, and keeps the registered
# injections -- so only the python-side "already computed" bookkeeping follows it
self.computer.compute_physical_violations = val
self.__computed = False
@property
def physical_violation_tol_mva(self):
"""Absolute slack on every comparison :attr:`compute_physical_violations` makes, so
that an element resting exactly on its limit is not reported over solver noise: a
violation needs ``value > limit + tol`` (or ``value < limit - tol`` for ``LOW_Q`` and
``HVDC_AC_EMULATION_RELEASE``).
Default: ``1e-4``. In MVA -- one noise floor for both halves, MW and MVAr being the
same scale. Changing it invalidates any previously-computed results.
"""
return self.computer.physical_violation_tol_mva
@physical_violation_tol_mva.setter
def physical_violation_tol_mva(self, val):
try:
val = float(val)
except (TypeError, ValueError):
raise ValueError("The `physical_violation_tol_mva` attribute must be a real number.")
if val == self.computer.physical_violation_tol_mva:
return
self.computer.physical_violation_tol_mva = val # validates, and drops base case + results
self.__computed = False
@property
def physical_violation_tol_vm_pu(self):
"""The same as :attr:`physical_violation_tol_mva`, in pu, for the comparisons
:attr:`compute_physical_violations` makes on a voltage: the PQ -> PV release check
reports a flagged PQ generator (``LSGrid.set_gen_can_be_pv``) whose regulated bus
is below (at ``min_q``) or above (at ``max_q``) its target by more than this
(``LOW_VOLTAGE_AT_MIN_Q`` / ``HIGH_VOLTAGE_AT_MAX_Q``), and the standby SVC check a
flagged idle SVC (``LSGrid.set_svc_standby``) whose regulated bus is outside its
automaton's thresholds by more than this (``LOW_VOLTAGE_SVC_STANDBY`` /
``HIGH_VOLTAGE_SVC_STANDBY``), and the remote voltage control check a remote
controller whose own bus is outside the realistic range by more than this
(``LOW_VOLTAGE_REMOTE_CONTROL`` / ``HIGH_VOLTAGE_REMOTE_CONTROL``). Default: ``1e-4``. Changing it invalidates any
previously-computed results.
"""
return self.computer.physical_violation_tol_vm_pu
@physical_violation_tol_vm_pu.setter
def physical_violation_tol_vm_pu(self, val):
try:
val = float(val)
except (TypeError, ValueError):
raise ValueError("The `physical_violation_tol_vm_pu` attribute must be a real number.")
if val == self.computer.physical_violation_tol_vm_pu:
return
self.computer.physical_violation_tol_vm_pu = val # validates, and drops base case + results
self.__computed = False
[docs]
def get_physical_violations(self):
"""Per step (same order as the ``modify_*`` inputs): the list of ``LimitViolation``
of the physical limits that step's solution leaves. Every entry has ``category ==
ViolationCategory.PHYSICAL`` and one of these shapes:
* ``element_type`` ``BUS``, ``violation_type`` ``LOW_Q`` / ``HIGH_Q``,
``element_id`` the grid bus id, ``value`` the reactive power the machines holding
that bus had to produce (MVAr) and ``limit`` their **summed** capability;
* ``element_type`` ``GENERATOR``, ``violation_type`` ``LOW_VOLTAGE_AT_MIN_Q`` /
``HIGH_VOLTAGE_AT_MAX_Q``, ``element_id`` the generator id, ``value`` the voltage
of the bus that flagged PQ machine would regulate and ``limit`` its target, in kV;
* ``element_type`` ``SVC``, ``violation_type`` ``LOW_VOLTAGE_SVC_STANDBY`` /
``HIGH_VOLTAGE_SVC_STANDBY``, ``element_id`` the svc id, ``value`` the voltage of
the bus that flagged idle standby SVC regulates and ``limit`` the automaton's
threshold, in kV;
* ``element_type`` ``GENERATOR``, ``violation_type`` ``LOW_VOLTAGE_REMOTE_CONTROL`` /
``HIGH_VOLTAGE_REMOTE_CONTROL``, ``element_id`` the generator id, ``value`` the
voltage of that remote controller's own bus and ``limit`` the realistic bound, in kV;
* ``element_type`` ``HVDC``, ``violation_type`` ``HIGH_P``, ``element_id`` the hvdc
line id, ``side`` the direction (1 for 1 -> 2), ``value`` the active power leaving
that side (MW, positive) and ``limit`` that direction's ``pmax``;
* ``element_type`` ``HVDC``, ``violation_type`` ``HVDC_AC_EMULATION_RELEASE``,
``element_id`` the hvdc line id, ``side`` the direction it is frozen in, ``value``
the flow its droop asks for in that direction and ``limit`` the limit it is frozen
at, in MW;
* ``element_type`` ``GENERATOR`` / ``STORAGE``, ``violation_type`` ``LOW_P`` /
``HIGH_P``, ``value`` the machine's converged active power (MW, generator
convention) and ``limit`` its ``min_p_mw`` / ``max_p_mw``.
A step that did not converge has an **empty** entry, not a sentinel -- use
``self.computer.converged_mask()`` to tell that from "converged, no violation".
Requires :attr:`compute_physical_violations` to be ``True`` (raises otherwise).
"""
return self.computer.get_physical_violations()
[docs]
def get_physical_violations_n(self):
"""Same as :func:`get_physical_violations`, for the base ("n") case every step is
solved from (the grid's own state, no injection change). Empty if that solve did not
converge. Requires :attr:`compute_physical_violations` to be ``True``."""
return self.computer.get_physical_violations_n()
[docs]
def get_injections(self, scenario_id=None, seed=None):
"""
This function allows to retrieve the injection of the given scenario, for the given seed
from the grid2op internal environment.
"""
if scenario_id is not None:
self.grid2op_env.set_id(scenario_id)
if seed is not None:
self.grid2op_env.seed(seed)
_ = self.grid2op_env.reset()
self.__computed = False
return self._extract_inj()
[docs]
def compute_V_from_inj(self, prod_p, load_p, load_q, v_init=None, ignore_errors=False):
"""
This function allows to compute the voltages, at each bus given a list of
productions and loads.
We do not recommend to use it directly, as the order of the load or generators might vary !
"""
if len(prod_p.shape) != 2:
raise RuntimeError("prod_p should be a matrix with rows representing time steps "
"and columns representing individual production.")
if len(load_p.shape) != 2:
raise RuntimeError("load_p should be a matrix with rows representing time steps "
"and columns representing individual loads.")
if len(load_q.shape) != 2:
raise RuntimeError("load_q should be a matrix with rows representing time steps "
"and columns representing individual loads.")
if prod_p.shape[0] != load_p.shape[0] or prod_p.shape[0] != load_q.shape[0]:
raise RuntimeError(f"prod_p, load_p and load_q should have the same number of "
f"rows. We found: prod_p.shape[0] = {prod_p.shape[0]}, load_p.shape[0] = {load_p.shape[0]} "
f"and load_q.shape[0] = {load_q.shape[0]}")
if prod_p.shape[1] != self.grid2op_env.n_gen:
raise RuntimeError(f"The number of generators on the grid {self.grid2op_env.n_gen} "
f"is different that the number of columns of the provided prod_p data: "
f"prod_p.shape[1] = {prod_p.shape[1]}")
if load_p.shape[1] != self.grid2op_env.n_load:
raise RuntimeError(f"The number of loads on the grid {self.grid2op_env.n_load} "
f"is different that the number of columns of the provided load_p data: "
f"load_p.shape[1] = {load_p.shape[1]}")
if load_q.shape[1] != self.grid2op_env.n_load:
raise RuntimeError(f"The number of loads on the grid {self.grid2op_env.n_load} "
f"is different that the number of columns of the provided load_q data: "
f"load_q.shape[1] = {load_q.shape[1]}")
if v_init is None:
v_init_comp = self.grid2op_env.backend.V
else:
v_init_comp = 1.0 * v_init # make a copy !
status = self.computer.compute_Vs(prod_p,
np.zeros((prod_p.shape[0], 0)), # no static generators for now !
load_p,
load_q,
v_init_comp,
self.grid2op_env.backend.max_it,
self.grid2op_env.backend.tol)
if status != 1 and not ignore_errors:
# raise an error if the powerflow diverged
raise RuntimeError(f"Some error occurred, the powerflow has diverged after {self.computer.nb_solved()} step(s)")
elif status != 1:
# only raise a warning in this case
warnings.warn(f"Some error occurred, the powerflow has diverged after {self.computer.nb_solved()} step(s)")
Vs = self.computer.get_voltages().copy() # If I don't copy, lazy eval may break stuff...
# eg test_time_series_dc.py does behave stochastically
self.__computed = True
return Vs
[docs]
def modify_gen_p(self, gen_p):
"""Per-step active generator setpoints, shape ``(n_simul, n_gen)``. Part of the
new setter-based API (see :func:`compute`); an alternative to
:func:`compute_V_from_inj`, not required if you use that call instead."""
gen_p = np.asarray(gen_p)
if gen_p.ndim != 2:
raise RuntimeError("gen_p should be a matrix with rows representing time steps "
"and columns representing individual production.")
if gen_p.shape[1] != self.grid2op_env.n_gen:
raise RuntimeError(f"The number of generators on the grid ({self.grid2op_env.n_gen}) "
f"differs from the number of columns of gen_p ({gen_p.shape[1]}).")
self.computer.modify_gen_p(gen_p)
self.__computed = False
[docs]
def modify_sgen_p(self, sgen_p):
"""Per-step active static generator setpoints, shape ``(n_simul, n_sgen)``. See
:func:`modify_gen_p`."""
sgen_p = np.asarray(sgen_p)
if sgen_p.ndim != 2:
raise RuntimeError("sgen_p should be a matrix with rows representing time steps "
"and columns representing individual static generation.")
n_sgen = len(self.grid2op_env.backend._grid.get_static_generators())
if sgen_p.shape[1] != n_sgen:
raise RuntimeError(f"The number of static generators on the grid ({n_sgen}) "
f"differs from the number of columns of sgen_p ({sgen_p.shape[1]}).")
self.computer.modify_sgen_p(sgen_p)
self.__computed = False
[docs]
def modify_load_p(self, load_p):
"""Per-step active load setpoints, shape ``(n_simul, n_load)``. See
:func:`modify_gen_p`."""
load_p = np.asarray(load_p)
if load_p.ndim != 2:
raise RuntimeError("load_p should be a matrix with rows representing time steps "
"and columns representing individual loads.")
if load_p.shape[1] != self.grid2op_env.n_load:
raise RuntimeError(f"The number of loads on the grid ({self.grid2op_env.n_load}) "
f"differs from the number of columns of load_p ({load_p.shape[1]}).")
self.computer.modify_load_p(load_p)
self.__computed = False
[docs]
def modify_load_q(self, load_q):
"""Per-step reactive load setpoints, shape ``(n_simul, n_load)``. See
:func:`modify_gen_p`."""
load_q = np.asarray(load_q)
if load_q.ndim != 2:
raise RuntimeError("load_q should be a matrix with rows representing time steps "
"and columns representing individual loads.")
if load_q.shape[1] != self.grid2op_env.n_load:
raise RuntimeError(f"The number of loads on the grid ({self.grid2op_env.n_load}) "
f"differs from the number of columns of load_q ({load_q.shape[1]}).")
self.computer.modify_load_q(load_q)
self.__computed = False
[docs]
def modify_gen_v(self, gen_v):
"""Per-step generator target voltage magnitude, shape ``(n_simul, n_gen)``, in
pu (``vm_pu``), NOT kV. Unlike :func:`modify_gen_p`/:func:`modify_load_p`/
:func:`modify_load_q`, this does NOT feed the injection (Sbus) -- it only
re-seeds ``|V|`` at each voltage-regulating generator's regulated bus before
that step's solve. See :func:`modify_gen_p`."""
gen_v = np.asarray(gen_v)
if gen_v.ndim != 2:
raise RuntimeError("gen_v should be a matrix with rows representing time steps "
"and columns representing individual production.")
if gen_v.shape[1] != self.grid2op_env.n_gen:
raise RuntimeError(f"The number of generators on the grid ({self.grid2op_env.n_gen}) "
f"differs from the number of columns of gen_v ({gen_v.shape[1]}).")
self.computer.modify_gen_v(gen_v)
self.__computed = False
[docs]
def compute(self, v_init=None, max_iter=None, tol=None, ignore_errors=False):
"""
Run the batch using whatever was set by :func:`modify_gen_p` / :func:`modify_sgen_p`
/ :func:`modify_load_p` / :func:`modify_load_q` (the new setter-based API -- an
alternative to the single bundled :func:`compute_V_from_inj` call). ``max_iter`` /
``tol`` default to the backend's own values when not given.
"""
if v_init is None:
v_init_comp = self.grid2op_env.backend.V
else:
v_init_comp = 1.0 * v_init # make a copy !
if max_iter is None:
max_iter = self.grid2op_env.backend.max_it
if tol is None:
tol = self.grid2op_env.backend.tol
self.computer.compute(v_init_comp, max_iter, tol)
status = self.computer.get_status()
if status != 1 and not ignore_errors:
raise RuntimeError(f"Some error occurred, the powerflow has diverged after {self.computer.nb_solved()} step(s)")
elif status != 1:
warnings.warn(f"Some error occurred, the powerflow has diverged after {self.computer.nb_solved()} step(s)")
self.__computed = True
return self.computer.get_voltages().copy()
[docs]
def compute_V(self, scenario_id=None, seed=None, v_init=None, ignore_errors=False):
"""
This function allows to retrieve the complex voltage at each bus of the grid for each step.
.. warning:: Topology fixed = no maintenance, no attacks, etc.
As the topology is fixed, this class does not allow to simulate the effect of maintenance or attacks !
"""
prod_p, load_p, load_q = self.get_injections(scenario_id=scenario_id, seed=seed)
Vs = self.compute_V_from_inj(prod_p, load_p, load_q, v_init, ignore_errors)
return Vs
[docs]
def compute_A(self):
"""
This function returns the current flows (in Amps, A) at the origin (for powerline) / high voltage (for transformer)
side
It does not recompute the voltages at each buses, it uses the information get from `compute_V` and
This is why you must call `compute_V(...)` first !
"""
if not self.__computed:
raise RuntimeError("This function can only be used if compute_V has been sucessfully called")
ampss = self.computer.compute_flows()
return 1000. * ampss
[docs]
def compute_P(self):
"""
This function returns the active power flows (in MW) at the origin (for powerline) / high voltage (for transformer)
side
It does not recompute the voltages at each buses, it uses the information get from `compute_V` and
This is why you must call `compute_V(...)` first !
"""
if not self.__computed:
raise RuntimeError("This function can only be used if compute_V has been sucessfully called")
mws = 1. * self.computer.compute_power_flows() # If I don't copy, lazy eval may break stuff...
# eg test_time_series_dc.py does behave stochastically
return mws
[docs]
def get_flows(self, scenario_id=None, seed=None, v_init=None, ignore_errors=False):
"""
Retrieve the flows for each step simulated.
Each row of the resulting flow matrix will correspond to a step.
Examples
--------
.. code-block:: python
import grid2op
from lightsim2grid import TimeSerie
from lightsim2grid import LightSimBackend
env_name = ...
env = grid2op.make(env_name, backend=LightSimBackend())
timeserie = TimeSerie(env)
res_p, res_a, res_v = timeserie.get_flows(scenario_id, seed, v_init, ignore_errors)
# in this results, then
# res_a[row_id] will be the flows, on all powerline corresponding to the `row_id` contingency.
# you can retrieve it with `security_analysis.contingency_order[row_id]`
"""
Vs = self.compute_V(scenario_id, seed, v_init, ignore_errors)
amps = self.compute_A()
Ps = self.compute_P()
return Ps, amps, Vs
[docs]
def clear(self):
"""
Clear everything, as if nothing has been computed
"""
self.computer.clear()
self.__computed = False
self.prod_p = None
self.load_p = None
self.load_q = None
[docs]
def close(self):
"""permanently close the object"""
self.grid2op_env.close()
self.clear()
self.computer.close()
def _extract_inj(self):
data_loader = None
if isinstance(self.grid2op_env.chronics_handler.real_data, Multifolder):
data_loader = self.grid2op_env.chronics_handler.real_data.data
else:
data_loader = self.grid2op_env.chronics_handler.data
if not isinstance(data_loader, GridStateFromFile):
raise RuntimeError("This function only work with chronics coming from files at the moment")
self.prod_p = 1.0 * data_loader.prod_p
self.load_p = 1.0 * data_loader.load_p
self.load_q = 1.0 * data_loader.load_q
return self.prod_p, self.load_p, self.load_q