Spaces:
Sleeping
Sleeping
File size: 12,184 Bytes
5ef652c e733d30 5ef652c 0c694ac 5ef652c e733d30 e8def5c 5ef652c e733d30 e8def5c e733d30 5ef652c e733d30 5ef652c e733d30 5ef652c dd725fc 5ef652c dd725fc 5ef652c dd725fc a13327d dd725fc 5ef652c dd725fc 5ef652c dd725fc 5ef652c dd725fc f1abbf3 0c694ac f1abbf3 0c694ac dd725fc 0c694ac dd725fc 0c694ac e733d30 0c694ac dd725fc 0c694ac dd725fc f1abbf3 dd725fc 0c694ac dd725fc 0c694ac e733d30 0c694ac e733d30 0c694ac e733d30 0c694ac dd725fc 0c694ac e733d30 0c694ac dd725fc e733d30 dd725fc 0c694ac e733d30 0c694ac a13327d 0c694ac dd725fc 0c694ac dd725fc 0c694ac dd725fc 0c694ac 5ef652c e733d30 5ef652c f1abbf3 5ef652c f1abbf3 e733d30 5ef652c f1abbf3 e733d30 5ef652c |
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 |
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()
|