/
baseEnv.py
4239 lines (3578 loc) · 178 KB
/
baseEnv.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# 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.
from datetime import datetime
import logging
import time
import copy
import os
import json
from typing import Optional, Tuple, Union, Dict, Any, Literal
import warnings
import numpy as np
from scipy.optimize import (minimize, LinearConstraint)
from abc import ABC, abstractmethod
from grid2op.Observation import (BaseObservation,
ObservationSpace,
HighResSimCounter)
from grid2op.Backend import Backend
from grid2op.dtypes import dt_int, dt_float, dt_bool
from grid2op.Space import GridObjects, RandomObject
from grid2op.Exceptions import (Grid2OpException,
EnvError,
InvalidRedispatching,
GeneratorTurnedOffTooSoon,
GeneratorTurnedOnTooSoon,
AmbiguousActionRaiseAlert,
ImpossibleTopology)
from grid2op.Parameters import Parameters
from grid2op.Reward import BaseReward, RewardHelper
from grid2op.Opponent import OpponentSpace, NeverAttackBudget, BaseOpponent
from grid2op.Action import DontAct, BaseAction, ActionSpace
from grid2op.operator_attention import LinearAttentionBudget
from grid2op.Action._backendAction import _BackendAction
from grid2op.Chronics import ChronicsHandler
from grid2op.Rules import AlwaysLegal, BaseRules, AlwaysLegal
from grid2op.typing_variables import STEP_INFO_TYPING, RESET_OPTIONS_TYPING
# TODO put in a separate class the redispatching function
DETAILED_REDISP_ERR_MSG = (
"\nThis is an attempt to explain why the dispatch did not succeed and caused a game over.\n"
"To compensate the {increase} of loads and / or {decrease} of "
"renewable energy (due to naturl causes but also through curtailment) and / or variation in the storage units, "
"the generators should {increase} their total production of {sum_move:.2f}MW (in total).\n"
"But, if you take into account the generator constraints ({pmax} and {max_ramp_up}) you "
"can have at most {avail_up_sum:.2f}MW.\n"
"Indeed at time t, generators are in state:\n\t{gen_setpoint}\ntheir ramp max is:"
"\n\t{ramp_up}\n and pmax is:\n\t{gen_pmax}\n"
"Wrapping up, each generator can {increase} at {maximum} of:\n\t{avail_up}\n"
"NB: if you did not do any dispatch during this episode, it would have been possible to "
"meet these constraints. This situation is caused by not having enough degree of freedom "
'to "compensate" the variation of the load due to (most likely) an "over usage" of '
"redispatching feature (some generators stuck at {pmax} as a consequence of your "
"redispatching. They can't increase their productions to meet the {increase} in demand or "
"{decrease} of renewables)"
)
BASE_TXT_COPYRIGHT = """# 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.
# THIS FILE HAS BEEN AUTOMATICALLY GENERATED BY "env.generate_classes()"
# WE DO NOT RECOMMEND TO ALTER IT IN ANY WAY
"""
class BaseEnv(GridObjects, RandomObject, ABC):
"""
INTERNAL
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
This class represent some usefull abstraction that is re used by :class:`Environment` and
:class:`grid2op.Observation._Obsenv` for example.
The documentation is showed here to document the common attributes of an "BaseEnvironment".
.. _danger-env-ownership:
Notes
------------------------
Note en environment data ownership
.. danger::
A non pythonic decision has been implemented in grid2op for various reasons: an environment
owns everything created from it.
This means that if you (or the python interpreter) deletes the environment, you might not
use some data generate with this environment.
More precisely, you cannot do something like:
.. code-block:: python
import grid2op
env = grid2op.make("l2rpn_case14_sandbox")
saved_obs = []
obs = env.reset()
saved_obs.append(obs)
obs2, reward, done, info = env.step(env.action_space())
saved_obs.append(obs2)
saved_obs[0].simulate(env.action_space()) # works
del env
saved_obs[0].simulate(env.action_space()) # DOES NOT WORK
It will raise an error like `Grid2OpException EnvError "This environment is closed. You cannot use it anymore."`
This will also happen if you do things inside functions, for example like this:
.. code-block:: python
import grid2op
def foo(manager):
env = grid2op.make("l2rpn_case14_sandbox")
obs = env.reset()
manager.append(obs)
obs2, reward, done, info = env.step(env.action_space())
manager.append(obs2)
manager[0].simulate(env.action_space()) # works
return manager
manager = []
manager = foo(manager)
manager[0].simulate(env.action_space()) # DOES NOT WORK
The same error is raised because the environment `env` is automatically deleted by python when the function `foo` ends
(well it might work on some cases, if the function is called before the variable `env` is actually deleted but you
should not rely on this behaviour.)
Attributes
----------
parameters: :class:`grid2op.Parameters.Parameters`
The parameters of the game (to expose more control on what is being simulated)
with_forecast: ``bool``
Whether the chronics allow to have some kind of "forecast". See :func:`BaseEnv.activate_forceast`
for more information
logger:
TO BE DONE: a way to log what is happening (**currently not implemented**)
time_stamp: ``datetime.datetime``
The actual time stamp of the current observation.
nb_time_step: ``int``
Number of time steps played in the current environment
current_obs: :class:`grid2op.Observation.BaseObservation`
The current observation (or None if it's not intialized)
backend: :class:`grid2op.Backend.Backend`
The backend used to compute the powerflows.
done: ``bool``
Whether the environment is "done". If ``True`` you need to call :func:`Environment.reset` in order
to continue.
current_reward: ``float``
The last computed reward (reward of the current step)
other_rewards: ``dict``
Dictionary with key being the name (identifier) and value being some RewardHelper. At each time step, all the
values will be computed by the :class:`Environment` and the information about it will be returned in the
"reward" key of the "info" dictionnary of the :func:`Environment.step`.
chronics_handler: :class:`grid2op.Chronics.ChronicsHandler`
The object in charge managing the "chronics", which store the information about load and generator for example.
reward_range: ``tuple``
For open ai gym compatibility. It represents the range of the rewards: reward min, reward max
_viewer:
For open ai gym compatibility.
viewer_fig:
For open ai gym compatibility.
_gen_activeprod_t:
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Should be initialized at 0. for "step" to properly recognize it's the first time step of the game
_no_overflow_disconnection: ``bool``
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Whether or not cascading failures are computed or not (TRUE = the powerlines above their thermal limits will
not be disconnected). This is initialized based on the attribute
:attr:`grid2op.Parameters.Parameters.NO_OVERFLOW_DISCONNECTION`.
_timestep_overflow: ``numpy.ndarray``, dtype: int
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Number of consecutive timesteps each powerline has been on overflow.
_nb_timestep_overflow_allowed: ``numpy.ndarray``, dtype: int
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Number of consecutive timestep each powerline can be on overflow. It is usually read from
:attr:`grid2op.Parameters.Parameters.NB_TIMESTEP_POWERFLOW_ALLOWED`.
_hard_overflow_threshold: ``float``
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Number of timestep before an :class:`grid2op.BaseAgent.BaseAgent` can reconnet a powerline that has been
disconnected
by the environment due to an overflow.
_env_dc: ``bool``
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Whether the environment computes the powerflow using the DC approximation or not. It is usually read from
:attr:`grid2op.Parameters.Parameters.ENV_DC`.
_names_chronics_to_backend: ``dict``
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Configuration file used to associated the name of the objects in the backend
(both extremities of powerlines, load or production for
example) with the same object in the data (:attr:`Environment.chronics_handler`). The idea is that, usually
data generation comes from a different software that does not take into account the powergrid infrastructure.
Hence, the same "object" can have a different name. This mapping is present to avoid the need to rename
the "object" when providing data. A more detailed description is available at
:func:`grid2op.ChronicsHandler.GridValue.initialize`.
_env_modification: :class:`grid2op.Action.Action`
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Representation of the actions of the environment for the modification of the powergrid.
_rewardClass: ``type``
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Type of reward used. Should be a subclass of :class:`grid2op.BaseReward.BaseReward`
_init_grid_path: ``str``
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
The path where the description of the powergrid is located.
_game_rules: :class:`grid2op.Rules.RulesChecker`
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
The rules of the game (define which actions are legal and which are not)
_action_space: :class:`grid2op.Action.ActionSpace`
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Helper used to manipulate more easily the actions given to / provided by the :class:`grid2op.Agent.BaseAgent`
(player)
_helper_action_env: :class:`grid2op.Action.ActionSpace`
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Helper used to manipulate more easily the actions given to / provided by the environment to the backend.
_observation_space: :class:`grid2op.Observation.ObservationSpace`
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Helper used to generate the observation that will be given to the :class:`grid2op.BaseAgent`
_reward_helper: :class:`grid2p.BaseReward.RewardHelper`
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
Helper that is called to compute the reward at each time step.
kwargs_observation: ``dict``
TODO
# TODO add the units (eg MW, MWh, MW/time step,etc.) in the redispatching related attributes
"""
ALARM_FILE_NAME = "alerts_info.json"
ALARM_KEY = "fixed"
ALERT_FILE_NAME = "alerts_info.json"
ALERT_KEY = "by_line"
CAN_SKIP_TS = False # each step is exactly one time step
#: this are the keys of the dictionnary `options`
#: that can be used when calling `env.reset(..., options={})`
KEYS_RESET_OPTIONS = {"time serie id"}
def __init__(
self,
init_env_path: os.PathLike,
init_grid_path: os.PathLike,
parameters: Parameters,
voltagecontrolerClass: type,
thermal_limit_a: Optional[np.ndarray] = None,
epsilon_poly: float = 1e-4, # precision of the redispatching algorithm
tol_poly: float = 1e-2, # i need to compute a redispatching if the actual values are "more than tol_poly" the values they should be
other_rewards: dict = None,
with_forecast: bool = True,
opponent_space_type: type = OpponentSpace,
opponent_action_class: type = DontAct,
opponent_class: type = BaseOpponent,
opponent_init_budget: float = 0.0,
opponent_budget_per_ts: float = 0.0,
opponent_budget_class: type = NeverAttackBudget,
opponent_attack_duration: int = 0,
opponent_attack_cooldown: int = 99999,
kwargs_opponent: dict = None,
has_attention_budget: bool = False,
attention_budget_cls: type = LinearAttentionBudget,
kwargs_attention_budget: dict = None,
logger: Optional[logging.Logger] = None,
kwargs_observation: Optional[dict] = None,
observation_bk_class=None, # type of backend for the observation space
observation_bk_kwargs=None, # type of backend for the observation space
highres_sim_counter=None,
update_obs_after_reward=False,
n_busbar=2,
_is_test: bool = False, # TODO not implemented !!
_init_obs: Optional[BaseObservation] =None
):
GridObjects.__init__(self)
RandomObject.__init__(self)
self._n_busbar = n_busbar # env attribute not class attribute !
if other_rewards is None:
other_rewards = {}
if kwargs_attention_budget is None:
kwargs_attention_budget = {}
if kwargs_opponent is None:
kwargs_opponent = {}
self._is_test: bool = _is_test
if logger is None:
self.logger = logging.getLogger(__name__)
self.logger.disabled = True
else:
self.logger: logging.Logger = logger.getChild("grid2op_BaseEnv")
if init_grid_path is not None:
self._init_grid_path: os.PathLike = os.path.abspath(init_grid_path)
else:
self._init_grid_path = None
self._DEBUG: bool = False
self._complete_action_cls: type = None
self.__closed: bool = False # by default the environment is not closed
# specific to power system
if not isinstance(parameters, Parameters):
raise Grid2OpException(
'Parameter "parameters" used to build the Environment should derived form the '
'grid2op.Parameters class, type provided is "{}"'.format(
type(parameters)
)
)
parameters.check_valid() # check the provided parameters are valid
self._parameters: Parameters = copy.deepcopy(parameters)
self.with_forecast: bool = with_forecast
self._forecasts = None
# some timers
self._time_apply_act: float = dt_float(0)
self._time_powerflow: float = dt_float(0)
self._time_extract_obs: float = dt_float(0)
self._time_create_bk_act: float = dt_float(0)
self._time_opponent: float = dt_float(0)
self._time_redisp: float = dt_float(0)
self._time_step: float = dt_float(0)
# data relative to interpolation
self._epsilon_poly: float = dt_float(epsilon_poly)
self._tol_poly: float = dt_float(tol_poly)
# class used for the action spaces
self._helper_action_class: ActionSpace = None
self._helper_observation_class: ActionSpace = None
# and calendar data
self.time_stamp: time.struct_time = None
self.nb_time_step: datetime.timedelta = dt_int(0)
self.delta_time_seconds = None # number of seconds between two consecutive step
# observation
self.current_obs: Optional[BaseObservation] = None
self._line_status: np.ndarray = None
self._ignore_min_up_down_times: bool = self._parameters.IGNORE_MIN_UP_DOWN_TIME
self._forbid_dispatch_off: bool = (
not self._parameters.ALLOW_DISPATCH_GEN_SWITCH_OFF
)
# type of power flow to play
# if True, then it will not disconnect lines above their thermal limits
self._no_overflow_disconnection: bool = (
self._parameters.NO_OVERFLOW_DISCONNECTION
)
self._timestep_overflow: np.ndarray = None
self._nb_timestep_overflow_allowed: np.ndarray = None
self._hard_overflow_threshold: np.ndarray = None
# store actions "cooldown"
self._times_before_line_status_actionable: np.ndarray = None
self._max_timestep_line_status_deactivated: int = (
self._parameters.NB_TIMESTEP_COOLDOWN_LINE
)
self._times_before_topology_actionable: np.ndarray = None
self._max_timestep_topology_deactivated: int = (
self._parameters.NB_TIMESTEP_COOLDOWN_SUB
)
self._nb_ts_reco: int = self._parameters.NB_TIMESTEP_RECONNECTION
# for maintenance operation
self._time_next_maintenance: np.ndarray = None
self._duration_next_maintenance: np.ndarray = None
# hazard (not used outside of this class, information is given in `times_before_line_status_actionable`
self._hazard_duration: np.ndarray = None
self._env_dc = self._parameters.ENV_DC
# redispatching data
self._target_dispatch: np.ndarray = None
self._already_modified_gen: np.ndarray = None
self._actual_dispatch: np.ndarray = None
self._gen_uptime: np.ndarray = None
self._gen_downtime: np.ndarray = None
self._gen_activeprod_t: np.ndarray = None
self._gen_activeprod_t_redisp: np.ndarray = None
self._thermal_limit_a: np.ndarray = thermal_limit_a
self._disc_lines: np.ndarray = None
# store environment modifications
self._injection = None
self._maintenance = None
self._hazards = None
self._env_modification = None
# to use the data
self.done = False
self.current_reward = None
self._helper_action_env: ActionSpace = None
self.chronics_handler : ChronicsHandler = None
self._game_rules = None
self._action_space: ActionSpace = None
self._rewardClass: type = None
self._actionClass: type = None
self._observationClass: type = None
self._legalActClass: type = None
self._observation_space: ObservationSpace = None
self._names_chronics_to_backend: dict = None
self._reward_helper = None
# gym compatibility
self.reward_range = None, None
self._viewer = None
self.viewer_fig = None
# other rewards
self.other_rewards = {}
for k, v in other_rewards.items():
if isinstance(v, type):
if not issubclass(v, BaseReward):
raise Grid2OpException(
'All values of "rewards" key word argument should be classes that inherit '
'from "grid2op.BaseReward"'
)
else:
if not isinstance(v, BaseReward):
raise Grid2OpException(
'All values of "rewards" key word argument should be classes that inherit '
'from "grid2op.BaseReward"'
)
if not isinstance(k, str):
raise Grid2OpException(
'All keys of "rewards" should be of string type.'
)
self.other_rewards[k] = RewardHelper(v, self.logger)
# opponent
self._opponent_action_class = (
opponent_action_class # class of the action of the opponent
)
self._opponent_space_type = opponent_space_type # type of the opponent action space
self._opponent_class = opponent_class # class of the opponent
self._opponent_init_budget = dt_float(opponent_init_budget)
self._opponent_attack_duration = dt_int(opponent_attack_duration)
self._opponent_attack_cooldown = dt_int(opponent_attack_cooldown)
self._opponent_budget_per_ts = dt_float(opponent_budget_per_ts)
self._kwargs_opponent = kwargs_opponent
self._opponent_budget_class = opponent_budget_class
# below initialized by _create_env, above: need to be called
self._opponent_action_space = None
self._compute_opp_budget = None
self._opponent = None
self._oppSpace = None
# voltage
self._voltagecontrolerClass = voltagecontrolerClass
self._voltage_controler = None
# backend action
self._backend_action_class : type = None
self._backend_action : _BackendAction = None
# specific to Basic Env, do not change
self.backend : Backend = None
self.__is_init = False
self.debug_dispatch = False
# to change the parameters
self.__new_param = None
self.__new_forecast_param = None
self.__new_reward_func = None
# storage units
# TODO storage: what to do when self.storage_Emin >0. and self.storage_loss > 0.
# TODO and we have self._storage_current_charge - self.storage_loss < self.storage_Emin
self._storage_current_charge = None # the current storage charge
self._storage_previous_charge = None # the previous storage charge
self._action_storage = None # the storage action performed
self._amount_storage = None # total amount of storage to be dispatched
self._amount_storage_prev = None
self._storage_power = None
self._storage_power_prev = None
# curtailment
self._limit_curtailment = None
self._limit_curtailment_prev = None
self._gen_before_curtailment = None
self._sum_curtailment_mw = None
self._sum_curtailment_mw_prev = None
self._limited_before = 0.0 # TODO curt
# attention budget
self._has_attention_budget = has_attention_budget
self._attention_budget = None
self._attention_budget_cls = attention_budget_cls
self._is_alarm_illegal = False
self._is_alarm_used_in_reward = False
# alert infos
self._is_alert_illegal = False
self._is_alert_used_in_reward = False
self._kwargs_attention_budget = copy.deepcopy(kwargs_attention_budget)
# to ensure self.get_obs() has a reproducible behaviour
self._last_obs = None
# to retrieve previous result (before 1.6.5 the seed of the
# action space or observation space was not done each reset)
self._has_just_been_seeded = False
if kwargs_observation is not None:
self._kwargs_observation = copy.deepcopy(kwargs_observation)
else:
self._kwargs_observation = {}
if init_env_path is not None:
self._init_env_path = os.path.abspath(init_env_path)
else:
self._init_env_path = None
# time_dependant attributes for the "forecast env"
if _init_obs is not None:
self._init_obs = _init_obs.copy()
self._init_obs._obs_env = None
else:
self._init_obs = None
self._observation_bk_class = observation_bk_class
self._observation_bk_kwargs = observation_bk_kwargs
if highres_sim_counter is not None:
self._highres_sim_counter = highres_sim_counter
else:
self._highres_sim_counter = HighResSimCounter()
self._update_obs_after_reward = update_obs_after_reward
# alert
self._last_alert = None
self._time_since_last_alert = None
self._alert_duration= None
self._total_number_of_alert = 0
self._time_since_last_attack = None
self._was_alert_used_after_attack = None
self._attack_under_alert = None
self._is_already_attacked = None
# general things that can be used by the reward
self._reward_to_obs = {}
@property
def highres_sim_counter(self):
return self._highres_sim_counter
@property
def nb_highres_called(self):
return self._highres_sim_counter.nb_highres_called
def _custom_deepcopy_for_copy(self, new_obj, dict_=None):
if self.__closed:
raise RuntimeError("Impossible to make a copy of a closed environment !")
if not self.backend._can_be_copied:
raise RuntimeError("Impossible to copy your environment: the backend "
"class you used cannot be copied.")
RandomObject._custom_deepcopy_for_copy(self, new_obj)
if dict_ is None:
dict_ = {}
new_obj._n_busbar = self._n_busbar
new_obj._init_grid_path = copy.deepcopy(self._init_grid_path)
new_obj._init_env_path = copy.deepcopy(self._init_env_path)
new_obj._DEBUG = self._DEBUG
new_obj._parameters = copy.deepcopy(self._parameters)
new_obj.with_forecast = self.with_forecast
new_obj._forecasts = copy.deepcopy(self._forecasts)
# some timers
new_obj._time_apply_act = self._time_apply_act
new_obj._time_powerflow = self._time_powerflow
new_obj._time_extract_obs = self._time_extract_obs
new_obj._time_create_bk_act = self._time_create_bk_act
new_obj._time_opponent = self._time_opponent
new_obj._time_redisp = self._time_redisp
new_obj._time_step = self._time_step
# data relative to interpolation
new_obj._epsilon_poly = self._epsilon_poly
new_obj._tol_poly = self._tol_poly
#
new_obj._complete_action_cls = copy.deepcopy(self._complete_action_cls)
# define logger
new_obj.logger = copy.deepcopy(self.logger) # TODO does that make any sense ?
# class used for the action spaces
new_obj._helper_action_class = self._helper_action_class # const
new_obj._helper_observation_class = self._helper_observation_class
# and calendar data
new_obj.time_stamp = self.time_stamp
new_obj.nb_time_step = self.nb_time_step
new_obj.delta_time_seconds = self.delta_time_seconds
# observation
if self.current_obs is not None:
new_obj.current_obs = self.current_obs.copy()
# backend
# backend action
new_obj._backend_action_class = self._backend_action_class
new_obj._backend_action = copy.deepcopy(self._backend_action)
# specific to Basic Env, do not change
new_obj.backend = self.backend.copy()
if self._thermal_limit_a is not None:
new_obj.backend.set_thermal_limit(self._thermal_limit_a)
new_obj._thermal_limit_a = copy.deepcopy(self._thermal_limit_a)
new_obj.__is_init = self.__is_init
new_obj.__closed = self.__closed
new_obj.debug_dispatch = self.debug_dispatch
new_obj._line_status = copy.deepcopy(self._line_status)
new_obj._ignore_min_up_down_times = self._ignore_min_up_down_times
new_obj._forbid_dispatch_off = self._forbid_dispatch_off
# type of power flow to play
# if True, then it will not disconnect lines above their thermal limits
new_obj._no_overflow_disconnection = self._no_overflow_disconnection
new_obj._timestep_overflow = copy.deepcopy(self._timestep_overflow)
new_obj._nb_timestep_overflow_allowed = copy.deepcopy(
self._nb_timestep_overflow_allowed
)
new_obj._hard_overflow_threshold = copy.deepcopy(self._hard_overflow_threshold)
# store actions "cooldown"
new_obj._times_before_line_status_actionable = copy.deepcopy(
self._times_before_line_status_actionable
)
new_obj._max_timestep_line_status_deactivated = (
self._max_timestep_line_status_deactivated
)
new_obj._times_before_topology_actionable = copy.deepcopy(
self._times_before_topology_actionable
)
new_obj._max_timestep_topology_deactivated = (
self._max_timestep_topology_deactivated
)
new_obj._nb_ts_reco = self._nb_ts_reco
# for maintenance operation
new_obj._time_next_maintenance = copy.deepcopy(self._time_next_maintenance)
new_obj._duration_next_maintenance = copy.deepcopy(
self._duration_next_maintenance
)
# hazard (not used outside of this class, information is given in `times_before_line_status_actionable`
new_obj._hazard_duration = copy.deepcopy(self._hazard_duration)
new_obj._env_dc = self._env_dc
# redispatching data
new_obj._target_dispatch = copy.deepcopy(self._target_dispatch)
new_obj._already_modified_gen = copy.deepcopy(self._already_modified_gen)
new_obj._actual_dispatch = copy.deepcopy(self._actual_dispatch)
new_obj._gen_uptime = copy.deepcopy(self._gen_uptime)
new_obj._gen_downtime = copy.deepcopy(self._gen_downtime)
new_obj._gen_activeprod_t = copy.deepcopy(self._gen_activeprod_t)
new_obj._gen_activeprod_t_redisp = copy.deepcopy(self._gen_activeprod_t_redisp)
new_obj._disc_lines = copy.deepcopy(self._disc_lines)
# store environment modifications
new_obj._injection = copy.deepcopy(self._injection)
new_obj._maintenance = copy.deepcopy(self._maintenance)
new_obj._hazards = copy.deepcopy(self._hazards)
new_obj._env_modification = copy.deepcopy(self._env_modification)
# to use the data
new_obj.done = self.done
new_obj.current_reward = copy.deepcopy(self.current_reward)
new_obj.chronics_handler = copy.deepcopy(self.chronics_handler)
new_obj._game_rules = copy.deepcopy(self._game_rules)
new_obj._helper_action_env = self._helper_action_env.copy()
new_obj._helper_action_env.legal_action = new_obj._game_rules.legal_action
new_obj._action_space = self._action_space.copy()
new_obj._action_space.legal_action = new_obj._game_rules.legal_action
new_obj._rewardClass = self._rewardClass
new_obj._actionClass = self._actionClass
new_obj._observationClass = self._observationClass
new_obj._legalActClass = self._legalActClass
new_obj._observation_space = self._observation_space.copy(copy_backend=True)
new_obj._observation_space._legal_action = (
new_obj._game_rules.legal_action
) # TODO this does not respect SOLID principles at all !
new_obj._kwargs_observation = copy.deepcopy(self._kwargs_observation)
new_obj._observation_space._ptr_kwargs_observation = new_obj._kwargs_observation
new_obj._names_chronics_to_backend = self._names_chronics_to_backend
new_obj._reward_helper = copy.deepcopy(self._reward_helper)
# gym compatibility
new_obj.reward_range = copy.deepcopy(self.reward_range)
new_obj._viewer = copy.deepcopy(self._viewer)
new_obj.viewer_fig = copy.deepcopy(self.viewer_fig)
# other rewards
new_obj.other_rewards = copy.deepcopy(self.other_rewards)
# opponent
new_obj._opponent_space_type = self._opponent_space_type
new_obj._opponent_action_class = self._opponent_action_class # const
new_obj._opponent_class = self._opponent_class # const
new_obj._opponent_init_budget = self._opponent_init_budget
new_obj._opponent_attack_duration = self._opponent_attack_duration
new_obj._opponent_attack_cooldown = self._opponent_attack_cooldown
new_obj._opponent_budget_per_ts = self._opponent_budget_per_ts
new_obj._kwargs_opponent = copy.deepcopy(self._kwargs_opponent)
new_obj._opponent_budget_class = copy.deepcopy(
self._opponent_budget_class
) # const
new_obj._opponent_action_space = self._opponent_action_space # const
new_obj._compute_opp_budget = self._opponent_budget_class(
self._opponent_action_space
)
# init the opponent
new_obj._opponent = new_obj._opponent_class.__new__(new_obj._opponent_class)
self._opponent._custom_deepcopy_for_copy(
new_obj._opponent, {"partial_env": new_obj, **new_obj._kwargs_opponent}
)
new_obj._oppSpace = new_obj._opponent_space_type(
compute_budget=new_obj._compute_opp_budget,
init_budget=new_obj._opponent_init_budget,
attack_duration=new_obj._opponent_attack_duration,
attack_cooldown=new_obj._opponent_attack_cooldown,
budget_per_timestep=new_obj._opponent_budget_per_ts,
opponent=new_obj._opponent,
)
state_me, state_opp = self._oppSpace._get_state()
new_obj._oppSpace._set_state(state_me)
# voltage
new_obj._voltagecontrolerClass = self._voltagecontrolerClass
new_obj._voltage_controler = self._voltage_controler.copy()
# to change the parameters
new_obj.__new_param = copy.deepcopy(self.__new_param)
new_obj.__new_forecast_param = copy.deepcopy(self.__new_forecast_param)
new_obj.__new_reward_func = copy.deepcopy(self.__new_reward_func)
# storage units
new_obj._storage_current_charge = copy.deepcopy(self._storage_current_charge)
new_obj._storage_previous_charge = copy.deepcopy(self._storage_previous_charge)
new_obj._action_storage = copy.deepcopy(self._action_storage)
new_obj._amount_storage = copy.deepcopy(self._amount_storage)
new_obj._amount_storage_prev = copy.deepcopy(self._amount_storage_prev)
new_obj._storage_power = copy.deepcopy(self._storage_power)
new_obj._storage_power_prev = copy.deepcopy(self._storage_power_prev)
# curtailment
new_obj._limit_curtailment = copy.deepcopy(self._limit_curtailment)
new_obj._limit_curtailment_prev = copy.deepcopy(self._limit_curtailment_prev)
new_obj._gen_before_curtailment = copy.deepcopy(self._gen_before_curtailment)
new_obj._sum_curtailment_mw = copy.deepcopy(self._sum_curtailment_mw)
new_obj._sum_curtailment_mw_prev = copy.deepcopy(self._sum_curtailment_mw_prev)
new_obj._limited_before = copy.deepcopy(self._limited_before)
# attention budget
new_obj._has_attention_budget = self._has_attention_budget
new_obj._attention_budget = copy.deepcopy(self._attention_budget)
new_obj._attention_budget_cls = self._attention_budget_cls # const
new_obj._is_alarm_illegal = copy.deepcopy(self._is_alarm_illegal)
new_obj._is_alarm_used_in_reward = copy.deepcopy(self._is_alarm_used_in_reward)
# alert
new_obj._is_alert_illegal = copy.deepcopy(self._is_alert_illegal)
new_obj._is_alert_used_in_reward = copy.deepcopy(self._is_alert_used_in_reward)
new_obj._kwargs_attention_budget = copy.deepcopy(self._kwargs_attention_budget)
new_obj._last_obs = self._last_obs.copy()
new_obj._has_just_been_seeded = self._has_just_been_seeded
# extra things used by the reward to pass to the obs
new_obj._reward_to_obs = copy.deepcopy(self._reward_to_obs)
# time_dependant attributes for the "forecast env"
if self._init_obs is None:
new_obj._init_obs = None
else:
new_obj._init_obs = self._init_obs.copy()
new_obj._observation_bk_class = self._observation_bk_class
new_obj._observation_bk_kwargs = self._observation_bk_kwargs
# do not forget !
new_obj._is_test = self._is_test
# do not copy it.
new_obj._highres_sim_counter = self._highres_sim_counter
# alert
new_obj._last_alert = copy.deepcopy(self._last_alert)
new_obj._time_since_last_alert = copy.deepcopy(self._time_since_last_alert)
new_obj._alert_duration = copy.deepcopy(self._alert_duration)
new_obj._total_number_of_alert = self._total_number_of_alert
new_obj._time_since_last_attack = copy.deepcopy(self._time_since_last_attack)
new_obj._is_already_attacked = copy.deepcopy(self._is_already_attacked)
new_obj._attack_under_alert = copy.deepcopy(self._attack_under_alert)
new_obj._was_alert_used_after_attack = copy.deepcopy(self._was_alert_used_after_attack)
new_obj._update_obs_after_reward = copy.deepcopy(self._update_obs_after_reward)
def get_path_env(self):
"""
Get the path that allows to create this environment.
It can be used for example in :func:`grid2op.utils.EpisodeStatistics`
to save the information directly inside
the environment data.
"""
if self.__closed:
raise EnvError("This environment is closed, you cannot get its path.")
res = self._init_env_path if self._init_env_path is not None else ""
return res
def _check_alarm_file_consistent(self, dict_):
if (self.ALERT_KEY not in dict_) and (self.ALARM_KEY not in dict_):
raise EnvError(
f'One of {self.ALERT_KEY} or {self.ALARM_KEY} should be present in the alarm data json, for now.'
)
def _set_no_alarm(self):
bk_cls = type(self.backend)
bk_cls.dim_alarms = 0
bk_cls.alarms_area_names = []
bk_cls.alarms_lines_area = {}
bk_cls.alarms_area_lines = []
def load_alarm_data(self):
"""
Internal
.. warning::
/!\\\\ Only valid with "l2rpn_icaps_2021" environment /!\\\\
Notes
------
This is called when the environment class is not created, so i need to read the data of the grid from the
backend.
I cannot use "self.name_line" for example.
This function update the backend INSTANCE. The backend class is then updated in the
:func:`BaseEnv._init_backend`
function with a call to `self.backend.assert_grid_correct()`
Returns
-------
"""
file_alarms = os.path.join(self.get_path_env(), BaseEnv.ALARM_FILE_NAME)
if os.path.exists(file_alarms) and os.path.isfile(file_alarms):
with open(file_alarms, mode="r", encoding="utf-8") as f:
dict_alarm = json.load(f)
self._check_alarm_file_consistent(dict_alarm)
if self.ALARM_KEY not in dict_alarm:
# not an alarm but an alert
self._set_no_alarm()
return # TODO update grid in this case !
nb_areas = len(dict_alarm[self.ALARM_KEY]) # need to be remembered
line_names = {
el: [] for el in self.backend.name_line
} # need to be remembered
area_names = sorted(dict_alarm[self.ALARM_KEY].keys()) # need to be remembered
area_lines = [[] for _ in range(nb_areas)] # need to be remembered
for area_id, area_name in enumerate(area_names):
# check that: all lines in files are in the grid
area = dict_alarm[self.ALARM_KEY][area_name]
for line in area:
if line not in line_names:
raise EnvError(
f"You provided a description of the area of the grid for the alarms, but a "
f'line named "{line}" is present in your file but not in the grid. Please '
f"check the file {file_alarms} and make sure it contains only the line named "
f"{sorted(self.backend.name_line)}."
)
# update the list and dictionary that remembers everything
line_names[line].append(area_name)
area_lines[area_id].append(line)
for line, li_area in line_names.items():
# check that all lines in the grid are in at least one area
if not li_area:
raise EnvError(
f"Line (on the grid) named {line} is not in any areas. This is not supported at "
f"the moment"
)
# every check pass, i update the backend class
bk_cls = type(self.backend)
bk_cls.tell_dim_alarm(nb_areas)
bk_cls.alarms_area_names = copy.deepcopy(area_names)
bk_cls.alarms_lines_area = copy.deepcopy(line_names)
bk_cls.alarms_area_lines = copy.deepcopy(area_lines)
else:
self._set_no_alarm()
def _set_no_alert(self):
bk_cls = type(self.backend)
bk_cls.tell_dim_alert(0)
bk_cls.alertable_line_names = []
bk_cls.alertable_line_ids = np.array([], dtype=dt_int)
def load_alert_data(self):
"""
Internal
Notes
------
This is called to get the alertable lines when the warning is raised "by line"
Returns
-------
"""
file_alarms = os.path.join(self.get_path_env(), BaseEnv.ALERT_FILE_NAME)
if os.path.exists(file_alarms) and os.path.isfile(file_alarms):
with open(file_alarms, mode="r", encoding="utf-8") as f: