random_generator_battery.py

# ------------------------------------------------------------------------
# Energy management environment for reinforcement learning agents developed by
# Hou Shengren, TU Delft, h.shengren@tudelft.nl
# ------------------------------------------------------------------------
from pathlib import Path

import numpy as np
import pandas as pd

try:
    import gym
    from gym import spaces
except ImportError:  # pragma: no cover - optional dependency
    class _FallbackEnv:
        pass

    class _FallbackBox:
        def __init__(self, low, high, shape, dtype=np.float32):
            self.shape = shape
            self.dtype = dtype
            self.low = np.asarray(low, dtype=dtype)
            self.high = np.asarray(high, dtype=dtype)
            if self.low.shape == ():
                self.low = np.full(shape, self.low.item(), dtype=dtype)
            else:
                self.low = self.low.reshape(shape)
            if self.high.shape == ():
                self.high = np.full(shape, self.high.item(), dtype=dtype)
            else:
                self.high = self.high.reshape(shape)

    class _FallbackSpaces:
        Box = _FallbackBox

    class _FallbackGym:
        Env = _FallbackEnv

    gym = _FallbackGym()
    spaces = _FallbackSpaces()

from Parameters import battery_parameters, dg_parameters


class Constant:
    MONTHS_LEN = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
    MAX_STEP_HOURS = 24 * 30


class DataManager:
    def __init__(self):
        self.PV_Generation = []
        self.Prices = []
        self.Electricity_Consumption = []

    def add_pv_element(self, element):
        self.PV_Generation.append(element)

    def add_price_element(self, element):
        self.Prices.append(element)

    def add_electricity_element(self, element):
        self.Electricity_Consumption.append(element)

    def get_pv_data(self, month, day, day_time):
        return self.PV_Generation[(sum(Constant.MONTHS_LEN[: month - 1]) + day - 1) * 24 + day_time]

    def get_price_data(self, month, day, day_time):
        return self.Prices[(sum(Constant.MONTHS_LEN[: month - 1]) + day - 1) * 24 + day_time]

    def get_electricity_cons_data(self, month, day, day_time):
        return self.Electricity_Consumption[
            (sum(Constant.MONTHS_LEN[: month - 1]) + day - 1) * 24 + day_time
        ]

    def get_series_pv_data(self, month, day):
        start = (sum(Constant.MONTHS_LEN[: month - 1]) + day - 1) * 24
        return self.PV_Generation[start : start + 24]

    def get_series_price_data(self, month, day):
        start = (sum(Constant.MONTHS_LEN[: month - 1]) + day - 1) * 24
        return self.Prices[start : start + 24]

    def get_series_electricity_cons_data(self, month, day):
        start = (sum(Constant.MONTHS_LEN[: month - 1]) + day - 1) * 24
        return self.Electricity_Consumption[start : start + 24]


class DG:
    def __init__(self, parameters):
        self.name = parameters.keys()
        self.a_factor = parameters["a"]
        self.b_factor = parameters["b"]
        self.c_factor = parameters["c"]
        self.power_output_max = parameters["power_output_max"]
        self.power_output_min = parameters["power_output_min"]
        self.ramping_up = parameters["ramping_up"]
        self.ramping_down = parameters["ramping_down"]
        self.last_step_output = None
        self.current_output = 0.0

    def step(self, action_gen):
        output_change = action_gen * self.ramping_up
        output = self.current_output + output_change
        if output > 0:
            output = max(self.power_output_min, min(self.power_output_max, output))
        else:
            output = 0
        self.current_output = output

    def _get_cost(self, output):
        if output <= 0:
            return 0
        return self.a_factor * pow(output, 2) + self.b_factor * output + self.c_factor

    def reset(self):
        self.current_output = 0


class Battery:
    def __init__(self, parameters):
        self.capacity = parameters["capacity"]
        self.max_soc = parameters["max_soc"]
        self.initial_capacity = parameters["initial_capacity"]
        self.min_soc = parameters["min_soc"]
        self.degradation = parameters["degradation"]
        self.max_charge = parameters["max_charge"]
        self.max_discharge = parameters["max_discharge"]
        self.efficiency = parameters["efficiency"]
        self.current_capacity = self.initial_capacity
        self.energy_change = 0.0

    def step(self, action_battery):
        energy = action_battery * self.max_charge
        updated_capacity = max(
            self.min_soc,
            min(self.max_soc, (self.current_capacity * self.capacity + energy) / self.capacity),
        )
        self.energy_change = (updated_capacity - self.current_capacity) * self.capacity
        self.current_capacity = updated_capacity

    def _get_cost(self, energy):
        return energy**2 * self.degradation

    def SOC(self):
        return self.current_capacity

    def reset(self):
        self.current_capacity = np.random.uniform(self.min_soc, self.max_soc)
        self.energy_change = 0.0


class Grid:
    def __init__(self):
        self.on = True
        self.exchange_ability = 30 if self.on else 0

    def _get_cost(self, current_price, energy_exchange):
        return current_price * energy_exchange

    def retrive_past_price(self):
        result = []
        if self.day < 1:
            past_price = self.past_price
        else:
            past_price = self.price[24 * (self.day - 1) : 24 * self.day]
        for item in past_price[(self.time - 24) :]:
            result.append(item)
        for item in self.price[24 * self.day : (24 * self.day + self.time)]:
            result.append(item)
        return result


class ESSEnv(gym.Env):
    DATA_DIR = Path(__file__).resolve().parent / "data"
    state_names = ("time_step", "price", "soc", "net_load", "dg1_output", "dg2_output", "dg3_output")
    action_names = ("battery", "dg1", "dg2", "dg3")

    def __init__(self, **kwargs):
        super().__init__()
        self.data_manager = DataManager()
        self._load_year_data()
        self.episode_length = kwargs.get("episode_length", 24)
        self.month = None
        self.day = None
        self.TRAIN = True
        self.current_time = None
        self.battery_parameters = kwargs.get("battery_parameters", battery_parameters)
        self.dg_parameters = kwargs.get("dg_parameters", dg_parameters)
        self.penalty_coefficient = 20
        self.sell_coefficient = 0.5

        self.grid = Grid()
        self.battery = Battery(self.battery_parameters)
        self.dg1 = DG(self.dg_parameters["gen_1"])
        self.dg2 = DG(self.dg_parameters["gen_2"])
        self.dg3 = DG(self.dg_parameters["gen_3"])

        self.action_space = spaces.Box(low=-1, high=1, shape=(4,), dtype=np.float32)
        self.state_space = spaces.Box(
            low=np.asarray([0.0, 0.0, 0.0, -1.0, 0.0, 0.0, 0.0], dtype=np.float32),
            high=np.ones(7, dtype=np.float32),
            shape=(7,),
            dtype=np.float32,
        )

        self.Length_max = 24
        self.Price_max = max(self.data_manager.Prices)
        self.Netload_max = max(self.data_manager.Electricity_Consumption)
        self.SOC_max = self.battery.max_soc
        self.DG1_max = self.dg1.power_output_max
        self.DG2_max = self.dg2.power_output_max
        self.DG3_max = self.dg3.power_output_max

        self.current_output = np.zeros(self.action_space.shape[0], dtype=np.float32)
        self.operation_cost = 0.0
        self.unbalance = 0.0
        self.real_unbalance = 0.0
        self.excess = 0.0
        self.shedding = 0.0

    def reset(self):
        self.month = np.random.randint(1, 13)
        if self.TRAIN:
            self.day = np.random.randint(1, 21)
        else:
            self.day = np.random.randint(21, Constant.MONTHS_LEN[self.month - 1])
        self.current_time = 0
        self.battery.reset()
        self.dg1.reset()
        self.dg2.reset()
        self.dg3.reset()
        return self._build_state()

    def _build_state(self):
        soc = self.battery.SOC() / self.SOC_max
        dg1_output = self.dg1.current_output / self.DG1_max
        dg2_output = self.dg2.current_output / self.DG2_max
        dg3_output = self.dg3.current_output / self.DG3_max
        time_step = self.current_time / (self.Length_max - 1)
        electricity_demand = self.data_manager.get_electricity_cons_data(self.month, self.day, self.current_time)
        pv_generation = self.data_manager.get_pv_data(self.month, self.day, self.current_time)
        price = self.data_manager.get_price_data(self.month, self.day, self.current_time) / self.Price_max
        net_load = (electricity_demand - pv_generation) / self.Netload_max
        return np.asarray(
            [time_step, price, soc, net_load, dg1_output, dg2_output, dg3_output],
            dtype=np.float32,
        )

    def _denormalize_net_load(self, state):
        return float(state[self.state_names.index("net_load")] * self.Netload_max)

    def _denormalize_price(self, state):
        return float(state[self.state_names.index("price")] * self.Price_max)

    def _compute_actual_production(self):
        return sum(self.current_output)

    def step(self, action):
        action = np.asarray(action, dtype=np.float32).reshape(self.action_space.shape[0])
        current_obs = self._build_state()

        self.battery.step(action[0])
        self.dg1.step(action[1])
        self.dg2.step(action[2])
        self.dg3.step(action[3])

        self.current_output = np.asarray(
            [
                self.dg1.current_output,
                self.dg2.current_output,
                self.dg3.current_output,
                -self.battery.energy_change,
            ],
            dtype=np.float32,
        )
        actual_production = self._compute_actual_production()
        netload = self._denormalize_net_load(current_obs)
        price = self._denormalize_price(current_obs)

        unbalance = actual_production - netload

        excess_penalty = 0.0
        deficient_penalty = 0.0
        sell_benefit = 0.0
        buy_cost = 0.0
        self.excess = 0.0
        self.shedding = 0.0

        if unbalance >= 0:
            if unbalance <= self.grid.exchange_ability:
                sell_benefit = self.grid._get_cost(price, unbalance) * self.sell_coefficient
            else:
                sell_benefit = self.grid._get_cost(price, self.grid.exchange_ability) * self.sell_coefficient
                self.excess = unbalance - self.grid.exchange_ability
                excess_penalty = self.excess * self.penalty_coefficient
        else:
            if abs(unbalance) <= self.grid.exchange_ability:
                buy_cost = self.grid._get_cost(price, abs(unbalance))
            else:
                buy_cost = self.grid._get_cost(price, self.grid.exchange_ability)
                self.shedding = abs(unbalance) - self.grid.exchange_ability
                deficient_penalty = self.shedding * self.penalty_coefficient

        battery_cost = self.battery._get_cost(self.battery.energy_change)
        dg1_cost = self.dg1._get_cost(self.dg1.current_output)
        dg2_cost = self.dg2._get_cost(self.dg2.current_output)
        dg3_cost = self.dg3._get_cost(self.dg3.current_output)

        reward = -(
            battery_cost
            + dg1_cost
            + dg2_cost
            + dg3_cost
            + excess_penalty
            + deficient_penalty
            - sell_benefit
            + buy_cost
        ) / 2e3

        self.operation_cost = (
            battery_cost
            + dg1_cost
            + dg2_cost
            + dg3_cost
            + buy_cost
            - sell_benefit
            + (self.shedding + self.excess) * self.penalty_coefficient
        )

        self.unbalance = unbalance
        self.real_unbalance = self.shedding + self.excess
        final_step_outputs = [
            self.dg1.current_output,
            self.dg2.current_output,
            self.dg3.current_output,
            self.battery.current_capacity,
        ]
        self.current_time += 1
        finish = self.current_time == self.episode_length
        if finish:
            self.final_step_outputs = final_step_outputs
            self.current_time = 0
            next_obs = self.reset()
        else:
            next_obs = self._build_state()
        return current_obs, next_obs, float(reward), finish

    def render(self, current_obs, next_obs, reward, finish):
        print(
            "day={},hour={:2d}, state={}, next_state={}, reward={:.4f}, terminal={}\n".format(
                self.day, self.current_time, current_obs, next_obs, reward, finish
            )
        )

    def get_actor_mip_config(self):
        return {
            "state_names": self.state_names,
            "action_names": self.action_names,
            "scaled_parameters": {
                "battery_power": self.battery.max_charge,
                "dg1_ramping": self.dg1.ramping_up,
                "dg2_ramping": self.dg2.ramping_up,
                "dg3_ramping": self.dg3.ramping_up,
                "net_load_scale": self.Netload_max,
                "dg1_output_scale": self.DG1_max,
                "dg2_output_scale": self.DG2_max,
                "dg3_output_scale": self.DG3_max,
            },
        }

    def _load_year_data(self):
        pv_df = pd.read_csv(self.DATA_DIR / "PV.csv", sep=";")
        price_df = pd.read_csv(self.DATA_DIR / "Prices.csv", sep=";")
        electricity_df = pd.read_csv(self.DATA_DIR / "H4.csv", sep=";")
        pv_data = pv_df["P_PV_"].apply(lambda value: value.replace(",", ".")).to_numpy(dtype=float)
        price = price_df["Price"].apply(lambda value: value.replace(",", ".")).to_numpy(dtype=float)
        electricity = electricity_df["Power"].apply(lambda value: value.replace(",", ".")).to_numpy(dtype=float)

        for element in pv_data:
            self.data_manager.add_pv_element(element * 100)
        for element in price:
            element /= 10
            if element <= 0.5:
                element = 0.5
            self.data_manager.add_price_element(element)
        for index in range(0, electricity.shape[0], 60):
            element = electricity[index : index + 60]
            self.data_manager.add_electricity_element(sum(element) * 300)


if __name__ == "__main__":
    env = ESSEnv()
    env.TRAIN = False
    rewards = []
    env.reset()
    env.day = 27
    tem_action = [0.1, 0.1, 0.1, 0.1]
    for _ in range(240):
        print(
            f"current month is {env.month}, current day is {env.day}, current time is {env.current_time}"
        )
        current_obs, next_obs, reward, finish = env.step(tem_action)
        env.render(current_obs, next_obs, reward, finish)
        current_obs = next_obs
        rewards.append(reward)