[docs]classBatteryEnvFwd(gym.Env):def__init__(self,env_config)->None:"""Creates battery envrionemnt Args: env_config (dict): Customizable environment confing. n_fwd_steps(int): Number of forward forecast steps available max_bat_cap(float): Maximun battery capacity in MWh charging_rate(float): Rate of charge of the battery reward_metod(function): Method used to calculate the reward """super(BatteryEnvFwd,self).__init__()n_fwd_steps=env_config['n_fwd_steps']max_bat_cap=env_config['max_bat_cap']charging_rate=env_config['charging_rate']self.observation_space=spaces.Box(low=np.float32(-1.0*np.ones(1+1+4+n_fwd_steps)),high=np.float32(1.0*np.ones(1+1+4+n_fwd_steps)))self.max_dc_pw_MW=env_config['max_dc_pw_MW']# 7.24 # in MWself.action_space=spaces.Discrete(3)self._action_to_direction={0:'charge',1:'discharge',2:'idle'}other_states_max=np.array([self.max_dc_pw_MW,max_bat_cap])other_states_min=np.array([0.0,0])self.observation_max=other_states_maxself.observation_min=other_states_minself.delta=self.observation_max-self.observation_minself.battery=batt.Battery2(capacity=max_bat_cap,current_load=0*max_bat_cap)self.n_fwd_steps=n_fwd_stepsself.charging_rate=charging_rateself.spot_CI=Noneself.ma_CI=Noneself.eta=0.7self.dataset_end=Trueself.dcload=0self.temp_state=Noneself.var_to_dc=0self.max_bat_cap=max_bat_capself.total_energy_with_battery=0self.ci=0self.ci_n=[]self.dcload_max=env_config['dcload_max']self.dcload_min=env_config['dcload_min']
[docs]defreset(self,*,seed=None,options=None):""" Reset `BatteryEnvFwd` to initial state. Args: seed (int, optional): Random seed. options (dict, optional): Environment options. Returns: temp_state (List[float]): Current state of the environmment info (dict): A dictionary that containing additional information about the environment state """self.battery.reset()# Reset the battery with a random current_load between 0 and 25% max capacityself.energy_added_removed=[]self.dcload=self.dcload_minself.raw_obs=self._hist_data_collector()self.temp_state=self._process_obs(self.raw_obs)returnself.temp_state,{'action':-1,'avg_dc_power_mw':self.raw_obs[0],'Grid_CI':0,'total_energy_with_battery':0,'CO2_footprint':0,'avg_CI':0,'battery SOC':self.battery.current_load,'total_energy_with_battery':0}
[docs]defstep(self,action_id):""" Step function Args: action_id (int): the action id Returns: obs (list): Current state of the environmment reward (float): reward value. done (bool): A boolean value signaling the if the episode has ended. info (dict): A dictionary that containing additional information about the environment state """action_instantaneous=self._action_to_direction[action_id]self.discharge_energy=self._simulate_battery_operation(self.battery,action_instantaneous,charging_rate=self.charging_rate)self.CO2_total=self.CO2_footprint(self.dcload,self.ci,action_instantaneous,self.discharge_energy)self.raw_obs=self._hist_data_collector()self.temp_state=self._process_obs(self.raw_obs)self.reward=0self.info={'bat_action':action_id,'bat_SOC':self.battery.current_load,'bat_CO2_footprint':self.CO2_total,'bat_avg_CI':self.ci,'bat_total_energy_without_battery_KWh':self.dcload*1e3*0.25,'bat_total_energy_with_battery_KWh':self.total_energy_with_battery,'bat_max_bat_cap':self.max_bat_cap,'bat_a_t':action_instantaneous,'bat_dcload_min':self.dcload_min,'bat_dcload_max':self.dcload_max,}#Done and truncated are managed by the main class, implement individual function if neededtruncated=Falsedone=Falsereturnself.temp_state,self.reward,done,truncated,self.info
def_process_obs(self,state):"""Normalizes observations Args: state (List[float]): Current environment state. Returns: normalized_observations (List[float]) """scaled_value=(state-self.observation_min)/self.deltareturnnp.float32(scaled_value)def_process_action(self,action_id):"""Maps agent actions to actoniable action for the model Args: action_id (int): Action to take. Returns: normalized_observations (string) """returnself._action_to_direction[action_id]
[docs]defupdate_state(self):"""Updates obsevation with current DC energy consumption Returns: normalized_observations (string) """self.temp_state[0]=self.dcloadreturnself.temp_state
[docs]defset_dcload(self,dc_load):"""Set the current DC energy consumption Args: dc_load float: DC energy consumption. """self.dcload=dc_load
def_hist_data_collector(self):"""Generates the observation for the agent Returns: raw_obs (List[Float]): Current state observation """raw_obs=np.array([self.dcload,self.battery.current_load])returnraw_obsdef_simulate_battery_operation(self,battery,battery_action,charging_rate=None):"""Simulates battery operation Args: battery (Class): Battery model. battery_action (string): Desired action. charging_rate (string): Battery charging rate. Returns: discharge_energy (float): Output energy. """discharge_energy=0ifbattery_action=='charge':self.var_to_dc=battery.charge(battery.capacity,self.charging_rate_modifier(battery)*15/60)elifbattery_action=='discharge':discharge_energy=battery.discharge(battery.capacity,self.discharging_rate_modifier(battery)*15/60,self.dcload/4)self.var_to_dc=-discharge_energyelse:discharge_energy=0self.var_to_dc=0self.var_to_dc=self.var_to_dc/self.max_bat_capreturndischarge_energy
[docs]defCO2_footprint(self,dc_load,ci,a_t,discharge_energy):"""Calculates carbon footprint Args: dc_load (float): Total energy consumption of the DC. ci (float): Carbon intensity at current time step. a_t (string): Agent's action. discharge_energy (float): Amount of energy to be discharged Returns: CO2_footprint (float): Carbon footprint produced at the current time step (gCO2e) """ifa_t=='charge':self.total_energy_with_battery=dc_load*1e3*0.25+self.battery.charging_load*1e3self.energy_added_removed.append(self.battery.charging_load*1e3)self.battery.charging_load=0# *Added*CO2_footprint=(self.total_energy_with_battery)*cielifa_t=='discharge':assertdc_load*1e3*0.25>=discharge_energy*1e3,"Battery discharge rate should not be higher than the datacenter energy consumption rate"self.total_energy_with_battery=dc_load*1e3*0.25-discharge_energy*1e3self.energy_added_removed.append(-1.0*discharge_energy*1e3)CO2_footprint=max(self.total_energy_with_battery,0)*ci# (KWh) * gCO2/KWhelse:self.total_energy_with_battery=dc_load*1e3*0.25CO2_footprint=self.total_energy_with_battery*cireturnCO2_footprint
[docs]defcharging_rate_modifier(self,battery):"""Calculates the battery state depending on the charging rate Args: battery (batt.Battery2): Battery model Returns: charging_rate (float): Battery charging rate """curr_load=battery.current_loadbat_max,bat_min=battery.capacity,0scaled_curr_load=(curr_load-bat_min)/(bat_max-bat_min)# Scale current load to [0, 1]# Use a sigmoid function to model the charging ratecharging_rate=0.5*(1-self.sigmoid(10*(scaled_curr_load-0.5)))# Shift and scale sigmoid to model charging ratereturnnp.round(charging_rate,4)
[docs]defdischarging_rate_modifier(self,battery):"""Calculates the battery state depending on the discharging rate Args: battery (batt.Battery2): Battery model Returns: discharging_rate (float): Battery discharging rate """curr_load=battery.current_loadbat_max,bat_min=battery.capacity,0scaled_curr_load=(curr_load-bat_min)/(bat_max-bat_min)# Scale current load to [0, 1]# Use a sigmoid function to model the discharging ratedischarging_rate=4*self.sigmoid(10*(scaled_curr_load-0.25))# Shift and scale sigmoid to model discharging ratereturnmax(0.5,discharging_rate)