v2/architecture.py

import numpy as np
import pandas as pd
import copy
from enum import IntEnum

# it's really just a network pricing model
from supplier import Supplier
from data_processing import read_datasets, add_production_field, interpolate_and_join, SolarParameters


STEPS_PER_HOUR = 12


# SOC is normalized so that minimal_depth_of_discharge = 0 and maximal_depth_of_discharge = 1.
# please set capacity_Ah = nominal_capacity_Ah * (max_dod - min_dod)
#
# TODO efficiency multiplier is not currently used, where best to put it?
class BatteryModel:
    def __init__(self, capacity_Ah, time_interval_h):
        self.capacity_Ah = capacity_Ah
        self.efficiency = 0.9 # [dimensionless]
        self.voltage_V = 600
        self.charge_kW = 50
        self.discharge_kW = 60
        self.time_interval_h = time_interval_h

        # the only non-constant member variable!
        # ratio of self.current_capacity_kWh and self.maximal_capacity_kWh
        self.soc = 0.0 

    @property
    def maximal_capacity_kWh(self):
        return self.capacity_Ah * self.voltage_V / 1000

    @property
    def current_capacity_kWh(self):
        return self.soc * self.maximal_capacity_kWh

    def satisfy_demand(self, demand_kW):
        assert 0 <= self.soc <= 1
        assert demand_kW >= 0
        # rate limited:
        possible_discharge_in_timestep_kWh = self.discharge_kW * self.time_interval_h
        # limited by current capacity:
        possible_discharge_in_timestep_kWh = min((possible_discharge_in_timestep_kWh, self.current_capacity_kWh))
        # limited by need:
        discharge_in_timestep_kWh = min((possible_discharge_in_timestep_kWh, demand_kW * self.time_interval_h))
        consumption_from_bess_kW = discharge_in_timestep_kWh / self.time_interval_h
        unsatisfied_demand_kW = demand_kW - consumption_from_bess_kW
        cap_of_battery_kWh = self.current_capacity_kWh - discharge_in_timestep_kWh
        soc = cap_of_battery_kWh / self.maximal_capacity_kWh
        assert 0 <= soc <= self.soc <= 1
        self.soc = soc
        return unsatisfied_demand_kW

    def charge(self, charge_kW):
        assert 0 <= self.soc <= 1
        assert charge_kW >= 0
        # rate limited:
        possible_charge_in_timestep_kWh = self.charge_kW * self.time_interval_h
        # limited by current capacity:
        possible_charge_in_timestep_kWh = min((possible_charge_in_timestep_kWh, self.maximal_capacity_kWh - self.current_capacity_kWh))
        # limited by supply:
        charge_in_timestep_kWh = min((possible_charge_in_timestep_kWh, charge_kW * self.time_interval_h))
        actual_charge_kW = charge_in_timestep_kWh / self.time_interval_h
        unused_charge_kW = charge_kW - actual_charge_kW
        cap_of_battery_kWh = self.current_capacity_kWh + charge_in_timestep_kWh
        soc = cap_of_battery_kWh / self.maximal_capacity_kWh
        assert 0 <= self.soc <= soc <= 1
        self.soc = soc
        return unused_charge_kW


class Decision(IntEnum):
    # use solar to satisfy consumption,
    # and if it is not enough, use network.
    # BESS is not discharged in this mode,
    # but might be charged if solar has surplus.
    PASSIVE = 0

    # use the battery if possible and necessary.
    # the possible part means that there's charge in it,
    # and the necessary part means that the consumption
    # is not already covered by solar.
    DISCHARGE = 1

    # use the network to charge the battery
    # this is similar to PASSIVE, but forces the
    # BESS to be charged, even if solar does not cover
    # the whole of consumption plus BESS.
    NETWORK_CHARGE = 2


# mock class as usual
# output_window_size is not yet used, always decides one timestep.
class Decider:
    def __init__(self):
        self.input_window_size = STEPS_PER_HOUR * 24 # day long window.
        self.random_seed = 0

    # prod_cons_pred is a dataframe starting at now, containing
    # fields Production and Consumption.
    # this function does not mutate its inputs.
    # battery_model is just queried for capacity and current soc.
    # the method returns a pd.Series of Decisions as integers.
    def decide(self, prod_pred, cons_pred, battery_model):
        # assert len(prod_pred) == len(cons_pred) == self.input_window_size
        self.random_seed += 1
        self.random_seed %= 3 # dummy rotates between Decisions
        self.random_seed = Decision.PASSIVE
        return self.random_seed
        # dummy decider always says DISCHARGE:
        # return pd.Series([Decision.DISCHARGE] * self.output_window_size, dtype=int)


# even mock-er class than usual.
# knows the future in advance, so it predicts it very well.
# it's also unrealistic in that it takes row index instead of date.
class DummyPredictor:
    def __init__(self, series):
        self.series = series

    def predict(self, indx, window_size):
        prediction = self.series.iloc[indx: indx + window_size]
        return prediction


# this function does not mutate its inputs.
# it makes a clone of battery_model and modifies that.
def simulator(battery_model, supplier, prod_cons, decider):
    battery_model = copy.copy(battery_model)

    demand_np = prod_cons['Consumption'].to_numpy()
    production_np = prod_cons['Production'].to_numpy()
    demand_prediction_np = prod_cons['Consumption_prediction'].to_numpy()
    production_prediction_np = prod_cons['Production_prediction'].to_numpy()
    assert len(demand_np) == len(production_np)
    step_in_minutes = prod_cons.index.freq.n
    assert step_in_minutes == 5

    print("Simulating for", len(demand_np), "time steps. Each step is", step_in_minutes, "minutes.")
    soc_series = []
    # by convention, we only call end user demand, demand,
    # and we only call end user consumption, consumption.
    # in our simple model, demand is always satisfied, hence demand=consumption.
    # BESS demand is called charge.
    consumption_from_solar_series = [] # demand satisfied by solar production
    consumption_from_network_series = [] # full demand satisfied by network, also includes consumption_from_network_to_bess.
    consumption_from_bess_series = [] # demand satisfied by BESS
    consumption_from_network_to_bess_series = [] # network used to charge BESS, also included in consumption_from_network.
    # the previous three must sum to demand_series.

    discarded_production_series = [] # solar power thrown away

    # 1 is not nominal but targeted (healthy) maximum charge.
    # we start with an empty battery, but not emptier than what's healthy for the batteries.

    # For the sake of simplicity 0 <= soc <= 1
    # soc=0 means battery is emptied till it's 20% and soc=1 means battery is charged till 80% of its capacity
    # soc = 1 - maximal_depth_of_discharge
    # and will use only maximal_depth_of_discharge percent of the real battery capacity

    time_interval = step_in_minutes / 60 # amount of time step in hours
    for i, (demand, production) in enumerate(zip(demand_np, production_np)):
        # these five are modified on the appropriate codepaths:
        consumption_from_solar = 0
        consumption_from_bess = 0
        consumption_from_network = 0
        discarded_production = 0
        network_used_to_charge = 0

        unsatisfied_demand = demand
        remaining_production = production

        # TODO what to call it, demand or consumption?
        # 1. sometimes demand is inappropriate, like consumption_from_solar vs demand_from_solar.
        # 2. sometimes consumption is inappropriate, like unsatisfied_demand vs unsatisfied_consumption.
        # 3. there should not be two of them.
        prod_prediction = production_prediction_np[i: i + decider.input_window_size]
        cons_prediction = demand_prediction_np[i: i + decider.input_window_size]
        decision = decider.decide(prod_prediction, cons_prediction, battery_model)

        production_used_to_charge = 0
        if unsatisfied_demand >= remaining_production:
            # all goes to demand, no rate limit between solar and consumer
            consumption_from_solar = remaining_production
            unsatisfied_demand -= consumption_from_solar
            remaining_production = 0
            if decision == Decision.DISCHARGE:
                # we try to cover the rest from BESS
                unsatisfied_demand = battery_model.satisfy_demand(unsatisfied_demand)
            # we cover the rest from network
            consumption_from_network = unsatisfied_demand
            unsatisfied_demand = 0
        else:
            # demand fully satisfied by production, no rate limit between solar and consumer
            consumption_from_solar = unsatisfied_demand
            remaining_production -= unsatisfied_demand
            unsatisfied_demand = 0
            if remaining_production > 0:
                # exploitable production still remains:
                discarded_production = battery_model.charge(remaining_production) # remaining_production [kW]

        if decision == Decision.NETWORK_CHARGE:
            # that is some random big number, the actual charge will be
            # determined by the combination of the BESS rate limit (battery_model.charge_kW)
            # and the BESS capacity
            fictional_network_charge_kW = 1000
            discarded_network_charge_kW = battery_model.charge(fictional_network_charge_kW)
            actual_network_charge_kW = fictional_network_charge_kW - discarded_network_charge_kW
            consumption_from_network_to_bess = actual_network_charge_kW # just a renaming.
            consumption_from_network += actual_network_charge_kW
        else:
            consumption_from_network_to_bess = 0

        soc_series.append(battery_model.soc)
        consumption_from_solar_series.append(consumption_from_solar)
        consumption_from_network_series.append(consumption_from_network)
        consumption_from_network_to_bess_series.append(consumption_from_network_to_bess)
        consumption_from_bess_series.append(consumption_from_bess)
        discarded_production_series.append(discarded_production)

    soc_series = np.array(soc_series)
    consumption_from_solar_series = np.array(consumption_from_solar_series)
    consumption_from_network_series = np.array(consumption_from_network_series)
    consumption_from_bess_series = np.array(consumption_from_bess_series)
    discarded_production_series = np.array(discarded_production_series)

    total_charge, consumption_charge_series, demand_charges = supplier.fee(
        pd.Series(consumption_from_network_series, index=prod_cons.index),
        provide_detail=True)
    print(f"All in all we have paid {total_charge} to network.")

    results = pd.DataFrame({'soc_series': soc_series, 'consumption_from_solar': consumption_from_solar_series,
                            'consumption_from_network': consumption_from_network_series,
                            'consumption_from_bess': consumption_from_bess_series,
                            'consumption_from_network_to_bess_series': consumption_from_network_to_bess_series,
                            'discarded_production': discarded_production_series,
                            'Consumption': prod_cons['Consumption'],
                            'Production': prod_cons['Production']
                            })
    results = results.set_index(prod_cons.index)
    return results


def main():

    supplier = Supplier(price=100) # Ft/kWh
    supplier.set_price_for_interval(9, 17, 150) # nine-to-five increased price.

    parameters = SolarParameters()

    met_2021_data, cons_2021_data = read_datasets()
    add_production_field(met_2021_data, parameters)
    all_data = interpolate_and_join(met_2021_data, cons_2021_data)

    all_data_with_predictions = all_data.copy()

    time_interval_min = all_data.index.freq.n
    time_interval_h = time_interval_min / 60
    battery_model = BatteryModel(capacity_Ah=600, time_interval_h=time_interval_h)

    decider = Decider()

    results = simulator(battery_model, supplier, all_data_with_predictions, decider)

    import matplotlib.pyplot as plt
    results['soc_series'].plot()
    plt.show()


if __name__ == '__main__':
    main()