utils package

Submodules

utils.base_agents module

class utils.base_agents.BaseBatteryAgent(parameters=None)

Bases: object

Base class for battery agents.

Parameters:

parameters (dict) – Dictionary containing the agent parameters.

act(*args, **kwargs)

Return the do nothing action regardless of the input parameters.

Parameters:

• *args – Arbitrary positional arguments.

• **kwargs – Arbitrary keyword arguments.

Returns:

The action (do nothing) to be taken.

Return type:

action (int)

do_nothing_action()

Return the do nothing action.

Returns:

The action (do nothing) to be taken.

Return type:

action (int)

class utils.base_agents.BaseHVACAgent(parameters=None)

Bases: object

Base class for HVAC agents.

Parameters:

parameters (dict) – Dictionary containing the agent parameters.

do_nothing_action()

Return the do nothing action.

Returns:

The action (do nothing) to be taken.

Return type:

action (int)

class utils.base_agents.BaseLoadShiftingAgent(parameters=None)

Bases: object

Base class for load shifting agents.

Parameters:

parameters (dict) – Dictionary containing the agent parameters.

do_nothing_action()

Return the do nothing action.

Returns:

The action (do nothing) to be taken.

Return type:

action (int)

utils.checkpoint_finder module

utils.create_trainable module

utils.dc_config module

This file contains the data center configurations that may be customized by the user for designing the data center. The use of this file has been deprecated. Any changes to this file will not be reflected in the actual data center design. Instead, modify utils/dc_config.json to design the data center.

utils.dc_config_reader module

This file is used to read the data center configuration from user inputs provided inside dc_config.json. It also performs some auxiliary steps to calculate the server power specifications based on the given parameters.

class utils.dc_config_reader.DC_Config(dc_config_file='dc_config.json', datacenter_capacity_mw=1)

Bases: object

utils.dcrl_wrapper module

utils.helper_methods module

utils.hierarchical_workload_optimizer module

class utils.hierarchical_workload_optimizer.WorkloadOptimizer(dc_ids)

Bases: object

compute_actions(obs: dict) → dict

compute_adjusted_workload(dc_states)

Computes the optimal adjustments to the workload of each data center to minimize carbon emissions.

Parameters:

dc_states – A dictionary with data center states, where each state includes: - capacity (MWh) - current workload - external temperature - energy carbon intensity

Returns:

A dictionary with the recommended workload adjustments for each data center.

utils.low_level_wrapper module

utils.make_envs_pyenv module

utils.make_envs_pyenv.make_bat_fwd_env(month, max_bat_cap_Mwh: float = 2.0, charging_rate: float = 0.5, max_dc_pw_MW: float = 7.23, dcload_max: float = 2.5, dcload_min: float = 0.1, n_fwd_steps: int = 4)

Method to build the Battery environment.

Parameters:

• month (int) – Month of the year in which the agent is training.

• max_bat_cap_Mwh (float, optional) – Max battery capacity. Defaults to 2.0.

• charging_rate (float, optional) – Charging rate of the battery. Defaults to 0.5.

• reward_method (str, optional) – Method used to calculate the rewards. Defaults to ‘default_bat_reward’.

Returns:

Batery environment.

Return type:

battery_env_fwd

utils.make_envs_pyenv.make_dc_pyeplus_env(month: int = 1, location: str = 'NYIS', dc_config_file: str = 'dc_config_file.json', datacenter_capacity_mw: int = 1, max_bat_cap_Mw: float = 2.0, add_cpu_usage: bool = True, add_CI: bool = True, episode_length_in_time: Timedelta | None = None, use_ls_cpu_load: bool = False, num_sin_cos_vars: int = 4)

Method that creates the data center environment with the timeline, location, proper data files, gym specifications and auxiliary methods

Parameters:

• month (int, optional) – The month of the year for which the Environment uses the weather and Carbon Intensity data. Defaults to 1.

• location (str, optional) – The geographical location in a standard format for which Carbon Intensity files are accessed. Supported options are ‘NYIS’, ‘AZPS’, ‘BPAT’. Defaults to ‘NYIS’.

• datacenter_capacity_mw (int, optional) – Maximum capacity (MW) of the data center. This value will scale the number of servers installed in the data center.

• max_bat_cap_Mw (float, optional) – The battery capacity in Megawatts for the installed battery. Defaults to 2.0.

• add_cpu_usage (bool, optional) – Boolean Flag to indicate whether cpu usage is part of the environment statespace. Defaults to True.

• add_CI (bool, optional) – Boolean Flag to indicate whether Carbon Intensity is part of the environment statespace. Defaults to True.

• episode_length_in_time (pd.Timedelta, optional) – Length of an episode in terms of pandas time-delta object. Defaults to None.

• use_ls_cpu_load (bool, optional) – Use the cpu workload value from a separate Load Shifting agent. This turns of reading default cpu data. Defaults to False.

• num_sin_cos_vars (int, optional) – Number of sin and cosine variable that will be added externally from the centralized data source

Returns:

The environment instantiated with the particular month.

Return type:

envs.dc_gym.dc_gymenv

utils.make_envs_pyenv.make_ls_env(month, n_vars_ci: int = 4, n_vars_energy: int = 4, n_vars_battery: int = 1, queue_max_len: int = 500, test_mode=False)

Method to build the Load shifting environment

Parameters:

• month (int) – Month of the year in which the agent is training.

• n_vars_energy (int, optional) – Number of variables from the Energy environment. Defaults to 4.

• n_vars_battery (int, optional) – Number of variables from the Battery environment. Defaults to 1.

• queue_max_len (int, optional) – The size of the queue where the tasks are stored to be processed latter. Default to 500.

Returns:

Load Shifting environment

Return type:

CarbonLoadEnv

utils.managers module

class utils.managers.CI_Manager(filename='', location='NYIS', init_day=0, future_steps=4, weight=0.1, desired_std_dev=5, timezone_shift=0)

Bases: object

Manager of the carbon intensity data

Parameters:

• filename (str, optional) – Filename of the CPU data. Defaults to ‘ ‘.

• location (str, optional) – Location identifier. Defaults to ‘NYIS’.

• init_day (int, optional) – Initial day of the episode. Defaults to 0.

• future_steps (int, optional) – Number of steps of the CI forecast. Defaults to 4.

• weight (float, optional) – Weight value for coherent noise. Defaults to 0.001.

• desired_std_dev (float, optional) – Desired standard deviation for coherent noise. Defaults to 0.025.

get_current_ci()

get_forecast_ci(steps=4)

get_total_ci()

Function to obtain the total carbon intensity

Returns:

Total carbon intesity

Return type:

List[float]

reset()

Reset CI_Manager

Returns:

Carbon intensity at current time step float: Normalized carbon intensity at current time step and it’s forecast

Return type:

float

step()

Step CI_Manager

Returns:

Carbon intensity at current time step float: Normalized carbon intensity at current time step and it’s forecast

Return type:

float

class utils.managers.CoherentNoise(base, weight, desired_std_dev=0.1, scale=1)

Bases: object

Class to add coherent noise to the data.

Parameters:

• base (List[float]) – Base data

• weight (float) – Weight of the noise to be added

• desired_std_dev (float, optional) – Desired standard deviation. Defaults to 0.1.

• scale (int, optional) – Scale. Defaults to 1.

generate(n_steps)

Generate coherent noise

Parameters:

n_steps (int) – Length of the data to generate.

Returns:

Array of generated coherent noise.

Return type:

numpy.ndarray

class utils.managers.GeoLag_Workload_Manager(workload_filename='', init_day=0, future_steps=4, weight=0.01, desired_std_dev=0.025, timezone_shift=0, sustained_duration: int = 4)

Bases: Workload_Manager

get_current_workload()

set_current_workload(workload)

step()

Step function for the Workload_Manager

Returns:

CPU workload at current time step

Return type:

float

class utils.managers.Time_Manager(init_day=0, days_per_episode=30, timezone_shift=0)

Bases: object

Class to manage the time dimenssion over an episode

Parameters:

• init_day (int, optional) – Day to start from. Defaults to 0.

• days_per_episode (int, optional) – Number of days that an episode would last. Defaults to 30.

isterminal()

Function to identify terminal state

Returns:

Signals if a state is terminal or not

Return type:

bool

reset()

Reset time manager to initial day

Returns:

Hour and day in sine and cosine form

Return type:

List[float]

step()

Step function for the time maneger

Returns:

Current hour and day in sine and cosine form. bool: Signal if the episode has reach the end.

Return type:

List[float]

class utils.managers.Weather_Manager(filename='', location='NY', init_day=0, weight=0.02, desired_std_dev=0.75, temp_column=6, rh_column=8, pres_column=9, timezone_shift=0)

Bases: object

Manager of the weather data. Where to obtain other weather files: https://climate.onebuilding.org/

Args:

filename (str, optional): Filename of the weather data. Defaults to ‘’. location (str, optional): Location identifier. Defaults to ‘NY’. init_day (int, optional): Initial day of the year. Defaults to 0. weight (float, optional): Weight value for coherent noise. Defaults to 0.001. desired_std_dev (float, optional): Desired standard deviation for coherent noise. Defaults to 0.025. temp_column (int, optional): Columng that contains the temperature data. Defaults to 6.

get_current_weather()

get_total_weather()

Obtain the weather data in a List form

Returns:

Total temperature data

Return type:

List[form]

reset()

Reset Weather_Manager

Returns:

Temperature a current step float: Normalized temperature a current step

Return type:

float

step()

Step on the Weather_Manager

Returns:

Temperature a current step float: Normalized temperature a current step

Return type:

float

class utils.managers.Workload_Manager(workload_filename='', init_day=0, future_steps=4, weight=0.01, desired_std_dev=0.025, timezone_shift=0)

Bases: object

get_current_workload()

get_n_step_future_workload(n)

get_total_wkl()

Get current workload

Returns:

CPU data

Return type:

List[float]

reset()

Reset Workload_Manager

Returns:

CPU workload at current time step float: Amount of daily flexible workload

Return type:

float

scale_array(arr)

Scales the input array so that approximately 90% of its values fall within the range of 0.2 to 0.8, based on the 5th and 95th percentiles.

Parameters: arr (np.array): The input numpy array to be scaled.

Returns: np.array: The scaled numpy array.

set_current_workload(workload)

set_n_step_future_workload(n, workload)

step()

Step function for the Workload_Manager

Returns:

CPU workload at current time step float: Amount of daily flexible workload

Return type:

float

utils.managers.normalize(v, min_v, max_v)

Function to normalize values

Parameters:

• v (float) – Value to be normalized

• min_v (float) – Lower limit

• max_v (float) – Upper limit

Returns:

Normalized value

Return type:

float

utils.managers.sc_obs(current_hour, current_day)

Generate sine and cosine of the hour and day

Parameters:

• current_hour (int) – Current hour of the day

• current_day (int) – Current day of the year

Returns:

Sine and cosine of the hour and day

Return type:

List[float]

utils.managers.standarize(v)

Function to standarize a list of values

Parameters:

v (float) – Values to be normalized

Returns:

Normalized values

Return type:

float

utils.rbc_agents module

class utils.rbc_agents.RBCBatteryAgent(look_ahead=3, smooth_window=1, max_soc=0.9, min_soc=0.2)

Bases: object

act(carbon_intensity_values, current_soc)

Determine the action for the battery based on the carbon intensity forecast.

Parameters:

• carbon_intensity_values (list) – Forecasted carbon intensity values.

• current_soc (float) – Current state of charge of the battery.

Returns:

Action to be taken (0: ‘charge’, 1: ‘discharge’, 2: ‘idle’).

Return type:

int

utils.reward_creator module

utils.reward_creator.custom_agent_reward(params: dict) → float

A template for creating a custom agent reward function.

Parameters:

params (dict) – Dictionary containing custom parameters for reward calculation.

Returns:

Custom reward value. Currently returns 0.0 as a placeholder.

Return type:

float

utils.reward_creator.default_bat_reward(params: dict) → float

Calculates a reward value based on the battery usage.

Parameters:

params (dict) – Dictionary containing parameters: total_energy_with_battery (float): Total energy with battery. norm_CI (float): Normalized Carbon Intensity. dcload_min (float): Minimum DC load. dcload_max (float): Maximum DC load.

Returns:

Reward value.

Return type:

float

utils.reward_creator.default_dc_reward(params: dict) → float

Calculates a reward value based on the data center’s total ITE Load and CT Cooling load.

Parameters:

params (dict) – Dictionary containing parameters: data_center_total_ITE_Load (float): Total ITE Load of the data center. CT_Cooling_load (float): CT Cooling load of the data center. energy_lb (float): Lower bound of the energy. energy_ub (float): Upper bound of the energy.

Returns:

Reward value.

Return type:

float

utils.reward_creator.default_ls_reward(params: dict) → float

Calculates a reward value based on normalized load shifting.

Parameters:

params (dict) – Dictionary containing parameters: norm_load_left (float): Normalized load left. out_of_time (bool): Indicator (alarm) whether the agent is in the last hour of the day. penalty (float): Penalty value.

Returns:

Reward value.

Return type:

float

utils.reward_creator.energy_PUE_reward(params: dict) → float

Calculates a reward value based on Power Usage Effectiveness (PUE).

Parameters:

params (dict) – Dictionary containing parameters: total_energy_consumption (float): Total energy consumption of the data center. it_equipment_energy (float): Energy consumed by the IT equipment.

Returns:

Reward value.

Return type:

float

utils.reward_creator.energy_efficiency_reward(params: dict) → float

Calculates a reward value based on energy efficiency.

Parameters:

params (dict) – Dictionary containing parameters: ITE_load (float): The amount of energy spent on computation (useful work). total_energy_consumption (float): Total energy consumption of the data center.

Returns:

Reward value.

Return type:

float

utils.reward_creator.get_reward_method(reward_method: str = 'default_dc_reward')

Maps the string identifier to the reward function

Parameters:

reward_method (string) – Identifier for the reward function.

Returns:

Reward function.

Return type:

function

utils.reward_creator.renewable_energy_reward(params: dict) → float

Calculates a reward value based on the usage of renewable energy sources.

Parameters:

params (dict) – Dictionary containing parameters: renewable_energy_ratio (float): Ratio of energy coming from renewable sources. total_energy_consumption (float): Total energy consumption of the data center.

Returns:

Reward value.

Return type:

float

utils.reward_creator.temperature_efficiency_reward(params: dict) → float

Calculates a reward value based on the efficiency of cooling in the data center.

Parameters:

params (dict) – Dictionary containing parameters: current_temperature (float): Current temperature in the data center. optimal_temperature_range (tuple): Tuple containing the minimum and maximum optimal temperatures for the data center.

Returns:

Reward value.

Return type:

float

utils.reward_creator.tou_reward(params: dict) → float

Calculates a reward value based on the Time of Use (ToU) of energy.

Parameters:

params (dict) – Dictionary containing parameters: energy_usage (float): The energy usage of the agent. hour (int): The current hour of the day (24-hour format).

Returns:

Reward value.

Return type:

float

utils.reward_creator.water_usage_efficiency_reward(params: dict) → float

Calculates a reward value based on the efficiency of water usage in the data center.

A lower value of water usage results in a higher reward, promoting sustainability and efficiency in water consumption.

Parameters:

params (dict) – Dictionary containing parameters: dc_water_usage (float): The amount of water used by the data center in a given period.

Returns:

Reward value. The reward is higher for lower values of water usage, promoting reduced water consumption.

Return type:

float

utils.rllib_callbacks module

utils.trim_and_respond module

https://mepacademy.com/top-6-hvac-control-strategies-to-save-energy/#:~:text=Using%20Trim%20and%20Respond%20control,setpoint%20to%2055.2%C2%B0F. Based on #6 Supply Air Temperature Reset

class utils.trim_and_respond.trim_and_respond_ctrl(TandR_monitor_idx=6, TandR_monitor: str = 'avg_room_temp', TandR_monitor_limit: float = 27)

Bases: object

action(obs)

set_limit(x)

utils.utils_cf module

utils.utils_cf.get_energy_variables(state)

Obtain energy variables from the energy observation

Parameters:

state (List[float]) – agent_dc observation

Returns:

Subset of the agent_dc observation

Return type:

List[float]

utils.utils_cf.get_init_day(start_month=0)

Obtain the initial day of the year to start the episode on

Parameters:

start_month (int, optional) – Starting month. Defaults to 0.

Returns:

Day of the year corresponding to the first day of the month

Return type:

int

utils.utils_cf.obtain_paths(location)

Obtain the correct name for the data files

Parameters:

location (string) – Location identifier

Raises:

ValueError – If location identifier is not defined

Returns:

Naming for the data files

Return type:

List[string]

Module contents

Includes common utilities, controllers, rewards, wrappers and custom callbacks.