-
Notifications
You must be signed in to change notification settings - Fork 112
/
baseObservation.py
4784 lines (3972 loc) · 201 KB
/
baseObservation.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.
import copy
import datetime
import warnings
import networkx
from abc import abstractmethod
import numpy as np
from scipy.sparse import csr_matrix
from typing import Optional
from packaging import version
from typing import Dict, Union, Tuple, List, Optional, Any, Literal
try:
from typing import Self
except ImportError:
from typing_extensions import Self
import grid2op # for type hints
from grid2op.typing_variables import STEP_INFO_TYPING
from grid2op.dtypes import dt_int, dt_float, dt_bool
from grid2op.Exceptions import (
Grid2OpException,
NoForecastAvailable,
BaseObservationError,
)
from grid2op.Space import GridObjects
# TODO have a method that could do "forecast" by giving the _injection by the agent,
# TODO if he wants to make custom forecasts
# TODO fix "bug" when action not initalized it should return nan in to_vect
# TODO be consistent with gen_* and prod_* also in dictionaries
ERROR_ONLY_SINGLE_EL = "You can only the inspect the effect of an action on one single element"
class BaseObservation(GridObjects):
"""
Basic class representing an observation.
All observation must derive from this class and implement all its abstract methods.
Attributes
----------
action_helper: :class:`grid2op.Action.ActionSpace`
A representation of the possible action space.
year: ``int``
The current year
month: ``int``
The current month (1 = january, 12 = december)
day: ``int``
The current day of the month (1 = first day of the month)
hour_of_day: ``int``
The current hour of the day (from O to 23)
minute_of_hour: ``int``
The current minute of the current hour (from 0 to 59)
day_of_week: ``int``
The current day of the week (monday = 0 and sunday = 6)
support_theta: ``bool``
This flag indicates whether the backend supports the retrieval of the
voltage angle. If so (which is the case for most backend) then
some supplementary attributes are available, such as
:attr:`BaseObservation.gen_theta`,
:attr:`BaseObservation.load_theta`,
:attr:`BaseObservation.storage_theta`,
:attr:`BaseObservation.theta_or` or
:attr:`BaseObservation.theta_ex` .
gen_p: :class:`numpy.ndarray`, dtype:float
The active production value of each generator (expressed in MW).
(the old name "prod_p" is still usable)
gen_q: :class:`numpy.ndarray`, dtype:float
The reactive production value of each generator (expressed in MVar).
(the old name "prod_q" is still usable)
gen_v: :class:`numpy.ndarray`, dtype:float
The voltage magnitude of the bus to which each generator is connected (expressed in kV).
(the old name "prod_v" is still usable)
gen_theta: :class:`numpy.ndarray`, dtype:float
The voltage angle (in degree) of the bus to which each generator is
connected. Only availble if the backend supports the retrieval of
voltage angles (see :attr:`BaseObservation.support_theta`).
load_p: :class:`numpy.ndarray`, dtype:float
The active load value of each consumption (expressed in MW).
load_q: :class:`numpy.ndarray`, dtype:float
The reactive load value of each consumption (expressed in MVar).
load_v: :class:`numpy.ndarray`, dtype:float
The voltage magnitude of the bus to which each consumption is connected (expressed in kV).
load_theta: :class:`numpy.ndarray`, dtype:float
The voltage angle (in degree) of the bus to which each consumption
is connected. Only availble if the backend supports the retrieval of
voltage angles (see :attr:`BaseObservation.support_theta`).
p_or: :class:`numpy.ndarray`, dtype:float
The active power flow at the origin side of each powerline (expressed in MW).
q_or: :class:`numpy.ndarray`, dtype:float
The reactive power flow at the origin side of each powerline (expressed in MVar).
v_or: :class:`numpy.ndarray`, dtype:float
The voltage magnitude at the bus to which the origin side of each powerline is connected (expressed in kV).
theta_or: :class:`numpy.ndarray`, dtype:float
The voltage angle at the bus to which the origin side of each powerline
is connected (expressed in degree). Only availble if the backend supports the retrieval of
voltage angles (see :attr:`BaseObservation.support_theta`).
a_or: :class:`numpy.ndarray`, dtype:float
The current flow at the origin side of each powerline (expressed in A).
p_ex: :class:`numpy.ndarray`, dtype:float
The active power flow at the extremity side of each powerline (expressed in MW).
q_ex: :class:`numpy.ndarray`, dtype:float
The reactive power flow at the extremity side of each powerline (expressed in MVar).
v_ex: :class:`numpy.ndarray`, dtype:float
The voltage magnitude at the bus to which the extremity side of each powerline is connected (expressed in kV).
theta_ex: :class:`numpy.ndarray`, dtype:float
The voltage angle at the bus to which the extremity side of each powerline
is connected (expressed in degree). Only availble if the backend supports the retrieval of
voltage angles (see :attr:`BaseObservation.support_theta`).
a_ex: :class:`numpy.ndarray`, dtype:float
The current flow at the extremity side of each powerline (expressed in A).
rho: :class:`numpy.ndarray`, dtype:float
The capacity of each powerline. It is defined at the observed current flow divided by the thermal limit of each
powerline (no unit)
topo_vect: :class:`numpy.ndarray`, dtype:int
For each object (load, generator, ends of a powerline) it gives on which bus this object is connected
in its substation. See :func:`grid2op.Backend.Backend.get_topo_vect` for more information.
line_status: :class:`numpy.ndarray`, dtype:bool
Gives the status (connected / disconnected) for every powerline (``True`` at position `i` means the powerline
`i` is connected)
timestep_overflow: :class:`numpy.ndarray`, dtype:int
Gives the number of time steps since a powerline is in overflow.
time_before_cooldown_line: :class:`numpy.ndarray`, dtype:int
For each powerline, it gives the number of time step the powerline is unavailable due to "cooldown"
(see :attr:`grid2op.Parameters.Parameters.NB_TIMESTEP_COOLDOWN_LINE` for more information). 0 means the
an action will be able to act on this same powerline, a number > 0 (eg 1) means that an action at this time step
cannot act on this powerline (in the example the agent have to wait 1 time step)
time_before_cooldown_sub: :class:`numpy.ndarray`, dtype:int
Same as :attr:`BaseObservation.time_before_cooldown_line` but for substations. For each substation, it gives the
number of timesteps to wait before acting on this substation (see
see :attr:`grid2op.Parameters.Parameters.NB_TIMESTEP_COOLDOWN_SUB` for more information).
time_next_maintenance: :class:`numpy.ndarray`, dtype:int
For each powerline, it gives the time of the next planned maintenance. For example if there is:
- `1` at position `i` it means that the powerline `i` will be disconnected for maintenance operation at
the next time step.
- `0` at position `i` means that powerline `i` is disconnected from the powergrid for maintenance operation
at the current time step.
- `-1` at position `i` means that powerline `i` will not be disconnected for maintenance reason for this
episode.
- `k` > 1 at position `i` it means that the powerline `i` will be disconnected for maintenance operation at
in `k` time steps
When a powerline is "in maintenance", it cannot be reconnected by the `Agent` before the end of this
maintenance.
duration_next_maintenance: :class:`numpy.ndarray`, dtype:int
For each powerline, it gives the number of time step that the maintenance will last (if any). This means that,
if at position `i` of this vector:
- there is a `0`: the powerline is not disconnected from the grid for maintenance
- there is a `1`, `2`, ... the powerline will be disconnected for at least `1`, `2`, ... timestep (**NB**
in all case, the powerline will stay disconnected until a :class:`grid2op.BaseAgent.BaseAgent` performs the
proper :class:`grid2op.BaseAction.BaseAction` to reconnect it).
When a powerline is "in maintenance", it cannot be reconnected by the `Agent` before the end of this
maintenance.
target_dispatch: :class:`numpy.ndarray`, dtype:float
For **each** generators, it gives the target redispatching, asked by the agent. This is the sum of all
redispatching asked by the agent for during all the episode. It for each generator it is a number between:
- pmax and pmax. Note that there is information about all generators there, even the one that are not
dispatchable.
actual_dispatch: :class:`numpy.ndarray`, dtype:float
For **each** generators, it gives the redispatching currently implemented by the environment.
Indeed, the environment tries to implement at best the :attr:`BaseObservation.target_dispatch`, but sometimes,
due to physical limitation (pmin, pmax, ramp min and ramp max) it cannot. In this case, only the best possible
redispatching is implemented at the current time step, and this is what this vector stores. Note that there is
information about all generators there, even the one that are not
dispatchable.
storage_charge: :class:`numpy.ndarray`, dtype:float
The actual 'state of charge' of each storage unit, expressed in MWh.
storage_power_target: :class:`numpy.ndarray`, dtype:float
For each storage units, give the setpoint of production / consumption as given by the agent
storage_power: :class:`numpy.ndarray`, dtype:float
Give the actual storage production / loads at the given state.
storage_theta: :class:`numpy.ndarray`, dtype:float
The voltage angle (in degree) of the bus to which each storage units
is connected. Only availble if the backend supports the retrieval of
voltage angles (see :attr:`BaseObservation.support_theta`).
gen_p_before_curtail: :class:`numpy.ndarray`, dtype:float
Give the production of renewable generator there would have been
if no curtailment were applied (**NB** it returns 0.0 for non renewable
generators that cannot be curtailed)
curtailment_limit: :class:`numpy.ndarray`, dtype:float
Limit (in ratio of gen_pmax) imposed on each renewable generator as set by the agent.
It is always 1. if no curtailment actions is acting on the generator.
This is the "curtailment" given in the action by the agent.
curtailment_limit_effective: :class:`numpy.ndarray`, dtype:float
Limit (in ratio of gen_pmax) imposed on each renewable generator effectively imposed by the environment.
It matches :attr:`BaseObservation.curtailment_limit` if `param.LIMIT_INFEASIBLE_CURTAILMENT_STORAGE_ACTION`
is ``False`` (default) otherwise the environment is able to limit the curtailment actions if too much
power would be needed to compensate the "loss" of generation due to renewables.
It is always 1. if no curtailment actions is acting on the generator.
curtailment_mw: :class:`numpy.ndarray`, dtype:float
Gives the amount of power curtailed for each generator (it is 0. for all
non renewable generators)
This is NOT the "curtailment" given in the action by the agent.
curtailment: :class:`numpy.ndarray`, dtype:float
Give the power curtailed for each generator. It is expressed in
ratio of gen_pmax (so between 0. - meaning no curtailment in effect for this
generator - to 1.0 - meaning this generator should have produced pmax, but
a curtailment action limits it to 0.)
This is NOT the "curtailment" given in the action by the agent.
current_step: ``int``
Current number of step performed up until this observation (NB this is not given in the observation if
it is transformed into a vector)
max_step: ``int``
Maximum number of steps possible for this episode
delta_time: ``float``
Time (in minutes) between the last step and the current step (usually constant in an episode, even in an environment)
is_alarm_illegal: ``bool``
whether the last alarm has been illegal (due to budget constraint). It can only be ``True`` if an alarm
was raised by the agent on the previous step. Otherwise it is always ``False`` (warning: /!\\\\ Only valid with "l2rpn_icaps_2021" environment /!\\\\)
time_since_last_alarm: ``int``
Number of steps since the last successful alarm has been raised. It is `-1` if no alarm has been raised yet. (warning: /!\\\\ Only valid with "l2rpn_icaps_2021" environment /!\\\\)
last_alarm: :class:`numpy.ndarray`, dtype:int
For each zones, gives how many steps since the last alarm was raised successfully for this zone (warning: /!\\\\ Only valid with "l2rpn_icaps_2021" environment /!\\\\)
attention_budget: ``int``
The current attention budget
was_alarm_used_after_game_over: ``bool``
Was the last alarm used to compute anything related
to the attention budget when there was a game over. It can only be set to ``True`` if the observation
corresponds to a game over, but not necessarily. (warning: /!\\\\ Only valid with "l2rpn_icaps_2021" environment /!\\\\)
gen_margin_up: :class:`numpy.ndarray`, dtype:float
From how much can you increase each generators production between this
step and the next.
It is always 0. for non renewable generators. For the others it is defined as
`np.minimum(type(self).gen_pmax - self.gen_p, self.gen_max_ramp_up)`
gen_margin_down: :class:`numpy.ndarray`, dtype:float
From how much can you decrease each generators production between this
step and the next.
It is always 0. for non renewable generators. For the others it is defined as
`np.minimum(self.gen_p - type(self).gen_pmin, self.gen_max_ramp_down)`
active_alert: :class:`numpy.ndarray`, dtype:bool
.. warning:: Only available if the environment supports the "alert" feature (*eg* "l2rpn_idf_2023").
.. seealso:: :ref:`grid2op-alert-module` section of the doc for more information
.. versionadded:: 1.9.1
This attribute gives the lines "under alert" at the given observation.
It is only relevant for the "real" environment and not for `obs.simulate` nor `obs.get_forecast_env`
time_since_last_alert: :class:`numpy.ndarray`, dtype:int
.. warning:: Only available if the environment supports the "alert" feature (*eg* "l2rpn_idf_2023").
.. seealso:: :ref:`grid2op-alert-module` section of the doc for more information
.. versionadded:: 1.9.1
Give the time since an alert has been raised for each powerline. If you just raise an
alert for attackable line `i` then obs.time_since_last_alert[i] = 0 (and counter
increase by 1 each step).
If attackable line `i` has never been "under alert" then obs.time_since_last_alert[i] = -1
alert_duration: :class:`numpy.ndarray`, dtype:int
.. warning:: Only available if the environment supports the "alert" feature (*eg* "l2rpn_idf_2023").
.. seealso:: :ref:`grid2op-alert-module` section of the doc for more information
.. versionadded:: 1.9.1
Give the time since an alert has started for all attackable line. If you just raise an
alert for attackable line `i` then obs.time_since_last_alert[i] = 1 and this counter
increase by 1 each step as long as the agent continues to "raise an alert on attackable line i"
When the attackable line `i` is not under an alert then obs.time_since_last_alert[i] = 0
total_number_of_alert: :class:`numpy.ndarray`, dtype:int
.. warning:: Only available if the environment supports the "alert" feature (*eg* "l2rpn_idf_2023").
.. seealso:: :ref:`grid2op-alert-module` section of the doc for more information
.. versionadded:: 1.9.1
This attribute stores, since the beginning of the current episode, the total number
of alerts (here 1 alert = one alert for 1 powerline for 1 step) sent by the agent.
time_since_last_attack: :class:`numpy.ndarray`, dtype:int
.. warning:: Only available if the environment supports the "alert" feature (*eg* "l2rpn_idf_2023").
.. seealso:: :ref:`grid2op-alert-module` section of the doc for more information
.. versionadded:: 1.9.1
Similar to `time_since_last_alert` but for the attack.
For each attackable line `i` it counts the number of steps since the powerline has
been attacked:
- obs.time_since_last_attack[i] = -1 then attackable line `i` has never been attacked
- obs.time_since_last_attack[i] = 0 then attackable line `i` has been attacked "for the
first time" this step
- obs.time_since_last_attack[i] = 1 then attackable line `i` has been attacked "for the
first time" the previous step
- obs.time_since_last_attack[i] = 2 then attackable line `i` has been attacked "for the
first time" 2 steps ago
.. note::
An attack "for the first time" is NOT an attack "for the first time of the scenario".
Indeed, for this attribute, if a powerline is under attack for say 5 consecutive steps,
then the opponent stops its attack on this line and says 6 or 7 steps later it
start again to attack it then obs.time_since_last_attack[i] = 0 at the "first time" the
opponent attacks again this powerline.
was_alert_used_after_attack: :class:`numpy.ndarray`, dtype:int
.. warning:: Only available if the environment supports the "alert" feature (*eg* "l2rpn_idf_2023").
.. danger::
This attribute is only filled
if you use a compatible reward (*eg* :class:`grid2op.Reward.AlertReward`)
as the main reward (or a "combined" reward with this reward being part of it)
.. seealso:: :ref:`grid2op-alert-module` section of the doc for more information
.. versionadded:: 1.9.1
For each attackable line `i` it says:
- obs.was_alert_used_after_attack[i] = 0 => attackable line i has not been attacked
- obs.was_alert_used_after_attack[i] = -1 => attackable line i has been attacked and for the last attack
the INCORRECT alert was sent (meaning that: if the agent survives, it sends an alert
and if the agent died it fails to send an alert)
- obs.was_alert_used_after_attack[i] = +1 => attackable line i has been attacked and for the last attack
the CORRECT alert was sent (meaning that: if the agent survives, it did not send an alert
and if the agent died it properly sent an alert)
By "last attack", we mean the last attack that occured until now.
attack_under_alert: :class:`numpy.ndarray`, dtype:int
.. warning:: Only available if the environment supports the "alert" feature (*eg* "l2rpn_idf_2023").
.. seealso:: :ref:`grid2op-alert-module` section of the doc for more information
.. versionadded:: 1.9.1
For each attackable line `i` it says:
- obs.attack_under_alert[i] = 0 => attackable line i has not been attacked OR it
has been attacked before the relevant window (`env.parameters.ALERT_TIME_WINDOW`)
- obs.attack_under_alert[i] = -1 => attackable line i has been attacked and (before
the attack) no alert was sent (so your agent expects to survive at least
`env.parameters.ALERT_TIME_WINDOW` steps)
- obs.attack_under_alert[i] = +1 => attackable line i has been attacked and (before
the attack) an alert was sent (so your agent expects to "game over" within the next
`env.parameters.ALERT_TIME_WINDOW` steps)
_shunt_p: :class:`numpy.ndarray`, dtype:float
Shunt active value (only available if shunts are available) (in MW)
_shunt_q: :class:`numpy.ndarray`, dtype:float
Shunt reactive value (only available if shunts are available) (in MVAr)
_shunt_v: :class:`numpy.ndarray`, dtype:float
Shunt voltage (only available if shunts are available) (in kV)
_shunt_bus: :class:`numpy.ndarray`, dtype:float
Bus (-1 disconnected, 1 for bus 1, 2 for bus 2) at which each shunt is connected
(only available if shunts are available)
"""
_attr_eq = [
"line_status",
"topo_vect",
"timestep_overflow",
"gen_p",
"gen_q",
"gen_v",
"load_p",
"load_q",
"load_v",
"p_or",
"q_or",
"v_or",
"a_or",
"p_ex",
"q_ex",
"v_ex",
"a_ex",
"time_before_cooldown_line",
"time_before_cooldown_sub",
"time_next_maintenance",
"duration_next_maintenance",
"target_dispatch",
"actual_dispatch",
"_shunt_p",
"_shunt_q",
"_shunt_v",
"_shunt_bus",
# storage
"storage_charge",
"storage_power_target",
"storage_power",
# curtailment
"gen_p_before_curtail",
"curtailment",
"curtailment_limit",
"curtailment_limit_effective",
# attention budget
"is_alarm_illegal",
"time_since_last_alarm",
"last_alarm",
"attention_budget",
"was_alarm_used_after_game_over",
# line alert
"active_alert",
"attack_under_alert",
"time_since_last_alert",
"alert_duration",
"total_number_of_alert",
"time_since_last_attack",
"was_alert_used_after_attack",
# gen up / down
"gen_margin_up",
"gen_margin_down",
]
attr_list_vect = None
# value to assess if two observations are equal
_tol_equal = 1e-3
def __init__(self,
obs_env=None,
action_helper=None,
random_prng=None,
kwargs_env=None):
GridObjects.__init__(self)
self._is_done = True
self.random_prng = random_prng
self.action_helper = action_helper
# handles the forecasts here
self._forecasted_grid_act = {}
self._forecasted_inj = []
self._env_internal_params = {}
self._obs_env = obs_env
self._ptr_kwargs_env = kwargs_env
# calendar data
self.year = dt_int(1970)
self.month = dt_int(1)
self.day = dt_int(1)
self.hour_of_day = dt_int(0)
self.minute_of_hour = dt_int(0)
self.day_of_week = dt_int(0)
cls = type(self)
self.timestep_overflow = np.empty(shape=(cls.n_line,), dtype=dt_int)
# 0. (line is disconnected) / 1. (line is connected)
self.line_status = np.empty(shape=cls.n_line, dtype=dt_bool)
# topological vector
self.topo_vect = np.empty(shape=cls.dim_topo, dtype=dt_int)
# generators information
self.gen_p = np.empty(shape=cls.n_gen, dtype=dt_float)
self.gen_q = np.empty(shape=cls.n_gen, dtype=dt_float)
self.gen_v = np.empty(shape=cls.n_gen, dtype=dt_float)
self.gen_margin_up = np.empty(shape=cls.n_gen, dtype=dt_float)
self.gen_margin_down = np.empty(shape=cls.n_gen, dtype=dt_float)
# loads information
self.load_p = np.empty(shape=cls.n_load, dtype=dt_float)
self.load_q = np.empty(shape=cls.n_load, dtype=dt_float)
self.load_v = np.empty(shape=cls.n_load, dtype=dt_float)
# lines origin information
self.p_or = np.empty(shape=cls.n_line, dtype=dt_float)
self.q_or = np.empty(shape=cls.n_line, dtype=dt_float)
self.v_or = np.empty(shape=cls.n_line, dtype=dt_float)
self.a_or = np.empty(shape=cls.n_line, dtype=dt_float)
# lines extremity information
self.p_ex = np.empty(shape=cls.n_line, dtype=dt_float)
self.q_ex = np.empty(shape=cls.n_line, dtype=dt_float)
self.v_ex = np.empty(shape=cls.n_line, dtype=dt_float)
self.a_ex = np.empty(shape=cls.n_line, dtype=dt_float)
# lines relative flows
self.rho = np.empty(shape=cls.n_line, dtype=dt_float)
# cool down and reconnection time after hard overflow, soft overflow or cascading failure
self.time_before_cooldown_line = np.empty(shape=cls.n_line, dtype=dt_int)
self.time_before_cooldown_sub = np.empty(shape=cls.n_sub, dtype=dt_int)
self.time_next_maintenance = 1 * self.time_before_cooldown_line
self.duration_next_maintenance = 1 * self.time_before_cooldown_line
# redispatching
self.target_dispatch = np.empty(shape=cls.n_gen, dtype=dt_float)
self.actual_dispatch = np.empty(shape=cls.n_gen, dtype=dt_float)
# storage unit
self.storage_charge = np.empty(shape=cls.n_storage, dtype=dt_float) # in MWh
self.storage_power_target = np.empty(
shape=cls.n_storage, dtype=dt_float
) # in MW
self.storage_power = np.empty(shape=cls.n_storage, dtype=dt_float) # in MW
# attention budget
self.is_alarm_illegal = np.ones(shape=1, dtype=dt_bool)
self.time_since_last_alarm = np.empty(shape=1, dtype=dt_int)
self.last_alarm = np.empty(shape=cls.dim_alarms, dtype=dt_int)
self.attention_budget = np.empty(shape=1, dtype=dt_float)
self.was_alarm_used_after_game_over = np.zeros(shape=1, dtype=dt_bool)
# alert
dim_alert = cls.dim_alerts
self.active_alert = np.empty(shape=dim_alert, dtype=dt_bool)
self.attack_under_alert = np.empty(shape=dim_alert, dtype=dt_int)
self.time_since_last_alert = np.empty(shape=dim_alert, dtype=dt_int)
self.alert_duration = np.empty(shape=dim_alert, dtype=dt_int)
self.total_number_of_alert = np.empty(shape=1 if dim_alert else 0, dtype=dt_int)
self.time_since_last_attack = np.empty(shape=dim_alert, dtype=dt_int)
self.was_alert_used_after_attack = np.empty(shape=dim_alert, dtype=dt_int)
# to save some computation time
self._connectivity_matrix_ = None
self._bus_connectivity_matrix_ = None
self._dictionnarized = None
self._vectorized = None
# for shunt (these are not stored!)
if cls.shunts_data_available:
self._shunt_p = np.empty(shape=cls.n_shunt, dtype=dt_float)
self._shunt_q = np.empty(shape=cls.n_shunt, dtype=dt_float)
self._shunt_v = np.empty(shape=cls.n_shunt, dtype=dt_float)
self._shunt_bus = np.empty(shape=cls.n_shunt, dtype=dt_int)
self._thermal_limit = np.empty(shape=cls.n_line, dtype=dt_float)
self.gen_p_before_curtail = np.empty(shape=cls.n_gen, dtype=dt_float)
self.curtailment = np.empty(shape=cls.n_gen, dtype=dt_float)
self.curtailment_limit = np.empty(shape=cls.n_gen, dtype=dt_float)
self.curtailment_limit_effective = np.empty(shape=cls.n_gen, dtype=dt_float)
# the "theta" (voltage angle, in degree)
self.support_theta = False
self.theta_or = np.empty(shape=cls.n_line, dtype=dt_float)
self.theta_ex = np.empty(shape=cls.n_line, dtype=dt_float)
self.load_theta = np.empty(shape=cls.n_load, dtype=dt_float)
self.gen_theta = np.empty(shape=cls.n_gen, dtype=dt_float)
self.storage_theta = np.empty(shape=cls.n_storage, dtype=dt_float)
# counter
self.current_step = dt_int(0)
self.max_step = dt_int(np.iinfo(dt_int).max)
self.delta_time = dt_float(5.0)
def _aux_copy(self, other : Self) -> None:
attr_simple = [
"max_step",
"current_step",
"support_theta",
"day_of_week",
"minute_of_hour",
"hour_of_day",
"day",
"month",
"year",
"delta_time",
"_is_done",
]
attr_vect = [
"storage_theta",
"gen_theta",
"load_theta",
"theta_ex",
"theta_or",
"curtailment_limit",
"curtailment",
"gen_p_before_curtail",
"_thermal_limit",
"is_alarm_illegal",
"time_since_last_alarm",
"last_alarm",
"attention_budget",
"was_alarm_used_after_game_over",
# alert (new in 1.9.1)
"active_alert",
"attack_under_alert",
"time_since_last_alert",
"alert_duration",
"total_number_of_alert",
"time_since_last_attack",
"was_alert_used_after_attack",
# other
"storage_power",
"storage_power_target",
"storage_charge",
"actual_dispatch",
"target_dispatch",
"duration_next_maintenance",
"time_next_maintenance",
"time_before_cooldown_sub",
"time_before_cooldown_line",
"rho",
"a_ex",
"v_ex",
"q_ex",
"p_ex",
"a_or",
"v_or",
"q_or",
"p_or",
"load_p",
"load_q",
"load_v",
"gen_p",
"gen_q",
"gen_v",
"topo_vect",
"line_status",
"timestep_overflow",
"gen_margin_up",
"gen_margin_down",
"curtailment_limit_effective",
]
if type(self).shunts_data_available:
attr_vect += ["_shunt_bus", "_shunt_v", "_shunt_q", "_shunt_p"]
for attr_nm in attr_simple:
setattr(other, attr_nm, copy.deepcopy(getattr(self, attr_nm)))
for attr_nm in attr_vect:
getattr(other, attr_nm)[:] = getattr(self, attr_nm)
def __copy__(self) -> Self:
res = type(self)(obs_env=self._obs_env,
action_helper=self.action_helper,
kwargs_env=self._ptr_kwargs_env)
# copy regular attributes
self._aux_copy(other=res)
# just copy
res._connectivity_matrix_ = copy.copy(self._connectivity_matrix_)
res._bus_connectivity_matrix_ = copy.copy(self._bus_connectivity_matrix_)
res._dictionnarized = copy.copy(self._dictionnarized)
res._vectorized = copy.copy(self._vectorized)
# handles the forecasts here
res._forecasted_grid_act = copy.copy(self._forecasted_grid_act)
res._forecasted_inj = copy.copy(self._forecasted_inj)
res._env_internal_params = copy.copy(self._env_internal_params )
return res
def __deepcopy__(self, memodict={}) -> Self:
res = type(self)(obs_env=self._obs_env,
action_helper=self.action_helper,
kwargs_env=self._ptr_kwargs_env)
# copy regular attributes
self._aux_copy(other=res)
# just deepcopy
res._connectivity_matrix_ = copy.deepcopy(self._connectivity_matrix_, memodict)
res._bus_connectivity_matrix_ = copy.deepcopy(
self._bus_connectivity_matrix_, memodict
)
res._dictionnarized = copy.deepcopy(self._dictionnarized, memodict)
res._vectorized = copy.deepcopy(self._vectorized, memodict)
# handles the forecasts here
res._forecasted_grid_act = copy.deepcopy(self._forecasted_grid_act, memodict)
res._forecasted_inj = copy.deepcopy(self._forecasted_inj, memodict)
res._env_internal_params = copy.deepcopy(self._env_internal_params, memodict)
return res
def state_of(
self,
_sentinel=None,
load_id=None,
gen_id=None,
line_id=None,
storage_id=None,
substation_id=None,
) -> Dict[Literal["p", "q", "v", "theta", "bus", "sub_id", "actual_dispatch", "target_dispatch",
"maintenance", "cooldown_time", "storage_power", "storage_charge",
"storage_power_target", "storage_theta",
"topo_vect", "nb_bus", "origin", "extremity"],
Union[int, float, Dict[Literal["p", "q", "v", "a", "sub_id", "bus", "theta"], Union[int, float]]]
]:
"""
Return the state of this action on a give unique load, generator unit, powerline of substation.
Only one of load, gen, line or substation should be filled.
The querry of these objects can only be done by id here (ie by giving the integer of the object in the backed).
The :class:`ActionSpace` has some utilities to access them by name too.
Parameters
----------
_sentinel: ``None``
Used to prevent positional parameters. Internal, do not use.
load_id: ``int``
ID of the load we want to inspect
gen_id: ``int``
ID of the generator we want to inspect
line_id: ``int``
ID of the powerline we want to inspect
line_id: ``int``
ID of the powerline we want to inspect
storage_id: ``int``
ID of the storage unit we want to inspect
substation_id: ``int``
ID of the substation unit we want to inspect
Returns
-------
res: :class:`dict`
A dictionary with keys and value depending on which object needs to be inspected:
- if a load is inspected, then the keys are:
- "p" the active value consumed by the load
- "q" the reactive value consumed by the load
- "v" the voltage magnitude of the bus to which the load is connected
- "theta" (optional) the voltage angle (in degree) of the bus to which the load is connected
- "bus" on which bus the load is connected in the substation
- "sub_id" the id of the substation to which the load is connected
- if a generator is inspected, then the keys are:
- "p" the active value produced by the generator
- "q" the reactive value consumed by the generator
- "v" the voltage magnitude of the bus to which the generator is connected
- "theta" (optional) the voltage angle (in degree) of the bus to which the gen. is connected
- "bus" on which bus the generator is connected in the substation
- "sub_id" the id of the substation to which the generator is connected
- "actual_dispatch" the actual dispatch implemented for this generator
- "target_dispatch" the target dispatch (cumulation of all previously asked dispatch by the agent)
for this generator
- if a powerline is inspected then the keys are "origin" and "extremity" each being dictionary with keys:
- "p" the active flow on line side (extremity or origin)
- "q" the reactive flow on line side (extremity or origin)
- "v" the voltage magnitude of the bus to which the line side (extremity or origin) is connected
- "theta" (optional) the voltage angle (in degree) of the bus to which line side (extremity or origin)
is connected
- "bus" on which bus the line side (extremity or origin) is connected in the substation
- "sub_id" the id of the substation to which the line side is connected
- "a" the current flow on the line side (extremity or origin)
In the case of a powerline, additional information are:
- "maintenance": information about the maintenance operation (time of the next maintenance and duration
of this next maintenance.
- "cooldown_time": for how many timestep i am not supposed to act on the powerline due to cooldown
(see :attr:`grid2op.Parameters.Parameters.NB_TIMESTEP_COOLDOWN_LINE` for more information)
- if a storage unit is inspected, information are:
- "storage_power": the power the unit actually produced / absorbed
- "storage_charge": the state of the charge of the storage unit
- "storage_power_target": the power production / absorbtion targer
- "storage_theta": (optional) the voltage angle of the bus at which the storage unit is connected
- "bus": the bus (1 or 2) to which the storage unit is connected
- "sub_id" : the id of the substation to which the sotrage unit is connected
- if a substation is inspected, it returns the topology to this substation in a dictionary with keys:
- "topo_vect": the representation of which object is connected where
- "nb_bus": number of active buses in this substations
- "cooldown_time": for how many timestep i am not supposed to act on the substation due to cooldown
(see :attr:`grid2op.Parameters.Parameters.NB_TIMESTEP_COOLDOWN_SUB` for more information)
Notes
-----
This function can only be used to retrieve the state of the element of the grid, and not the alarm sent
or not, to the operator.
Raises
------
Grid2OpException
If _sentinel is modified, or if None of the arguments are set or alternatively if 2 or more of the
parameters are being set.
"""
if _sentinel is not None:
raise Grid2OpException(
"action.effect_on should only be called with named argument."
)
if (
load_id is None
and gen_id is None
and line_id is None
and substation_id is None
and storage_id is None
):
raise Grid2OpException(
"You ask the state of an object in a observation without specifying the object id. "
'Please provide "load_id", "gen_id", "line_id", "storage_id" or '
'"substation_id"'
)
cls = type(self)
if load_id is not None:
if (
gen_id is not None
or line_id is not None
or substation_id is not None
or storage_id is not None
):
raise Grid2OpException(ERROR_ONLY_SINGLE_EL)
if load_id >= len(self.load_p):
raise Grid2OpException(
'There are no load of id "load_id={}" in this grid.'.format(load_id)
)
if load_id < 0:
raise Grid2OpException("`load_id` should be a positive integer")
res = {
"p": self.load_p[load_id],
"q": self.load_q[load_id],
"v": self.load_v[load_id],
"bus": self.topo_vect[self.load_pos_topo_vect[load_id]],
"sub_id": cls.load_to_subid[load_id],
}
if self.support_theta:
res["theta"] = self.load_theta[load_id]
elif gen_id is not None:
if (
line_id is not None
or substation_id is not None
or storage_id is not None
):
raise Grid2OpException(ERROR_ONLY_SINGLE_EL)
if gen_id >= len(self.gen_p):
raise Grid2OpException(
'There are no generator of id "gen_id={}" in this grid.'.format(
gen_id
)
)
if gen_id < 0:
raise Grid2OpException("`gen_id` should be a positive integer")
res = {
"p": self.gen_p[gen_id],
"q": self.gen_q[gen_id],
"v": self.gen_v[gen_id],
"bus": self.topo_vect[self.gen_pos_topo_vect[gen_id]],
"sub_id": cls.gen_to_subid[gen_id],
"target_dispatch": self.target_dispatch[gen_id],
"actual_dispatch": self.target_dispatch[gen_id],
"curtailment": self.curtailment[gen_id],
"curtailment_limit": self.curtailment_limit[gen_id],
"curtailment_limit_effective": self.curtailment_limit_effective[gen_id],
"p_before_curtail": self.gen_p_before_curtail[gen_id],
"margin_up": self.gen_margin_up[gen_id],
"margin_down": self.gen_margin_down[gen_id],
}
if self.support_theta:
res["theta"] = self.gen_theta[gen_id]
elif line_id is not None:
if substation_id is not None or storage_id is not None:
raise Grid2OpException(ERROR_ONLY_SINGLE_EL)
if line_id >= len(self.p_or):
raise Grid2OpException(
'There are no powerline of id "line_id={}" in this grid.'.format(
line_id
)
)
if line_id < 0:
raise Grid2OpException("`line_id` should be a positive integer")
res = {}
# origin information
res["origin"] = {
"p": self.p_or[line_id],
"q": self.q_or[line_id],
"v": self.v_or[line_id],
"a": self.a_or[line_id],
"bus": self.topo_vect[cls.line_or_pos_topo_vect[line_id]],
"sub_id": cls.line_or_to_subid[line_id],
}
if self.support_theta:
res["origin"]["theta"] = self.theta_or[line_id]
# extremity information
res["extremity"] = {
"p": self.p_ex[line_id],
"q": self.q_ex[line_id],
"v": self.v_ex[line_id],
"a": self.a_ex[line_id],
"bus": self.topo_vect[cls.line_ex_pos_topo_vect[line_id]],
"sub_id": cls.line_ex_to_subid[line_id],
}
if self.support_theta:
res["origin"]["theta"] = self.theta_ex[line_id]
# maintenance information
res["maintenance"] = {
"next": self.time_next_maintenance[line_id],
"duration_next": self.duration_next_maintenance[line_id],
}
# cooldown
res["cooldown_time"] = self.time_before_cooldown_line[line_id]
elif storage_id is not None:
if substation_id is not None:
raise Grid2OpException(ERROR_ONLY_SINGLE_EL)
if storage_id >= cls.n_storage:
raise Grid2OpException(
'There are no storage unit with id "storage_id={}" in this grid.'.format(
storage_id
)
)
if storage_id < 0:
raise Grid2OpException("`storage_id` should be a positive integer")
res = {}
res["p"] = self.storage_power[storage_id]
res["storage_power"] = self.storage_power[storage_id]
res["storage_charge"] = self.storage_charge[storage_id]
res["storage_power_target"] = self.storage_power_target[storage_id]
res["bus"] = self.topo_vect[cls.storage_pos_topo_vect[storage_id]]
res["sub_id"] = cls.storage_to_subid[storage_id]