Source code for grid2op.Chronics.handlers.jsonMaintenanceHandler

# Copyright (c) 2019-2023, RTE (
# 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
# 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 json
import os
from grid2op.Chronics.GSFFWFWM import GridStateFromFileWithForecastsWithMaintenance
from grid2op.Chronics.gridValue import GridValue

from grid2op.Chronics.handlers.baseHandler import BaseHandler

[docs]class JSONMaintenanceHandler(BaseHandler): """This type of handlers will generate maintenance based on some json files. Maintenance generated with this class will be stochastic: some different maintenance time / duration will be generated for each new episode (of course you can seed your environment for a purely deterministic process) The json file it will read should be called `json_file_name` (by default `"maintenance_meta.json"`) It should contain the data: - "line_to_maintenance": the list of the name of the powerline that can be "in maintenance" for this episode - "maintenance_starting_hour" : the starting hour for all maintenance - "maintenance_ending_hour" : the hour at which each maintenance ends - "daily_proba_per_month_maintenance" : it's a list having 12 elements (one for each month of the year) that gives, for each month the probability for any given line to be in maintenance. For example if `daily_proba_per_month_maintenance[6] = 0.1` it means that for the 6th month of the year (june) there is a 10% for each powerline to be in maintenance - "max_daily_number_per_month_maintenance": maximum number of powerlines allowed in maintenance at the same time. """ def __init__(self, array_name="maintenance", json_file_name="maintenance_meta.json", max_iter=-1, _duration_episode_default=24*12, # if max_iter is not set, then maintenance are computed for a whole day ): super().__init__(array_name, max_iter) self.json_file_name = json_file_name self.dict_meta_data = None self.maintenance = None self.maintenance_time = None self.maintenance_duration = None self.n_line = None # used in one of the GridStateFromFileWithForecastsWithMaintenance functions self._duration_episode_default = _duration_episode_default self.current_index = 0 def get_maintenance_time_1d(self, maintenance): return GridValue.get_maintenance_time_1d(maintenance) def get_maintenance_duration_1d(self, maintenance): return GridValue.get_maintenance_duration_1d(maintenance) def _create_maintenance_arrays(self, current_datetime): # create the self.maintenance, self.maintenance_time and self.maintenance_duration self.maintenance = GridStateFromFileWithForecastsWithMaintenance._generate_matenance_static( self._order_backend_arrays, self.max_episode_duration if self.max_episode_duration is not None else self._duration_episode_default, self.dict_meta_data["line_to_maintenance"], self.time_interval, current_datetime, self.dict_meta_data["maintenance_starting_hour"], self.dict_meta_data["maintenance_ending_hour"], self.dict_meta_data["daily_proba_per_month_maintenance"], self.dict_meta_data["max_daily_number_per_month_maintenance"], self.space_prng ) GridStateFromFileWithForecastsWithMaintenance._fix_maintenance_format(self)
[docs] def initialize(self, order_backend_arrays, names_chronics_to_backend): self._order_backend_arrays = copy.deepcopy(order_backend_arrays) self.names_chronics_to_backend = copy.deepcopy(names_chronics_to_backend) self.n_line = len(self._order_backend_arrays) self.current_index = 0 # read the description file with open(os.path.join(self.path, self.json_file_name), "r", encoding="utf-8") as f: self.dict_meta_data = json.load(f) # and now sample the maintenance self._create_maintenance_arrays(self.init_datetime)
[docs] def check_validity(self, backend): # TODO pass
[docs] def load_next_maintenance(self): maint_time = 1 * self.maintenance_time[self.current_index, :] maint_duration = 1 * self.maintenance_duration[self.current_index, :] return maint_time, maint_duration
[docs] def load_next(self, dict_): self.current_index += 1 if self.current_index >= self.maintenance.shape[0]: # regenerate some maintenance if needed self.current_index = 0 self.init_datetime += self.maintenance.shape[0] * self.time_interval self._create_maintenance_arrays(self.init_datetime) return copy.deepcopy(self.maintenance[self.current_index, :])
def _clear(self): super()._clear() self.dict_meta_data = None self.maintenance = None self.maintenance_time = None self.maintenance_duration = None self.n_line = None self.current_index = 0
[docs] def done(self): # maintenance can be generated on the fly so they are never "done" return False