#Copyright (c) 2019-2021, 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 typing import Optional, List
import numpy as np
from grid2op.dtypes import dt_int
from grid2op.Opponent.baseOpponent import BaseOpponent
from grid2op.Opponent.geometricOpponent import GeometricOpponent
from grid2op.Exceptions import OpponentError
[docs]class GeometricOpponentMultiArea(BaseOpponent):
"""
This opponent is a combination of several similar opponents (of Kind Geometric Opponent at this stage) attacking on different areas.
The difference between unitary opponents is mainly the attackable lines (which belongs to different pre-identified areas
"""
def __init__(self, action_space):
BaseOpponent.__init__(self, action_space)
self.list_opponents : Optional[List[GeometricOpponent]] = None
self._new_attack_time_counters : Optional[np.ndarray] = None
self._previous_attacks = None
[docs] def init(
self,
partial_env,
lines_attacked=None,
attack_every_xxx_hour=24,
average_attack_duration_hour=4,
minimum_attack_duration_hour=2,
pmax_pmin_ratio=4,
**kwargs,
):
"""
Generic function used to initialize the derived classes. For example, if an opponent reads from a file, the
path where is the file is located should be pass with this method.
This is based on init from GeometricOpponent, only parameter lines_attacked becomes a list of list
Parameters
----------
partial_env: grid2op Environment
see the GeometricOpponent::init documentation
lines_attacked: ``list(list)``
The lists of lines attacked by each unitary opponent (this is a list of list: the size
of the outer list is the number of underlying opponent / number of areas and for each inner
list it gives the name of the lines to attack.)
attack_every_xxx_hour: ``float``
see the GeometricOpponent::init documentation
average_attack_duration_hour: ``float``
see the GeometricOpponent::init documentation
minimum_attack_duration_hour: ``int``
see the GeometricOpponent::init documentation
pmax_pmin_ratio: ``float``
see the GeometricOpponent::init documentation
"""
if lines_attacked is None:
partial_env.logger.warning("GeometricOpponentMultiArea: no area provided, the opponent will be deactivated.")
return
self.list_opponents = [GeometricOpponent(action_space=self.action_space) for _ in lines_attacked]
self._previous_attacks = [None for _ in lines_attacked]
for lines_attacked, opp in zip(lines_attacked, self.list_opponents):
opp.init(
partial_env=partial_env,
lines_attacked=lines_attacked,
attack_every_xxx_hour=attack_every_xxx_hour,
average_attack_duration_hour=average_attack_duration_hour,
minimum_attack_duration_hour=minimum_attack_duration_hour,
pmax_pmin_ratio=pmax_pmin_ratio,
**kwargs,
)
self._new_attack_time_counters = np.array([-1 for _ in lines_attacked])#ou plutôt 0 comme dans Geometric Opponent ?
[docs] def reset(self, initial_budget):
self._new_attack_time_counters = np.array([-1 for _ in self.list_opponents])
for opp in self.list_opponents: # maybe loop in different orders each time
opp.reset(initial_budget)
[docs] def attack(self, observation, agent_action, env_action, budget, previous_fails):
"""
This method is the equivalent of "attack" for a regular agent.
Opponent, in this framework can have more information than a regular agent (in particular it can
view time step t+1), it has access to its current budget etc.
Here we take the combination of unitary opponent attacks if they happen at the same time.
We choose the attack duration as the minimum duration of several simultaneous attacks if that happen.
Parameters
----------
observation: :class:`grid2op.Observation.Observation`
see the GeometricOpponent::attack documentation
opp_reward: ``float``
see the GeometricOpponent::attack documentation
done: ``bool``
see the GeometricOpponent::attack documentation
agent_action: :class:`grid2op.Action.Action`
see the GeometricOpponent::attack documentation
env_action: :class:`grid2op.Action.Action`
see the GeometricOpponent::attack documentation
budget: ``float``
see the GeometricOpponent::attack documentation
previous_fails: ``bool``
see the GeometricOpponent::attack documentation
Returns
-------
attack: :class:`grid2op.Action.Action`
see the GeometricOpponent::attack documentation
duration: ``int``
see the GeometricOpponent::attack documentation
"""
#go through opponents and check if attack or not. As soon as one attack, stop there
self._new_attack_time_counters -= 1
self._new_attack_time_counters[self._new_attack_time_counters < -1] = -1
attack_combined = None
for opp_id, opp in enumerate(self.list_opponents):
if self._new_attack_time_counters[opp_id] == -1:
attack_opp, attack_duration_opp = opp.attack(observation, agent_action, env_action, budget, previous_fails)
if attack_opp is not None:
self._new_attack_time_counters[opp_id] = attack_duration_opp
self._previous_attacks[opp_id] = attack_opp
if attack_combined is None:
attack_combined = attack_opp.copy()
else:
attack_combined += attack_opp
else:
self._previous_attacks[opp_id] = None
else:
opp.tell_attack_continues(observation, agent_action, env_action, budget)
if attack_combined is None:
attack_combined = self._previous_attacks[opp_id].copy()
else:
attack_combined += self._previous_attacks[opp_id]
return attack_combined, 1
[docs] def tell_attack_continues(self, observation, agent_action, env_action, budget):
raise RuntimeError("I should not get there !")
[docs] def get_state(self):
return (self._new_attack_time_counters,
self._previous_attacks,
[opp.get_state() for opp in self.list_opponents])
[docs] def set_state(self, my_state):
self._new_attack_time_counters = np.array(my_state[0])
self._previous_attacks = [el.copy() if el is not None else None for el in my_state[1]]
for el, opp in zip(my_state[2], self.list_opponents):
opp.set_state(el)
def _custom_deepcopy_for_copy(self, new_obj, dict_=None):
new_obj._new_attack_time_counters = 1 * self._new_attack_time_counters
new_obj._previous_attacks = [el.copy() if el is not None else None
for el in self._previous_attacks]
new_obj.list_opponents = []
for opp in self.list_opponents:
new_opp = type(opp).__new__(type(opp))
opp._custom_deepcopy_for_copy(new_opp, dict_)
new_obj.list_opponents.append(new_opp)
super()._custom_deepcopy_for_copy(new_obj)
return new_obj
[docs] def seed(self, seed):
"""
INTERNAL
.. warning:: /!\\\\ Internal, do not use unless you know what you are doing /!\\\\
We do not recommend to use this function outside of the two examples given in the description of this class.
Set the seeds of the source of pseudo random number used for these several unitary opponents.
Parameters
----------
seed: ``int``
The root seed to be set for the random number generator.
Returns
-------
seeds: ``list``
The associated list of seeds used.
"""
seeds = []
super().seed(seed)
max_seed = np.iinfo(dt_int).max # 2**32 - 1
for opp in self.list_opponents:
this_seed = self.space_prng.randint(max_seed)
seeds.append(opp.seed(this_seed))
return (seed, seeds)